Deployments are server-side representations of flows that can be executed:
This guide covers how to trigger deployments on demand.
Prerequisites
In order to run a deployment, you need to have:
Run a deployment from the CLI
The simplest way to trigger a deployment run is using the Prefect CLI:
prefect deployment run my-flow/my-deployment
CLI options
Add parameters and customize the run:
# Pass parameters
prefect deployment run my-flow/my-deployment \
--param my_param=42 \
--param another_param="hello"
# Schedule for later
prefect deployment run my-flow/my-deployment --start-in "2 hours"
# Watch the run until completion
prefect deployment run my-flow/my-deployment --watch
# Set custom run name
prefect deployment run my-flow/my-deployment --name "custom-run-name"
# Add tags to the run
prefect deployment run my-flow/my-deployment --tag production --tag critical
Run a deployment from Python
Use the run_deployment
function for programmatic control:
from prefect.deployments import run_deployment
# Basic usage
flow_run = run_deployment(
name="my-flow/my-deployment"
)
# With parameters
flow_run = run_deployment(
name="my-flow/my-deployment",
parameters={
"my_param": 42,
"another_param": "hello"
}
)
# With job variables (environment variables, etc.)
flow_run = run_deployment(
name="my-flow/my-deployment",
parameters={"my_param": 42},
job_variables={"env": {"MY_ENV_VAR": "production"}}
)
# Don't wait for completion
flow_run = run_deployment(
name="my-flow/my-deployment",
timeout=0 # returns immediately
)
# Wait with custom timeout (seconds)
flow_run = run_deployment(
name="my-flow/my-deployment",
timeout=300 # wait up to 5 minutes
)
# Schedule for later
from datetime import datetime, timedelta
flow_run = run_deployment(
name="my-flow/my-deployment",
scheduled_time=datetime.now() + timedelta(hours=2)
)
# With custom tags
flow_run = run_deployment(
name="my-flow/my-deployment",
tags=["production", "critical"]
)
By default, deployments triggered via run_deployment
from within another flow will be treated as a subflow of the parent flow in the UI. To disable this, set as_subflow=False
.
Async usage
In an async context, you can use the run_deployment
function as a coroutine:
import asyncio
from prefect.deployments import run_deployment
async def trigger_deployment():
flow_run = await run_deployment(
name="my-flow/my-deployment",
parameters={"my_param": 42}
)
return flow_run
# Run it
flow_run = asyncio.run(trigger_deployment())
Further reading