Gabriel Rufino
11/25/2025, 9:16 PMkeep_jobs set to false so even if it's duplicating some job it's probably also deleting.
We're not tracking the flow state or anything like that from python, we just "fire and forget" these flows.
This is my code to launch them:
for model in collection_models:
scenarios = model_to_scenarios.get(model, [])
if not scenarios:
logger.warning("No scenarios found for model %s, skipping", model)
continue
# Submit task for this model's subflow with model-specific name
future = launch_model_subflow_task.with_options(
name=f"launch-{model}",
retries=2,
retry_delay_seconds=60,
).submit(
model=model,
scenarios=scenarios,
collection_limit_per_brand_model=request.collection_limit_per_brand_model,
)
deployment_futures.append(future)
and the definition:
flow_run = await run_deployment( # type: ignore[misc]
name=f"process-batch-subflow/{PREFECT_DEPLOYMENT_SUFFIX}-subflow",
parameters={
"model": model,
"collection_scenarios": scenarios,
"collection_limit_per_brand_model": collection_limit_per_brand_model,
},
job_variables={
"cpu": CPU_CORES,
"memory": "32G",
},
timeout=0, # Fire-and-forget: don't wait for subflow completion (execution timeout is set on the subflow itself)
tags=[model],
)Nathan Low
11/25/2025, 9:59 PMIsaac
11/26/2025, 2:23 PMBring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.
Powered by