Deployment CRUD Events

Goal

Emit events for deployment create, update, and delete operations in OSS, enabling users to build automations that respond to deployment lifecycle changes.

Issue: https://github.com/PrefectHQ/prefect/issues/20176

Background

Current State

OSS currently emits:

prefect.deployment.{status} - status change events (ready/not-ready)
prefect.work-pool.updated - field change events for work pools
prefect.work-queue.updated - field change events for work queues

Missing:

prefect.deployment.created
prefect.deployment.updated
prefect.deployment.deleted

Cloud Parity

Cloud emits these events via the Auditor system, which adds actor-related resources for audit trail. The event names use the prefect. prefix (not prefect-cloud.) because deployments are "orchestration objects":

python

# Cloud uses prefect.* prefix for orchestration objects
orchestration_object_types: set[str] = {
    "deployment",  # Uses prefect.deployment.* naming
    "flow",
    "work-pool",
    ...
}

Consistency: OSS events will be identical in shape to Cloud, just without actor-related resources. Automations using these events will be portable between OSS and Cloud.

Event Specifications

prefect.deployment.created

Emitted when a new deployment is created (not on upsert/update of existing).

python

Event(
    event="prefect.deployment.created",
    resource={
        "prefect.resource.id": f"prefect.deployment.{deployment.id}",
        "prefect.resource.name": deployment.name,
    },
    related=[
        {
            "prefect.resource.id": f"prefect.flow.{flow.id}",
            "prefect.resource.name": flow.name,
            "prefect.resource.role": "flow",
        },
        # work-queue if present
        # work-pool if present
    ],
    payload={...},  # deployment data
)

prefect.deployment.updated

Emitted when an existing deployment is modified. Includes changed fields in payload.

python

Event(
    event="prefect.deployment.updated",
    resource={
        "prefect.resource.id": f"prefect.deployment.{deployment.id}",
        "prefect.resource.name": deployment.name,
    },
    related=[...],  # flow, work-queue, work-pool
    payload={
        "updated_fields": ["description", "parameters"],
        "updates": {
            "description": {"from": "...", "to": "..."},
            "parameters": {"from": {...}, "to": {...}},
        },
    },
)

prefect.deployment.deleted

Emitted when a deployment is deleted.

python

Event(
    event="prefect.deployment.deleted",
    resource={
        "prefect.resource.id": f"prefect.deployment.{deployment.id}",
        "prefect.resource.name": deployment.name,
    },
    related=[...],  # flow, work-queue, work-pool (captured before deletion)
)

Implementation

Phase 1: Event Factory Functions

Location: src/prefect/server/models/events.py

Add three new functions following the existing work_pool_updated_event pattern:

python

async def deployment_created_event(
    session: AsyncSession,
    deployment: ORMDeployment,
    occurred: DateTime,
) -> Event:
    """Create an event for deployment creation."""
    ...

async def deployment_updated_event(
    session: AsyncSession,
    deployment: ORMDeployment,
    changed_fields: Dict[str, Dict[str, Any]],
    occurred: DateTime,
) -> Event:
    """Create an event for deployment field updates."""
    ...

async def deployment_deleted_event(
    session: AsyncSession,
    deployment: ORMDeployment,
    occurred: DateTime,
) -> Event:
    """Create an event for deployment deletion."""
    ...

Helper function (similar to _flow_run_related_resources_from_orm):

python

async def _deployment_related_resources(
    session: AsyncSession,
    deployment: ORMDeployment,
) -> RelatedResourceList:
    """Get related resources for a deployment event."""
    # Returns flow, work-queue, work-pool as related resources
    ...

Status:

deployment_created_event function
deployment_updated_event function
deployment_deleted_event function
_deployment_related_resources helper
Unit tests for event factory functions (tests/server/models/test_events.py)

Phase 2: Emit Functions

Location: src/prefect/server/models/deployments.py

Add emit wrapper functions (following emit_work_queue_updated_event pattern):

python

async def emit_deployment_created_event(
    session: AsyncSession,
    deployment: ORMDeployment,
) -> None:
    """Emit an event when a deployment is created."""
    async with PrefectServerEventsClient() as events_client:
        await events_client.emit(
            await deployment_created_event(
                session=session,
                deployment=deployment,
                occurred=now("UTC"),
            )
        )

async def emit_deployment_updated_event(
    session: AsyncSession,
    deployment: ORMDeployment,
    changed_fields: Dict[str, Dict[str, Any]],
) -> None:
    """Emit an event when a deployment is updated."""
    if not changed_fields:
        return
    async with PrefectServerEventsClient() as events_client:
        await events_client.emit(
            await deployment_updated_event(
                session=session,
                deployment=deployment,
                changed_fields=changed_fields,
                occurred=now("UTC"),
            )
        )

async def emit_deployment_deleted_event(
    session: AsyncSession,
    deployment: ORMDeployment,
) -> None:
    """Emit an event when a deployment is deleted."""
    async with PrefectServerEventsClient() as events_client:
        await events_client.emit(
            await deployment_deleted_event(
                session=session,
                deployment=deployment,
                occurred=now("UTC"),
            )
        )

Status:

emit_deployment_created_event function
emit_deployment_updated_event function
emit_deployment_deleted_event function
Unit tests for emit functions

Phase 3: Integration with Model Layer

Emit from src/prefect/server/models/deployments.py, following the existing work-pool/work-queue pattern.

Create/Upsert

In create_deployment model function:

Track whether deployment is new (created >= invocation_time) or existing
After successful operation, call emit_deployment_created_event or emit_deployment_updated_event
For updates, compute changed_fields by comparing old vs new values

Update

In update_deployment model function:

Capture deployment state before update
After update, compute changed fields
Call emit_deployment_updated_event with changed fields

Delete

In delete_deployment model function:

Read deployment before deletion (to capture data for event)
After successful deletion, call emit_deployment_deleted_event

Status:

Integrate created event into create/upsert flow
Integrate updated event into update flow
Integrate deleted event into delete flow
Handle bulk delete if applicable
Integration tests (tests/server/api/test_deployments.py)
- creating a deployment emits prefect.deployment.created
- upserting (updating existing) emits prefect.deployment.updated
- updating via PATCH emits prefect.deployment.updated
- deleting emits prefect.deployment.deleted
- events include correct related resources

Edge Cases

Scenario	Behavior
Upsert creates new deployment	Emit `created` event
Upsert updates existing deployment	Emit `updated` event
Update with no actual changes	No event emitted (changed_fields empty)
Delete non-existent deployment	No event (404 returned)
Deployment without work pool	Events still emitted, work-pool not in related
Bulk delete	Emit `deleted` event for each deployment

Verification Checklist

Automated

Unit tests pass
Integration tests pass
Type checker passes

Manual

Create deployment → verify created event in event stream
Update deployment → verify updated event with changed fields
Delete deployment → verify deleted event
Create automation triggered by prefect.deployment.created
Verify event structure matches Cloud (minus actor resources)

Out of Scope

Actor tracking (Cloud-only feature)
Audit log UI/branding (Cloud-only feature)
Events for other resources (flows, flow runs, etc.)