@Marvin How do I put flow runs in a workqueue without blocking? I currently have a for loop calling run_deployment, but I want to submit a lot of flowruns without waiting for them to finish

✅ thought for 187.6 seconds

timeout=0 we'll see if marvin gets it

drumroll

Short answer: don’t wait on the runs. In Prefect 3.x, you can enqueue flow runs to a work queue “fire-and-forget” style by either: - calling

run_deployment(..., as_subflow=False)

in a loop, or - using the Prefect client to call

create_flow_run_from_deployment(...)

repeatedly. If you’re calling from inside another flow, the key is

as_subflow=False

. Otherwise the parent flow will treat each run as a subflow and wait for them. Example using run_deployment (simple and non-blocking):

from prefect.deployments import run_deployment

run_ids = []
for i in range(1000):
    fr = run_deployment(
        name="my-flow/my-deployment",           # "flow_name/deployment_name"
        parameters={"i": i},
        tags=["bulk"],
        work_queue_name="my-queue",             # optional override
        as_subflow=False                        # fire-and-forget
    )
    run_ids.append(fr.id)

print(f"Submitted {len(run_ids)} flow runs")

Example using the client (also non-blocking and gives you more control):

from prefect.client.orchestration import SyncPrefectClient

client = SyncPrefectClient()

# If you know the deployment ID already, you can use it directly.
# Otherwise, look up the deployment once by name, then reuse its id.
deployment = client.read_deployment_by_name("my-flow/my-deployment")

run_ids = []
for i in range(1000):
    fr = client.create_flow_run_from_deployment(
        deployment_id=deployment.id,
        parameters={"i": i},
        tags=["bulk"],
        work_queue_name="my-queue",             # optional override
        idempotency_key=f"bulk-{i}"             # optional, avoids accidental duplicates
    )
    run_ids.append(fr.id)

print(f"Submitted {len(run_ids)} flow runs")

Notes: - These calls return immediately after the flow run is created; they do not wait for completion. - Your worker(s) will pick runs from the target work queue as capacity allows. If you’re submitting a large burst, consider your work pool’s concurrency limits and scaling. - If you want to check on completion later, store the returned IDs and poll them or view them in the UI. Docs for reference: - Run deployments programmatically - API: create flow run from deployment Quick question to make sure I’m steering you the right way: are you triggering these from inside another Prefect flow, or from a plain Python script/notebook? The

as_subflow=False

bit matters if it’s inside a flow.

wait @Nate is Marvin right

no, marvin is suggesting run_deployment timeout=0 by reimplementing that util by calling the client methods yourself

:(