workers

​

prefect.server.api.workers

Routes for interacting with work queue objects.

​

Functions

​

schedule_readiness_update

schedule_readiness_update(docket: Docket, function: Callable[P, Awaitable[None]], key: str) -> Callable[P, Awaitable[Execution]]

Schedule a keyed readiness update from a poll.

Polls are deduplicated by key so repeated polls don’t pile up tasks. When the poll would move something to READY, any pending task is replaced instead: a previous attempt can be left pending with its key still known (for example, when a worker’s claim fails during a Redis disruption) and would otherwise block every later poll until docket’s redelivery_timeout.

​

create_work_pool

create_work_pool(work_pool: schemas.actions.WorkPoolCreate, db: PrefectDBInterface = Depends(provide_database_interface), prefect_client_version: Optional[str] = Depends(dependencies.get_prefect_client_version)) -> schemas.responses.WorkPoolResponse

Creates a new work pool. If a work pool with the same name already exists, an error will be raised.

For more information, see https://www.prefect.io/docs/v3/concepts/work-pools.

​

read_work_pool

read_work_pool(work_pool_name: str = Path(..., description='The work pool name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface), prefect_client_version: Optional[str] = Depends(dependencies.get_prefect_client_version)) -> schemas.responses.WorkPoolResponse

Read a work pool by name

​

read_work_pools

read_work_pools(work_pools: Optional[schemas.filters.WorkPoolFilter] = None, limit: int = dependencies.LimitBody(), offset: int = Body(0, ge=0), db: PrefectDBInterface = Depends(provide_database_interface), prefect_client_version: Optional[str] = Depends(dependencies.get_prefect_client_version)) -> List[schemas.responses.WorkPoolResponse]

Read multiple work pools

​

count_work_pools

count_work_pools(work_pools: Optional[schemas.filters.WorkPoolFilter] = Body(None, embed=True), db: PrefectDBInterface = Depends(provide_database_interface)) -> int

Count work pools

​

update_work_pool

update_work_pool(work_pool: schemas.actions.WorkPoolUpdate, work_pool_name: str = Path(..., description='The work pool name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None

Update a work pool

​

delete_work_pool

delete_work_pool(work_pool_name: str = Path(..., description='The work pool name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None

Delete a work pool

​

read_work_pool_concurrency_status

read_work_pool_concurrency_status(work_pool_name: str = Path(..., description='The work pool name', alias='name'), page: int = Body(1, ge=1), limit: int = dependencies.LimitBody(), flow_run_limit: int = Body(10, ge=0, le=200, description='Max flow runs per queue'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> schemas.responses.WorkPoolConcurrencyStatus

Read concurrency status for a work pool, including per-queue breakdown with flow run summaries. Queues are paginated; flow runs per queue are capped by flow_run_limit.

​

get_scheduled_flow_runs

get_scheduled_flow_runs(docket: dependencies.Docket, work_pool_name: str = Path(..., description='The work pool name', alias='name'), work_queue_names: List[str] = Body(None, description='The names of work pool queues'), scheduled_before: DateTime = Body(None, description='The maximum time to look for scheduled flow runs'), scheduled_after: DateTime = Body(None, description='The minimum time to look for scheduled flow runs'), limit: int = dependencies.LimitBody(), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> List[schemas.responses.WorkerFlowRunResponse]

Load scheduled runs for a worker

​

create_work_queue

create_work_queue(work_queue: schemas.actions.WorkQueueCreate, work_pool_name: str = Path(..., description='The work pool name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> schemas.responses.WorkQueueResponse

Creates a new work pool queue. If a work pool queue with the same name already exists, an error will be raised.

For more information, see https://www.prefect.io/docs/v3/concepts/work-pools#work-queues.

​

read_work_queue

read_work_queue(work_pool_name: str = Path(..., description='The work pool name'), work_queue_name: str = Path(..., description='The work pool queue name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> schemas.responses.WorkQueueResponse

Read a work pool queue

​

read_work_queues

read_work_queues(work_pool_name: str = Path(..., description='The work pool name'), work_queues: schemas.filters.WorkQueueFilter = None, limit: int = dependencies.LimitBody(), offset: int = Body(0, ge=0), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> List[schemas.responses.WorkQueueResponse]

Read all work pool queues

​

update_work_queue

update_work_queue(work_queue: schemas.actions.WorkQueueUpdate, work_pool_name: str = Path(..., description='The work pool name'), work_queue_name: str = Path(..., description='The work pool queue name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None

Update a work pool queue

​

delete_work_queue

delete_work_queue(work_pool_name: str = Path(..., description='The work pool name'), work_queue_name: str = Path(..., description='The work pool queue name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None

Delete a work pool queue

​

worker_channel_connect

worker_channel_connect(websocket: WebSocket, work_pool_name: str = Path(..., description='The work pool name'), db: PrefectDBInterface = Depends(provide_database_interface)) -> None

​

worker_heartbeat

worker_heartbeat(work_pool_name: str = Path(..., description='The work pool name'), name: str = Body(..., description='The worker process name', embed=True), heartbeat_interval_seconds: Optional[int] = Body(None, description="The worker's heartbeat interval in seconds", embed=True), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None

​

read_workers

read_workers(work_pool_name: str = Path(..., description='The work pool name'), workers: Optional[schemas.filters.WorkerFilter] = None, limit: int = dependencies.LimitBody(), offset: int = Body(0, ge=0), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> List[schemas.responses.WorkerResponse]

Read all worker processes

​

delete_worker

delete_worker(work_pool_name: str = Path(..., description='The work pool name'), worker_name: str = Path(..., description="The work pool's worker name", alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None

Delete a work pool’s worker

​

Classes

​

WorkerLookups

​

WorkerChannelSetupError

​

WorkerChannelWorkPoolUpdateEvent