Sam Joel
10/03/2023, 3:52 PMTyson Chavarie
10/03/2023, 4:52 PMSam Joel
10/03/2023, 5:51 PMTyson Chavarie
10/03/2023, 5:55 PMNate
10/04/2023, 12:15 AMrun_deployment
like
from prefect.deployments import run_deployment
my_deployment = "my-flow/first-deployment"
flow_run = run_deployment(
name=my_deployment,
parameters={"answer": 42}, # assumes `my_flow` function accepts an `answer: int`
scheduled_time=datetime.now() + timedelta(hours=1) # start in an hour
# timeout = 0 # if you don't want this to wait for the triggered flow run to finish
)
assert flow_run.state.is_completed(), f"{flow_run.name} did not complete :("
Sam Joel
10/04/2023, 4:19 AM