<@ULVA73B9P> how can create a deployment of a flow...
# marvin-ai
k
@Marvin how can create a deployment of a flow in Prefect 2.20.11 using Deployment.apply() ?
m
thought for 156.6 seconds
Great question. In Prefect 2.20.11 you can create/register a deployment entirely from Python by constructing a
Deployment
and calling
apply()
. Here are the most common patterns, from minimal to more “real world.” Minimal (local code, default Process infrastructure, default work queue)
Copy code
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
Copy code
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)
Copy code
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
Copy code
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)
Copy code
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.
If you share whether you’re using agents + work queues or workers + work pools (and whether your code lives in remote storage vs. the runtime workspace), I can tailor the snippet exactly to your setup. Also, if you’re hitting failures in Concourse, drop the error message and I’ll help pinpoint the cause.