Aniket Rawat
03/31/2026, 7:01 PMMarvin
03/31/2026, 7:02 PMMarvin
03/31/2026, 7:05 PMmy_flow.deploy(...)
- Flow code in a repo or folder: load it with Flow.from_source(...).deploy(...)
- Multiple flows in one go: use prefect.deployments.deploy(...)
Here are compact examples to get you started.
1) Single-file flow (build and register a deployment)
from prefect import flow
@flow(log_prints=True)
def my_flow(name: str = "world"):
print(f"Hello, {name}!")
if __name__ == "__main__":
my_flow.deploy(
name="hello-deployment",
work_pool_name="my-work-pool", # e.g., a Docker or Process pool you’ve created
image="my-registry/my-image:latest", # optional but typical for Docker pools
build=True, # build an image for this deployment
push=False, # push to registry if needed
parameters={"name": "Prefect"},
# optional scheduling (pick one):
# interval=300, # every 5 minutes
# cron="0 9 * * *", # every day at 9:00 UTC
# rrule="FREQ=DAILY;INTERVAL=1",
work_queue_name="high-priority", # optional
job_variables={"env": {"LOG_LEVEL": "INFO"}} # overrides base job template vars
)
2) Load your flow from source (e.g., GitHub repo or local folder) and deploy
from prefect import flow
# If the flow is not in this file:
# - source can be a Git URL or a local path
# - entrypoint is "<path/to/file.py>:<flow_function_name>"
if __name__ == "__main__":
flow.from_source(
source="<https://github.com/your-org/your-repo>", # or a local dir like "./"
entrypoint="flows/my_flow.py:my_flow",
).deploy(
name="repo-backed-deployment",
work_pool_name="my-work-pool",
image="my-registry/my-image:latest",
build=True,
push=False
)
3) Deploy multiple flows with a single image build
from prefect import flow, deployments
@flow
def flow_one():
print("Flow one")
@flow
def flow_two():
print("Flow two")
if __name__ == "__main__":
deployments.deploy(
flow_one.to_deployment("flow-one-deploy"),
flow_two.to_deployment("flow-two-deploy"),
work_pool_name="my-work-pool",
image="my-registry/my-shared-image:latest",
build=True,
push=False,
)
Notes and gotchas
- This is for Prefect 3.x. In 3.x, the old 2.x APIs are removed:
- Don’t use Deployment.build_from_flow() — use flow.deploy(...)
- Don’t use prefect deployment build — use prefect deploy (CLI) or the Python methods above
- You still need a running worker on your work pool to pick up runs.
- For fine-grained scheduling, you can also pass schedules=[...] with `prefect.schedules.Cron`/`Interval` objects if you prefer objects over strings.
Would you like a tailored example for your environment? If you can share:
- Which work pool type you’re using (process, docker, kubernetes)
- Whether your flow code lives locally or in Git
- If you need a schedule and/or default parameters
…I can drop in a ready-to-run snippet.
Docs: Prefect 3.x docsAniket Rawat
03/31/2026, 7:06 PMMarvin
03/31/2026, 7:06 PMMarvin
03/31/2026, 7:11 PMfrom prefect.deployments.runner import RunnerDeployment
# Adjust these to match your image and code layout inside the image
IMAGE = "my-registry/my-app:latest"
WORK_POOL = "my-docker-or-k8s-pool"
ENTRYPOINT = "flows/etl.py:etl_flow" # relative to /app because WORKDIR=/app
FLOW_NAME = "etl_flow" # must match the @flow function name
deployment = RunnerDeployment(
name="etl-prod",
flow_name=FLOW_NAME,
entrypoint=ENTRYPOINT,
work_pool_name=WORK_POOL,
work_queue_name="default", # optional
parameters={"run_mode": "prod"}, # optional
tags=["prod"], # optional
# If your code isn’t available to infer schema at apply-time, keep this True or set False if needed
enforce_parameter_schema=True,
)
# Register the deployment with Prefect
deployment_id = deployment.apply(
image=IMAGE, # prebuilt image containing your flow code
work_pool_name=WORK_POOL
)
print(f"Deployment registered: {deployment_id}")
About the working directory (/app)
- Since you set WORKDIR /app in your Dockerfile, Prefect will run your flow there. Your entrypoint should be relative to /app (e.g., flows/etl.py:etl_flow).
- If you ever need to set a working directory via infrastructure instead of Dockerfile:
- Docker work pool: customize the pool’s base job template to include a working_dir (if desired), or keep using WORKDIR in the image.
- Kubernetes work pool: customize the base job template to set spec.template.spec.containers[0].workingDir: "/app". Otherwise, WORKDIR in your image is sufficient.
Cluster/self-hosted checklist
- Make sure the machine running this script can reach your self-hosted Prefect API:
- Set PREFECT_API_URL (and PREFECT_API_KEY if auth is enabled) before running the script.
- Ensure a worker is running against your target work pool in the same cluster and can pull my-registry/my-app:latest (image pull secret, network, etc.).
- Start runs from the UI or via prefect deployment run after registration.
If you share:
- your work pool type (Docker vs Kubernetes)
- the exact path to your flow file and function name inside the image
…I can tailor the ENTRYPOINT and any job variable/template tips specifically for your setup.
Docs: Prefect 3.x docsAniket Rawat
03/31/2026, 7:19 PMMarvin
03/31/2026, 7:19 PMMarvin
03/31/2026, 7:21 PMprefect.deployments.steps.set_working_directory) to a deployment. That knob exists in prefect.yaml, but the Python SDK doesn’t expose pull-step configuration on Flow.deploy() or RunnerDeployment.apply().
Good news: you can get the same behavior without pull steps. Pick one of these:
1) Use an absolute entrypoint (no cwd needed)
If your image has the code at /app and you want to avoid cwd issues entirely, point the entrypoint at an absolute path:
from prefect.deployments.runner import RunnerDeployment
deployment = RunnerDeployment(
name="etl-prod",
flow_name="etl_flow",
entrypoint="/app/flows/etl.py:etl_flow", # absolute path avoids cwd reliance
work_pool_name="my-docker-or-k8s-pool",
)
deployment.apply(image="my-registry/my-app:latest", work_pool_name="my-docker-or-k8s-pool")
2) Set the working directory in the image (recommended)
If your Dockerfile sets WORKDIR, Prefect uses it:
# Dockerfile
FROM python:3.12
WORKDIR /app
COPY . .
Then your entrypoint can be relative to /app:
entrypoint="flows/etl.py:etl_flow"
3) Set the working directory via the work pool’s base job template
- Kubernetes: add workingDir: "/app" to the container in your pool’s base job template.
- Fetch the default template:
prefect work-pool get-default-base-job-template --type kubernetes --file k8s-template.json
- Edit k8s-template.json to set:
- job_configuration.job_manifest.spec.template.spec.containers[0].workingDir to /app
- Apply it:
prefect work-pool update my-k8s-pool --base-job-template k8s-template.json
- Docker: if you prefer setting working_dir at the pool level, fetch the default Docker template, add a working_dir override, then update:
prefect work-pool get-default-base-job-template --type docker --file docker-template.json
# edit docker-template.json to include working_dir: "/app"
prefect work-pool update my-docker-pool --base-job-template docker-template.json
4) Process work pool only
If you were using a Process work pool, you can set:
job_variables={"working_directory": "/app"}
when calling deploy() or .apply().
If you share which work pool you’re using (Docker vs Kubernetes), I can show the exact JSON snippet to drop into the base job template. Docs: Prefect 3.x docsAniket Rawat
03/31/2026, 9:26 PMMarvin
03/31/2026, 9:29 PMMarvin
03/31/2026, 9:31 PM.apply().
Example: cron + interval schedules
from datetime import timedelta, datetime
from prefect.deployments.runner import RunnerDeployment
from prefect.deployments.schedules import normalize_to_deployment_schedule
from prefect.schedules import Cron, Interval
IMAGE = "my-registry/my-app:latest"
POOL = "my-docker-or-k8s-pool"
# Define one or more schedules
schedules = normalize_to_deployment_schedule([
# Daily at 9 AM New York time with custom parameters
Cron("0 9 * * *", timezone="America/New_York", slug="daily-9am",
parameters={"run_mode": "prod"}),
# Every 30 minutes (UTC); anchor_date optional for Interval
Interval(timedelta(minutes=30), timezone="UTC", slug="every-30m"),
])
# Create deployment without importing the flow code
deployment = RunnerDeployment(
name="etl-prod",
flow_name="etl_flow", # your @flow function name
entrypoint="/app/flows/etl.py:etl_flow", # absolute path inside the image
work_pool_name=POOL,
# paused=True, # optional: register with schedules paused
)
# Register with schedules
deployment_id = deployment.apply(
image=IMAGE,
work_pool_name=POOL,
schedules=schedules,
)
print(f"Deployment registered: {deployment_id}")
RRule example
from prefect.deployments.runner import RunnerDeployment
from prefect.deployments.schedules import normalize_to_deployment_schedule
from prefect.schedules import RRule
schedules = normalize_to_deployment_schedule([
RRule("FREQ=DAILY;INTERVAL=1", timezone="UTC", slug="daily-rrule")
])
deployment = RunnerDeployment(
name="etl-rrule",
flow_name="etl_flow",
entrypoint="/app/flows/etl.py:etl_flow",
work_pool_name="my-docker-or-k8s-pool",
)
deployment.apply(
image="my-registry/my-app:latest",
work_pool_name="my-docker-or-k8s-pool",
schedules=schedules,
)
Notes
- Use Cron, Interval, or RRule from prefect.schedules, then pass through normalize_to_deployment_schedule(...) before .apply(...).
- Set timezone on each schedule (defaults to UTC if omitted).
- You can set parameters per schedule; they’ll be applied to runs created by that schedule.
- If you want the deployment to start paused, set paused=True on RunnerDeployment.