Skip to content

prefect.server.api.deployments

Routes for interacting with Deployment objects.

count_deployments async

Count deployments.

Source code in prefect/server/api/deployments.py
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
@router.post("/count")
async def count_deployments(
    flows: schemas.filters.FlowFilter = None,
    flow_runs: schemas.filters.FlowRunFilter = None,
    task_runs: schemas.filters.TaskRunFilter = None,
    deployments: schemas.filters.DeploymentFilter = None,
    work_pools: schemas.filters.WorkPoolFilter = None,
    work_pool_queues: schemas.filters.WorkQueueFilter = None,
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> int:
    """
    Count deployments.
    """
    async with db.session_context() as session:
        return await models.deployments.count_deployments(
            session=session,
            flow_filter=flows,
            flow_run_filter=flow_runs,
            task_run_filter=task_runs,
            deployment_filter=deployments,
            work_pool_filter=work_pools,
            work_queue_filter=work_pool_queues,
        )

create_deployment async

Gracefully creates a new deployment from the provided schema. If a deployment with the same name and flow_id already exists, the deployment is updated.

If the deployment has an active schedule, flow runs will be scheduled. When upserting, any scheduled runs from the existing deployment will be deleted.

Source code in prefect/server/api/deployments.py
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@router.post("/")
async def create_deployment(
    deployment: schemas.actions.DeploymentCreate,
    response: Response,
    worker_lookups: WorkerLookups = Depends(WorkerLookups),
    created_by: Optional[schemas.core.CreatedBy] = Depends(dependencies.get_created_by),
    updated_by: Optional[schemas.core.UpdatedBy] = Depends(dependencies.get_updated_by),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.responses.DeploymentResponse:
    """
    Gracefully creates a new deployment from the provided schema. If a deployment with
    the same name and flow_id already exists, the deployment is updated.

    If the deployment has an active schedule, flow runs will be scheduled.
    When upserting, any scheduled runs from the existing deployment will be deleted.
    """

    data = deployment.dict(exclude_unset=True)
    data["created_by"] = created_by.dict() if created_by else None
    data["updated_by"] = updated_by.dict() if created_by else None

    async with db.session_context(begin_transaction=True) as session:
        if (
            deployment.work_pool_name
            and deployment.work_pool_name != DEFAULT_AGENT_WORK_POOL_NAME
        ):
            # Make sure that deployment is valid before beginning creation process
            work_pool = await models.workers.read_work_pool_by_name(
                session=session, work_pool_name=deployment.work_pool_name
            )
            if work_pool is None:
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f'Work pool "{deployment.work_pool_name}" not found.',
                )

            await validate_job_variables_for_deployment(
                session,
                work_pool,
                deployment,
            )

        # hydrate the input model into a full model
        deployment_dict = deployment.dict(exclude={"work_pool_name"})
        if deployment.work_pool_name and deployment.work_queue_name:
            # If a specific pool name/queue name combination was provided, get the
            # ID for that work pool queue.
            deployment_dict[
                "work_queue_id"
            ] = await worker_lookups._get_work_queue_id_from_name(
                session=session,
                work_pool_name=deployment.work_pool_name,
                work_queue_name=deployment.work_queue_name,
                create_queue_if_not_found=True,
            )
        elif deployment.work_pool_name:
            # If just a pool name was provided, get the ID for its default
            # work pool queue.
            deployment_dict[
                "work_queue_id"
            ] = await worker_lookups._get_default_work_queue_id_from_work_pool_name(
                session=session,
                work_pool_name=deployment.work_pool_name,
            )
        elif deployment.work_queue_name:
            # If just a queue name was provided, ensure that the queue exists and
            # get its ID.
            work_queue = await models.work_queues.ensure_work_queue_exists(
                session=session, name=deployment.work_queue_name
            )
            deployment_dict["work_queue_id"] = work_queue.id

        deployment = schemas.core.Deployment(**deployment_dict)
        # check to see if relevant blocks exist, allowing us throw a useful error message
        # for debugging
        if deployment.infrastructure_document_id is not None:
            infrastructure_block = (
                await models.block_documents.read_block_document_by_id(
                    session=session,
                    block_document_id=deployment.infrastructure_document_id,
                )
            )
            if not infrastructure_block:
                raise HTTPException(
                    status_code=status.HTTP_409_CONFLICT,
                    detail=(
                        "Error creating deployment. Could not find infrastructure"
                        f" block with id: {deployment.infrastructure_document_id}. This"
                        " usually occurs when applying a deployment specification that"
                        " was built against a different Prefect database / workspace."
                    ),
                )

        if deployment.storage_document_id is not None:
            storage_block = await models.block_documents.read_block_document_by_id(
                session=session,
                block_document_id=deployment.storage_document_id,
            )
            if not storage_block:
                raise HTTPException(
                    status_code=status.HTTP_409_CONFLICT,
                    detail=(
                        "Error creating deployment. Could not find storage block with"
                        f" id: {deployment.storage_document_id}. This usually occurs"
                        " when applying a deployment specification that was built"
                        " against a different Prefect database / workspace."
                    ),
                )

        # Ensure that `paused` and `is_schedule_active` are consistent.
        if "paused" in data:
            deployment.is_schedule_active = not data["paused"]
        elif "is_schedule_active" in data:
            deployment.paused = not data["is_schedule_active"]

        now = pendulum.now("UTC")
        model = await models.deployments.create_deployment(
            session=session, deployment=deployment
        )

        if model.created >= now:
            response.status_code = status.HTTP_201_CREATED

        return schemas.responses.DeploymentResponse.from_orm(model)

create_flow_run_from_deployment async

Create a flow run from a deployment.

Any parameters not provided will be inferred from the deployment's parameters. If tags are not provided, the deployment's tags will be used.

If no state is provided, the flow run will be created in a SCHEDULED state.

Source code in prefect/server/api/deployments.py
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
@router.post("/{id}/create_flow_run")
async def create_flow_run_from_deployment(
    flow_run: schemas.actions.DeploymentFlowRunCreate,
    deployment_id: UUID = Path(..., description="The deployment id", alias="id"),
    created_by: Optional[schemas.core.CreatedBy] = Depends(dependencies.get_created_by),
    db: PrefectDBInterface = Depends(provide_database_interface),
    worker_lookups: WorkerLookups = Depends(WorkerLookups),
    response: Response = None,
) -> schemas.responses.FlowRunResponse:
    """
    Create a flow run from a deployment.

    Any parameters not provided will be inferred from the deployment's parameters.
    If tags are not provided, the deployment's tags will be used.

    If no state is provided, the flow run will be created in a SCHEDULED state.
    """
    async with db.session_context(begin_transaction=True) as session:
        # get relevant info from the deployment
        deployment = await models.deployments.read_deployment(
            session=session, deployment_id=deployment_id
        )

        if not deployment:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="Deployment not found"
            )

        try:
            dehydrated_params = deployment.parameters
            dehydrated_params.update(flow_run.parameters or {})
            ctx = await HydrationContext.build(
                session=session,
                raise_on_error=True,
                render_jinja=True,
                render_workspace_variables=True,
            )
            parameters = hydrate(dehydrated_params, ctx)
        except HydrationError as exc:
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST,
                detail=f"Error hydrating flow run parameters: {exc}",
            )

        # default
        enforce_parameter_schema = deployment.enforce_parameter_schema

        # run override
        if flow_run.enforce_parameter_schema is not None:
            enforce_parameter_schema = flow_run.enforce_parameter_schema

        if enforce_parameter_schema:
            if not isinstance(deployment.parameter_openapi_schema, dict):
                raise HTTPException(
                    status.HTTP_409_CONFLICT,
                    detail=(
                        "Error updating deployment: Cannot update parameters because"
                        " parameter schema enforcement is enabled and the deployment"
                        " does not have a valid parameter schema."
                    ),
                )
            try:
                validate(
                    parameters, deployment.parameter_openapi_schema, raise_on_error=True
                )
            except ValidationError as exc:
                raise HTTPException(
                    status.HTTP_409_CONFLICT,
                    detail=f"Error creating flow run: {exc}",
                )
            except CircularSchemaRefError:
                raise HTTPException(
                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail="Invalid schema: Unable to validate schema with circular references.",
                )

        await validate_job_variables_for_deployment_flow_run(
            session, deployment, flow_run
        )

        work_queue_name = deployment.work_queue_name
        work_queue_id = deployment.work_queue_id

        if flow_run.work_queue_name:
            # can't mutate the ORM model or else it will commit the changes back
            work_queue_id = await worker_lookups._get_work_queue_id_from_name(
                session=session,
                work_pool_name=deployment.work_queue.work_pool.name,
                work_queue_name=flow_run.work_queue_name,
                create_queue_if_not_found=True,
            )
            work_queue_name = flow_run.work_queue_name

        # hydrate the input model into a full flow run / state model
        flow_run = schemas.core.FlowRun(
            **flow_run.dict(
                exclude={
                    "parameters",
                    "tags",
                    "infrastructure_document_id",
                    "work_queue_name",
                    "enforce_parameter_schema",
                }
            ),
            flow_id=deployment.flow_id,
            deployment_id=deployment.id,
            deployment_version=deployment.version,
            parameters=parameters,
            tags=set(deployment.tags).union(flow_run.tags),
            infrastructure_document_id=(
                flow_run.infrastructure_document_id
                or deployment.infrastructure_document_id
            ),
            work_queue_name=work_queue_name,
            work_queue_id=work_queue_id,
            created_by=created_by,
        )

        if not flow_run.state:
            flow_run.state = schemas.states.Scheduled()

        now = pendulum.now("UTC")
        model = await models.flow_runs.create_flow_run(
            session=session, flow_run=flow_run
        )
        if model.created >= now:
            response.status_code = status.HTTP_201_CREATED
        return schemas.responses.FlowRunResponse.from_orm(model)

