Parvez Mohammed
10/02/2025, 2:11 PMMarvin
10/02/2025, 2:11 PMMarvin
10/02/2025, 2:14 PMfrom prefect import flow, get_run_logger
@flow
def etl(name: str = "world"):
logger = get_run_logger()
<http://logger.info|logger.info>(f"Hello, {name}!")
if __name__ == "__main__":
# Creates/updates a deployment for this flow
deployment_id = etl.deploy(
name="etl-dev",
work_pool_name="my-work-pool", # replace with your work pool
description="Example deployment defined in Python",
parameters={"name": "Prefect"},
cron="0 * * * *", # run hourly; or use interval=60 or remove for ad-hoc
tags=["example"],
# If your work pool uses containers (Docker, K8s, ECS), set an image:
# image="<http://ghcr.io/your-org/your-image:latest|ghcr.io/your-org/your-image:latest>",
# job_variables={} # pool-specific overrides; set as needed for your worker type
)
print(f"Created deployment: {deployment_id}")
Example B: Deploy a flow from a GitHub repo (code-based deployment from source)
from prefect import flow
from prefect.flows import Flow
@flow
def etl(name: str = "world"):
# Flow body can be minimal; actual code will be read from the repo at runtime
pass
if __name__ == "__main__":
Flow.from_source(
source="<https://github.com/your-org/your-repo>",
entrypoint="flows/etl.py:etl", # path within repo + ':' + flow function name
).deploy(
name="etl-prod",
work_pool_name="k8s-pool", # replace with your work pool
image="<http://ghcr.io/your-org/your-image:latest|ghcr.io/your-org/your-image:latest>",
parameters={"name": "Production"},
cron="0 0 * * *", # run daily at midnight
tags=["prod"],
# job_variables={} # optional, depends on your work pool (e.g., env, resources)
)
Notes
- This assumes Prefect 3.x. In 3.x, use Flow.from_source(...).deploy(...) for repo-based deployments and my_flow.deploy(...) for local code. The deprecated 2.x APIs like Deployment.build_from_flow() and prefect deployment build are removed.
- Make sure you have a worker running for the target work pool so scheduled and ad-hoc runs get picked up.
- job_variables are work-pool specific — set them to match your pool’s job template (e.g., env, resources, volumes).
Docs and references
- Create deployments via Python: Deploy via Python
- Deployment how-to overview: Create deployments
- REST/API reference for deployments: Create deployment API
Want me to tailor this to your setup? Share:
- Your work pool type/name (e.g., local, Docker, Kubernetes, ECS)
- Whether your flow code lives locally or in a Git repo
- Any schedule and default parameters you needParvez Mohammed
10/02/2025, 2:30 PMMarvin
10/02/2025, 2:30 PMMarvin
10/02/2025, 2:35 PMDeployment class; the 3.x equivalent is RunnerDeployment plus the deploy(...) helper. Below are complete 3.x scripts using that pattern. If you really are on 2.x and want the 2.x Deployment class example, say so and I’ll send it.
Prefect 3.x: code-based deployment with RunnerDeployment (local entrypoint)
from prefect import flow
from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.deployments.schedules import create_deployment_schedule_create
from prefect.schedules import Cron
@flow
def etl(name: str = "world"):
print(f"Hello, {name}!")
if __name__ == "__main__":
d = RunnerDeployment(
name="etl-dev",
flow_name="etl", # Must match the @flow function name
entrypoint="path/to/your_file.py:etl", # Update to your file path
description="Example deployment defined with RunnerDeployment",
parameters={"name": "Prefect"},
tags=["example"],
schedules=[create_deployment_schedule_create(Cron("0 * * * *"))], # hourly
work_pool_name="my-work-pool", # replace with your work pool
# job_variables={"env": {"FOO":"BAR"}} # optional: pool-specific overrides
)
# Optionally set an image (e.g., for Docker/K8s/ECS pools); build/push default True
[deployment_id] = deploy(d, image="<http://ghcr.io/your-org/your-image:latest|ghcr.io/your-org/your-image:latest>")
print(f"Created deployment: {deployment_id}")
Prefect 3.x: deployment from a Git repo (source + entrypoint)
from prefect import flow
from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.deployments.schedules import create_deployment_schedule_create
from prefect.schedules import Cron
from prefect.storage import create_storage_from_source
@flow
def etl(name: str = "world"):
pass # actual code will be loaded from the repo at run time
if __name__ == "__main__":
storage = create_storage_from_source("<https://github.com/your-org/your-repo>")
d = RunnerDeployment(
name="etl-prod",
flow_name="etl",
entrypoint="flows/etl.py:etl", # path in the repo + ":" + flow name
storage=storage,
description="Prod deployment from GitHub",
parameters={"name": "Production"},
tags=["prod"],
schedules=[create_deployment_schedule_create(Cron("0 0 * * *"))], # daily
work_pool_name="k8s-pool",
)
[deployment_id] = deploy(d, image="<http://ghcr.io/your-org/your-image:latest|ghcr.io/your-org/your-image:latest>")
print(f"Created deployment: {deployment_id}")
Notes
- In Prefect 3.x, the old 2.x Deployment.build_from_flow() and prefect deployment build were removed. Use RunnerDeployment or flow.deploy(...) instead.
- Make sure a worker is running for your target work pool so runs get picked up.
Docs
- Deploy via Python: Deploy via Python
- Create deployments overview: Create deployments
Are you on Prefect 2.x and specifically want the 2.x Deployment class pattern? If yes, tell me your storage (e.g., local path, GitHub), schedule, and work queue/pool and I’ll tailor the snippet.