<@ULVA73B9P> if I deploy a flow I now have a deplo...
# ask-marvin
a
@Marvin if I deploy a flow I now have a deployment. It will trigger on a schedule (if it exists) or I can trigger it manually with
prefect 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?
m
thought for 526.0 seconds
Great questions — here’s how this works in Prefect 3.x and how I’d choose between the options. High-level - Calling a flow function directly runs it ad-hoc in your current Python process. It shows up in the UI but is not attached to any deployment and doesn’t use workers/infrastructure. - Triggering a deployment creates a deployment-backed flow run that will be picked up by a worker (Docker/K8s/etc.), with schedule, retries, tags, and job variables applied. Answers to your questions 1) How should I run a flow from code? via a deployment or ad-hoc? - Use ad-hoc (call the function) for local dev/tests, synchronous “just run this now” inside the same process, or tiny internal jobs. - Use a deployment when you want Cloud/worker execution, consistent infra, schedules, permissions, and observability tied to that deployment. For “cron job and a user-trigger entrypoint,” keep a deployment and trigger it both on its schedule and programmatically. 2) Should I run a flow from code by just calling the function, or using an API in the prefect package? - If you want remote execution under a deployment, use the Prefect API: - Convenience helper:
Copy code
from 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):
Copy code
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:
Copy code
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:
Copy code
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(...)
If you’d like, share how you’re currently creating deployments (code vs prefect.yaml). I can suggest a minimal pattern to keep dev/prod duplication small across 10+ flows.
a
@Marvin I don't actually have a prod/dev deployment here. In the case of a flow that I want to run on a timer but also be able to trigger, should I have just one deployment for it, or one for each? Is it easier to track or debug runs in one case compared to the other?
m
thought for 151.3 seconds
Short answer: use a single deployment per flow for both the timer and the manual trigger. Why - Simpler: one place for schedules, parameters, infra, notifications. - Easier to track/debug: all runs are grouped under the same deployment. You can still distinguish how they were created. - Scheduled runs have
auto_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:
Copy code
prefect deployment run flow_name/deployment_name
- Python:
Copy code
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:
Copy code
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:
Copy code
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.
a
@Marvin how do you do the equivalent of the flow_run_name parameter for an automated run based on a schedule? I want the name to follow a custom template instead of the default
m
thought for 152.2 seconds
In Prefect 3.x, scheduled runs don’t take a deployment-level name template. To control run names for both scheduled and manual runs, set the template on the flow itself. What to do - Define a template on the flow:
Copy code
from 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:
Copy code
# 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:
Copy code
prefect deployment run flow/deployment --flow-run-name "manual-{dataset}-{run_date}" --param dataset=sales --param run_date=2025-01-01
Or programmatically:
Copy code
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.
a
@Marvin how can I set a parameter based on a dynamic value in a schedule, for example the current date?
m
thought for 142.1 seconds
Short answer: schedule parameters are static in Prefect 3.x. To use “current date” (or similar) per run, compute it at runtime inside your flow based on the scheduled start time, or trigger via an Automation that supports Jinja. Two practical patterns 1) Compute inside the flow using runtime context - Access the scheduled time via
prefect.runtime.flow_run.scheduled_start_time
and derive your parameter if it wasn’t provided.
Copy code
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:
Copy code
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:
Copy code
# 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.