<@ULVA73B9P> How to run a flow/deployment via pyth...
# ask-marvin
p
@Marvin How to run a flow/deployment via python with specified time to run?
m
thought for 116.6 seconds
If you want to schedule a run from Python in Prefect 3.x, the usual options are: For an existing deployment: use
run_deployment(..., scheduled_time=...)
For lower-level control: use the client and create a flow run with a
Scheduled(...)
state Example: run a deployment at a specific time
Copy code
python
from datetime import datetime, timedelta
from prefect.deployments import run_deployment

scheduled_time = datetime.now() + timedelta(minutes=30)

flow_run = run_deployment(
    name="my-flow/my-deployment",
    scheduled_time=scheduled_time,
    timeout=0,  # don't wait for completion; just create the run
)

print(flow_run.id)
A couple notes: -
name
is typically
"flow-name/deployment-name"
-
scheduled_time
should be a
datetime
-
timeout=0
is useful if you only want to submit the run and return immediately Lower-level client example
Copy code
python
from datetime import datetime, timedelta
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.states import Scheduled

@flow
def my_flow(x: int):
    print(x)

with get_client(sync_client=True) as client:
    flow_run = client.create_flow_run(
        flow=my_flow,
        parameters={"x": 42},
        state=Scheduled(
            scheduled_time=datetime.now() + timedelta(minutes=30)
        ),
    )

print(flow_run.id)
For a deployment via client
Copy code
python
from datetime import datetime, timedelta
from prefect.client.orchestration import get_client
from prefect.states import Scheduled

deployment_id = "YOUR-DEPLOYMENT-ID"

with get_client(sync_client=True) as client:
    flow_run = client.create_flow_run_from_deployment(
        deployment_id=deployment_id,
        state=Scheduled(
            scheduled_time=datetime.now() + timedelta(minutes=30)
        ),
    )

print(flow_run.id)
Important distinction - Calling a flow directly like
my_flow()
runs immediately - If you want it to run later, create a flow run/deployment run with a
Scheduled
state or use
run_deployment(..., scheduled_time=...)
- For deployment runs, a worker must be running to pick up the run at that time Relevant APIs: - `run_deployment` - Prefect client API - Prefect states If you want, I can give you the exact snippet for either: 1. schedule one existing deployment, or 2. create a deployment and then schedule it from Python.