Kledia
12/11/2025, 12:30 PMMarvin
12/11/2025, 12:30 PMMarvin
12/11/2025, 12:33 PMDeployment and calling apply(). Here are the most common patterns, from minimal to more “real world.”
Minimal (local code, default Process infrastructure, default work queue)
from prefect import flow
from prefect.deployments import Deployment
@flow
def my_flow():
return "hi"
# Best in 2.x: build from the flow object so entrypoint is inferred
deployment = Deployment.build_from_flow(
flow=my_flow,
name="my-deployment",
work_queue_name="default", # or your queue
)
deployment.apply() # registers/updates the deployment
Agent model: custom infrastructure block (e.g., Docker) + work queue
from prefect import flow
from prefect.deployments import Deployment
from prefect.infrastructure import DockerContainer
@flow
def my_flow():
...
docker = DockerContainer(
image="my-registry/my-image:latest",
image_pull_policy="ALWAYS",
env={"PREFECT_LOGGING_LEVEL": "INFO"},
)
deployment = Deployment.build_from_flow(
flow=my_flow,
name="docker-deploy",
work_queue_name="prod", # agents poll queues
infrastructure=docker, # infra block is used with agents
)
deployment.apply()
Worker model: target a work pool (no infra block; use job_variables)
from prefect import flow
from prefect.deployments import Deployment
@flow
def my_flow():
...
deployment = Deployment.build_from_flow(
flow=my_flow,
name="pool-deploy",
work_pool_name="k8s-prod", # workers pull from pools
# Customize the worker’s base job template via job_variables
job_variables={
"env": {"PREFECT_LOGGING_LEVEL": "DEBUG"},
# other fields match your pool’s base job template
},
)
deployment.apply()
Remote storage (S3) with upload
from prefect import flow
from prefect.deployments import Deployment
from prefect.filesystems import S3
@flow
def my_flow():
...
storage = S3.load("my-s3-block") # created beforehand in UI/API
deployment = Deployment.build_from_flow(
flow=my_flow,
name="s3-deploy",
storage=storage,
path="flows", # path within the bucket
)
deployment.apply(upload=True) # uploads the project files to S3
Scheduling (2.x supports either a single schedule or multiple schedules; multiple is preferred)
from prefect.client.schemas.schedules import CronSchedule
from prefect.client.schemas.objects import MinimalDeploymentSchedule
deployment = Deployment.build_from_flow(
flow=my_flow,
name="scheduled-deploy",
schedules=[
MinimalDeploymentSchedule(
schedule=CronSchedule(cron="0 2 * * *"),
active=True,
)
],
)
deployment.apply()
What’s required for apply()
- At minimum, a deployment name and a flow target.
- If you use Deployment.build_from_flow(...), the flow target is inferred and the entrypoint is auto-generated.
- If you instantiate Deployment(...) manually instead, you must provide name, flow_name, and typically entrypoint and path if Prefect can’t infer them.
- Optional but common:
- Agent model: work_queue_name and an infrastructure block
- Worker model: work_pool_name and job_variables
- storage for remote code plus upload=True to push files
Notes and tips
- In CI (e.g., Concourse), ensure PREFECT_API_URL (and PREFECT_API_KEY for Cloud) are set before running this Python code so apply() can register with your server.
- apply(upload=True) only uploads when you’ve configured a remote storage block.
- In 2.x, Deployment.build_from_flow(...).apply() is the cleanest route; Deployment.apply() is deprecated in favor of 3.x patterns but is still supported in 2.20.11.
- When you move to 3.x, switch to flow.deploy(...) or Flow.from_source(...).deploy(...) and the prefect deploy CLI; Deployment.build_from_flow() is removed in 3.x.Marvin
12/11/2025, 12:33 PM