delete_deployment async

Delete a deployment by id.

Source code in prefect/server/api/deployments.py
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_deployment(
    deployment_id: UUID = Path(..., description="The deployment id", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
):
    """
    Delete a deployment by id.
    """
    async with db.session_context(begin_transaction=True) as session:
        result = await models.deployments.delete_deployment(
            session=session, deployment_id=deployment_id
        )
    if not result:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Deployment not found"
        )

get_scheduled_flow_runs_for_deployments async

Get scheduled runs for a set of deployments. Used by a runner to poll for work.

Source code in prefect/server/api/deployments.py
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
@router.post("/get_scheduled_flow_runs")
async def get_scheduled_flow_runs_for_deployments(
    background_tasks: BackgroundTasks,
    deployment_ids: List[UUID] = Body(
        default=..., description="The deployment IDs to get scheduled runs for"
    ),
    scheduled_before: DateTimeTZ = Body(
        None, description="The maximum time to look for scheduled flow runs"
    ),
    limit: int = dependencies.LimitBody(),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> List[schemas.responses.FlowRunResponse]:
    """
    Get scheduled runs for a set of deployments. Used by a runner to poll for work.
    """
    async with db.session_context() as session:
        orm_flow_runs = await models.flow_runs.read_flow_runs(
            session=session,
            limit=limit,
            deployment_filter=schemas.filters.DeploymentFilter(
                id=schemas.filters.DeploymentFilterId(any_=deployment_ids),
            ),
            flow_run_filter=schemas.filters.FlowRunFilter(
                next_scheduled_start_time=schemas.filters.FlowRunFilterNextScheduledStartTime(
                    before_=scheduled_before
                ),
                state=schemas.filters.FlowRunFilterState(
                    type=schemas.filters.FlowRunFilterStateType(
                        any_=[schemas.states.StateType.SCHEDULED]
                    )
                ),
            ),
            sort=schemas.sorting.FlowRunSort.NEXT_SCHEDULED_START_TIME_ASC,
        )

        flow_run_responses = [
            schemas.responses.FlowRunResponse.from_orm(orm_flow_run=orm_flow_run)
            for orm_flow_run in orm_flow_runs
        ]

    background_tasks.add_task(
        mark_deployments_ready,
        deployment_ids=deployment_ids,
    )

    return flow_run_responses

pause_deployment async

Set a deployment schedule to inactive. Any auto-scheduled runs still in a Scheduled state will be deleted.

Source code in prefect/server/api/deployments.py
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
@router.post("/{id:uuid}/set_schedule_inactive")  # Legacy route
@router.post("/{id:uuid}/pause_deployment")
async def pause_deployment(
    deployment_id: UUID = Path(..., description="The deployment id", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> None:
    """
    Set a deployment schedule to inactive. Any auto-scheduled runs still in a Scheduled
    state will be deleted.
    """
    async with db.session_context(begin_transaction=False) as session:
        deployment = await models.deployments.read_deployment(
            session=session, deployment_id=deployment_id
        )
        if not deployment:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="Deployment not found"
            )
        deployment.is_schedule_active = False
        deployment.paused = True

        # Ensure that we're updating the replicated schedule's `active` field,
        # if there is only a single schedule. This is support for legacy
        # clients.

        number_of_schedules = len(deployment.schedules)

        if number_of_schedules == 1:
            deployment.schedules[0].active = False
        elif number_of_schedules > 1:
            raise _multiple_schedules_error(deployment_id)

        # commit here to make the inactive schedule "visible" to the scheduler service
        await session.commit()

        # delete any auto scheduled runs
        await models.deployments._delete_scheduled_runs(
            session=session,
            deployment_id=deployment_id,
            auto_scheduled_only=True,
        )

        await session.commit()

read_deployment async

Get a deployment by id.

Source code in prefect/server/api/deployments.py
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
@router.get("/{id}")
async def read_deployment(
    deployment_id: UUID = Path(..., description="The deployment id", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.responses.DeploymentResponse:
    """
    Get a deployment by id.
    """
    async with db.session_context() as session:
        deployment = await models.deployments.read_deployment(
            session=session, deployment_id=deployment_id
        )
        if not deployment:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="Deployment not found"
            )
        return schemas.responses.DeploymentResponse.from_orm(deployment)

read_deployment_by_name async

Get a deployment using the name of the flow and the deployment.

Source code in prefect/server/api/deployments.py
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
@router.get("/name/{flow_name}/{deployment_name}")
async def read_deployment_by_name(
    flow_name: str = Path(..., description="The name of the flow"),
    deployment_name: str = Path(..., description="The name of the deployment"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> schemas.responses.DeploymentResponse:
    """
    Get a deployment using the name of the flow and the deployment.
    """
    async with db.session_context() as session:
        deployment = await models.deployments.read_deployment_by_name(
            session=session, name=deployment_name, flow_name=flow_name
        )
        if not deployment:
            raise HTTPException(
                status.HTTP_404_NOT_FOUND, detail="Deployment not found"
            )
        return schemas.responses.DeploymentResponse.from_orm(deployment)

read_deployments async

Query for deployments.

Source code in prefect/server/api/deployments.py
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
@router.post("/filter")
async def read_deployments(
    limit: int = dependencies.LimitBody(),
    offset: int = Body(0, ge=0),
    flows: schemas.filters.FlowFilter = None,
    flow_runs: schemas.filters.FlowRunFilter = None,
    task_runs: schemas.filters.TaskRunFilter = None,
    deployments: schemas.filters.DeploymentFilter = None,
    work_pools: schemas.filters.WorkPoolFilter = None,
    work_pool_queues: schemas.filters.WorkQueueFilter = None,
    sort: schemas.sorting.DeploymentSort = Body(
        schemas.sorting.DeploymentSort.NAME_ASC
    ),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> List[schemas.responses.DeploymentResponse]:
    """
    Query for deployments.
    """
    async with db.session_context() as session:
        response = await models.deployments.read_deployments(
            session=session,
            offset=offset,
            sort=sort,
            limit=limit,
            flow_filter=flows,
            flow_run_filter=flow_runs,
            task_run_filter=task_runs,
            deployment_filter=deployments,
            work_pool_filter=work_pools,
            work_queue_filter=work_pool_queues,
        )
        return [
            schemas.responses.DeploymentResponse.from_orm(orm_deployment=deployment)
            for deployment in response
        ]

resume_deployment async

Set a deployment schedule to active. Runs will be scheduled immediately.

Source code in prefect/server/api/deployments.py
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
@router.post("/{id:uuid}/set_schedule_active")  # Legacy route
@router.post("/{id:uuid}/resume_deployment")
async def resume_deployment(
    deployment_id: UUID = Path(..., description="The deployment id", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> None:
    """
    Set a deployment schedule to active. Runs will be scheduled immediately.
    """
    async with db.session_context(begin_transaction=True) as session:
        deployment = await models.deployments.read_deployment(
            session=session, deployment_id=deployment_id
        )
        if not deployment:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="Deployment not found"
            )
        deployment.is_schedule_active = True
        deployment.paused = False

        # Ensure that we're updating the replicated schedule's `active` field,
        # if there is only a single schedule. This is support for legacy
        # clients.

        number_of_schedules = len(deployment.schedules)

        if number_of_schedules == 1:
            deployment.schedules[0].active = True
        elif number_of_schedules > 1:
            raise _multiple_schedules_error(deployment_id)

schedule_deployment async

Schedule runs for a deployment. For backfills, provide start/end times in the past.

This function will generate the minimum number of runs that satisfy the min and max times, and the min and max counts. Specifically, the following order will be respected.

- Runs will be generated starting on or after the `start_time`
- No more than `max_runs` runs will be generated
- No runs will be generated after `end_time` is reached
- At least `min_runs` runs will be generated
- Runs will be generated until at least `start_time + min_time` is reached
Source code in prefect/server/api/deployments.py
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
@router.post("/{id}/schedule")
async def schedule_deployment(
    deployment_id: UUID = Path(..., description="The deployment id", alias="id"),
    start_time: DateTimeTZ = Body(None, description="The earliest date to schedule"),
    end_time: DateTimeTZ = Body(None, description="The latest date to schedule"),
    min_time: datetime.timedelta = Body(
        None,
        description=(
            "Runs will be scheduled until at least this long after the `start_time`"
        ),
    ),
    min_runs: int = Body(None, description="The minimum number of runs to schedule"),
    max_runs: int = Body(None, description="The maximum number of runs to schedule"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> None:
    """
    Schedule runs for a deployment. For backfills, provide start/end times in the past.

    This function will generate the minimum number of runs that satisfy the min
    and max times, and the min and max counts. Specifically, the following order
    will be respected.

        - Runs will be generated starting on or after the `start_time`
        - No more than `max_runs` runs will be generated
        - No runs will be generated after `end_time` is reached
        - At least `min_runs` runs will be generated
        - Runs will be generated until at least `start_time + min_time` is reached
    """
    async with db.session_context(begin_transaction=True) as session:
        await models.deployments.schedule_runs(
            session=session,
            deployment_id=deployment_id,
            start_time=start_time,
            min_time=min_time,
            end_time=end_time,
            min_runs=min_runs,
            max_runs=max_runs,
        )

work_queue_check_for_deployment async

Get list of work-queues that are able to pick up the specified deployment.

This endpoint is intended to be used by the UI to provide users warnings about deployments that are unable to be executed because there are no work queues that will pick up their runs, based on existing filter criteria. It may be deprecated in the future because there is not a strict relationship between work queues and deployments.

Source code in prefect/server/api/deployments.py
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
@router.get("/{id}/work_queue_check", deprecated=True)
async def work_queue_check_for_deployment(
    deployment_id: UUID = Path(..., description="The deployment id", alias="id"),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> List[schemas.core.WorkQueue]:
    """
    Get list of work-queues that are able to pick up the specified deployment.

    This endpoint is intended to be used by the UI to provide users warnings
    about deployments that are unable to be executed because there are no work
    queues that will pick up their runs, based on existing filter criteria. It
    may be deprecated in the future because there is not a strict relationship
    between work queues and deployments.
    """
    try:
        async with db.session_context() as session:
            work_queues = await models.deployments.check_work_queues_for_deployment(
                session=session, deployment_id=deployment_id
            )
    except ObjectNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Deployment not found"
        )
    return work_queues