Alastair
12/29/2025, 9:36 PMprefect deployment run flow/deployment. If I import the module from python and execute it, it'll also create a run (e.g. I may want a cron job and a way for users to trigger this process). I can run this function without creating a deployment - in the dashboard I see a flow with no related deployment. But I can also have multiple deployments for the same flow.
Questions:
• How should I run a flow from code? via a deployment or ad-hoc?
• Should I run a flow from code by just calling the function, or using an API in the prefect package?
• If I run it via a deployment, how do I select which deployment to attach it to if there are multiple for the flow?Marvin
12/29/2025, 9:37 PMMarvin
12/29/2025, 9:45 PMfrom prefect.deployments.flow_runs import run_deployment
# by name "flow_name/deployment_name"
flow_run = run_deployment(
name="my_flow/prod",
parameters={"x": 1, "y": 2},
tags=["manual-trigger"],
flow_run_name="manual-run-1",
# optional: schedule for later
# scheduled_time=datetime.utcnow() + timedelta(minutes=10),
# optional: override work queue
# work_queue_name="high-priority",
# optional: infrastructure overrides
# job_variables={"env.LOG_LEVEL": "DEBUG"}
)
print(flow_run.id)
- Full client control (async):
import asyncio
from prefect.client.orchestration import get_client
async def main():
async with get_client() as client:
deployment = await client.read_deployment_by_name("my_flow/prod")
flow_run = await client.create_flow_run_from_deployment(
deployment_id=deployment.id,
parameters={"x": 1, "y": 2},
tags=["manual-trigger"]
)
print(flow_run.id)
asyncio.run(main())
- If you just call the function, it runs locally and cannot be “attached” to a deployment after the fact:
result = my_flow(x=1, y=2) # ad-hoc local run, returns function result
3) If I run it via a deployment, how do I select which deployment to attach it to if there are multiple for the flow?
- Pass the exact deployment name or its UUID:
- By name: run_deployment(name="flow_name/deployment_name", ...)
- By ID: look up the deployment and use create_flow_run_from_deployment(deployment_id=...)
- CLI also supports this:
prefect deployment run flow_name/deployment_name
You can also pass --id <deployment_id> if you prefer IDs.
Notes for your dev/prod pattern
- Since your only difference is the code pull/source and you want local dev vs Cloud with Docker, keep separate deployments per environment (e.g., my_flow/dev and my_flow/prod) so each deployment points to the correct Flow.from_source(..., entrypoint=...) and runner/infra. Then:
- Local development: call the flow function directly or trigger the dev deployment.
- Production: trigger the prod deployment (on schedule or via API) so it uses Docker workers and the prod source.
- For many flows (10+), keep names consistent (e.g., flow_name/dev and flow_name/prod). Your “user-trigger” path can always call run_deployment("flow_name/prod", parameters=...) without duplicating orchestration logic in each flow.
Handy references
- CLI help (verified): prefect deployment run --help
- Docs home: Prefect 3.x Docs
- API to trigger deployments: prefect.deployments.flow_runs.run_deployment and client .create_flow_run_from_deployment(...)Marvin
12/29/2025, 9:45 PMAlastair
12/29/2025, 9:57 PMMarvin
12/29/2025, 9:57 PMMarvin
12/29/2025, 10:00 PMauto_scheduled=True and include the auto-scheduled tag.
- Manual triggers can add a tag (e.g., manual-trigger) or a flow_run_name prefix.
When two deployments help
- Different infrastructure or job variables (e.g., separate work pools/queues).
- Different defaults/parameters or secrets per trigger type.
- Different concurrency/priority or notification policies per trigger type.
- You want fully separate run histories/views.
How to trigger the single deployment
- CLI:
prefect deployment run flow_name/deployment_name
- Python:
from prefect.deployments.flow_runs import run_deployment
run = run_deployment(
name="flow_name/deployment_name",
parameters={"x": 1},
tags=["manual-trigger"],
flow_run_name="manual-myflow-{{timestamp}}"
)
Multiple schedules on one deployment
- If you need more than one timer, just add schedules to the same deployment:
prefect deployment schedule create flow_name/deployment_name --cron "0 9 * * *"
prefect deployment schedule create flow_name/deployment_name --interval 3600
- You can list and manage them:
prefect deployment schedule ls flow_name/deployment_name
Docs
- Deployments: Deployments
- Schedules: Schedules
If you share any special cases (e.g., different infra, params, or permissions for manual vs scheduled), I can suggest whether that justifies splitting into two deployments.Alastair
12/29/2025, 10:11 PMMarvin
12/29/2025, 10:11 PMMarvin
12/29/2025, 10:14 PMfrom prefect import flow
@flow(flow_run_name="etl-{dataset}-{run_date}")
def etl(dataset: str, run_date: str):
...
- The template can reference flow parameters only.
- For scheduled runs, the name is initially a random slug; when the flow starts, it’s renamed using the template and the effective parameters (deployment defaults + schedule overrides).
- Provide parameters so the template can render:
- In your deployment defaults or per-schedule overrides:
# defaults on the deployment
parameters={"dataset": "sales", "run_date": "2025-01-01"}
- Or per schedule via prefect deployment schedule create ... --param key=value
- For manual triggers, you can still override the name:
prefect deployment run flow/deployment --flow-run-name "manual-{dataset}-{run_date}" --param dataset=sales --param run_date=2025-01-01
Or programmatically:
from prefect.deployments.flow_runs import run_deployment
run_deployment(
name="flow/deployment",
parameters={"dataset": "sales", "run_date": "2025-01-01"},
flow_run_name="manual-sales-2025-01-01"
)
Notes and limits
- There isn’t a deployment- or schedule-level field for a name template in prefect.yaml or the CLI.
- The template can’t use run ID or scheduled time directly; only flow parameters. If you need timestamps, pass them as parameters (e.g., run_date) and compute them in schedule parameters.
If you share your desired naming pattern, I can suggest a minimal parameter setup so scheduled runs always render correctly.Alastair
12/30/2025, 10:51 AMMarvin
12/30/2025, 10:52 AMMarvin
12/30/2025, 10:54 AMprefect.runtime.flow_run.scheduled_start_time and derive your parameter if it wasn’t provided.
from datetime import timedelta
from prefect import flow
from prefect.runtime import flow_run
@flow
def process(data_date: str | None = None, lookback_days: int = 1):
if data_date is None:
scheduled = flow_run.scheduled_start_time
# e.g., use yesterday relative to scheduled time
data_date = (scheduled - timedelta(days=lookback_days)).strftime("%Y-%m-%d")
print(f"Processing {data_date}")
- Schedule with no dynamic params:
process.serve(name="daily-process", cron="0 9 * * *")
Or set only static values like lookback_days.
2) Use an Automation to compute params with Jinja
- Automations can render parameters with Jinja at trigger time (schedules themselves cannot).
- Example action conceptually:
# In an Automation "Run Deployment" action
parameters:
data_date:
__prefect_kind: jinja
template: "{{ now() | strftime('%Y-%m-%d') }}"
- This lets you inject the current date when the automation creates the run.
Notes
- There is no deployment- or schedule-level templating for parameters.
- If you want the date in the flow run name too, pass it as a parameter and reference it in your @flow(flow_run_name="...") template.