<@ULVA73B9P> how to deploy a prefect deployment fr...
# marvin-ai
a
@Marvin how to deploy a prefect deployment from python? I was using a deployment yaml file earlier, but not want to create this deployment using python sdk
m
thought for 182.2 seconds
Great question — in Prefect 3.x you can create deployments entirely from Python, no YAML needed. The core patterns are: - Single flow: call
my_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)
Copy code
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
Copy code
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
Copy code
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 docs
a
@Marvin few things 1. the flow code is not available, i am trying to deploy a flow (which is inside a container image) from a server to the self hosted prefect in the same cluster 2. I am also setting the working dir to /app in my docker container
m
thought for 304.2 seconds
Got it — you’ve got a prebuilt image that already contains the flow code, and you want to register a deployment that points at that image (no code checkout on the machine doing the registration). That’s a great fit for Prefect 3.x using RunnerDeployment directly. Key points for your case - You do NOT need the flow code at registration time. - You can register a deployment by specifying: - the image to run - the entrypoint to your flow inside the image - the work pool/queue - Since your image sets WORKDIR to /app, make the entrypoint path relative to /app. Minimal Python example (no flow import needed)
Copy code
from 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 docs
a
@Marvin I had to set pull: - prefect.deployments.steps.set_working_directory: directory: /app in my prefect-yaml. how to set this here
m
thought for 90.1 seconds
Short answer: there isn’t a Python argument to add “pull steps” (like
prefect.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:
Copy code
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:
Copy code
# Dockerfile
FROM python:3.12
WORKDIR /app
COPY . .
Then your entrypoint can be relative to /app:
Copy code
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:
Copy code
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:
Copy code
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:
Copy code
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:
Copy code
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 docs
a
@Marvin how do you give a schedule
m
thought for 117.2 seconds
You’ll add schedules when you register the RunnerDeployment. Since you’re not importing the flow, use schedule objects plus the normalizer, then pass them to
.apply()
. Example: cron + interval schedules
Copy code
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
Copy code
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
.