Short answer: yes—Prefect 3.x can control this from the Prefect side. You have a few layers you can use, and they work across pods/workers so they’ll protect your DB cluster-wide. Here are the most common patterns: 1) Global concurrency limits (recommended for shared resources like a DB) - Create a named global concurrency limit (GCL) once, then acquire a slot in code anywhere you hit the DB. Prefect coordinates this across all pods/subflows/tasks. - CLI:

prefect gcl create db --limit 10

This ensures at most 10 concurrent DB operations across your entire Prefect deployment. - Wrap DB access in your tasks/subflows: Async version:

from prefect import flow, task
  from prefect.concurrency.asyncio import concurrency

  @task
  async def fetch_from_db(query):
      async with concurrency("db"):
          return await do_async_query(query)

  @flow
  async def parent_flow(queries):
      results = await fetch_from_db.map(queries)
      return results

Sync version:

from prefect.concurrency.sync import concurrency

  @task
  def fetch_from_db_sync(query):
      with concurrency("db"):
          return do_sync_query(query)

- Docs: Global concurrency limits Tip: You can also rate-limit by setting a decay on the limit for requests-per-second control, e.g. create the limit with

--slot-decay-per-second

and use `rate_limit(...)`: - Docs: Rate limits 2) Work pool or work queue concurrency (limits number of concurrent flow runs/pods) - If each subflow is its own flow run and you just want fewer pods starting, set a pool/queue concurrency cap: - Work pool:

prefect work-pool set-concurrency-limit "my-pool" 10

- Work queue:

prefect work-queue set-concurrency-limit "my-queue" 10

- Docs: Work pools 3) Task runner concurrency (limits parallelism inside one flow run) - Useful if a single pod is over-parallelizing DB tasks:

from prefect import flow
  from prefect.task_runners import ThreadPoolTaskRunner

  @flow(task_runner=ThreadPoolTaskRunner(max_workers=5))
  def my_flow():
      ...

4) Tag-based concurrency (alternative pattern) - Tag tasks that access the DB (e.g.,

@task(tags={"db"})

) and configure a concurrency limit for that tag. This is a bit more implicit; most users prefer the explicit

concurrency("db")

guard. - Docs: Tag-based concurrency limits What I’d recommend for your case - Use a global concurrency limit to protect the DB, and wrap DB access in a concurrency context. This will cap requests across all subflows and pods. - Optionally, also cap your work pool or work queue to limit how many subflows run in parallel (and thus how many pods spin up). If you can share: - Are your DB calls inside tasks or in the subflow bodies? - Do you want a global cap on DB queries (e.g., 10 at a time) or a per-deployment/pod cap? - OSS or Cloud? I can tailor the exact snippets/commands for your setup. Helpful references: - Global concurrency limits - Tag-based concurrency limits - Work pools Verified CLI commands: -

prefect gcl create db --limit 10

-

prefect work-pool set-concurrency-limit "my-pool" 10

-

prefect work-queue set-concurrency-limit "my-queue" 10

Great — thanks for the details. Since your DB calls are in subflows, you’re on OSS, and you want a per-deployment/pod cap, you have two good levers in Prefect 3.x: Option A: Limit how many subflow pods run at once (per deployment) - Put the subflow deployment on its own work queue and set a concurrency limit on that queue. This caps how many subflow flow runs are picked up concurrently (i.e., how many pods spin up). - CLI (verified):

prefect work-queue create my-subflow-queue --pool my-pool --limit 3
  prefect work-queue set-concurrency-limit my-subflow-queue 3

- Assign your subflow deployment to

my-subflow-queue

in your deployment config (via

prefect deploy

config or UI). Docs: Work pools and queues Option B: Limit DB requests inside each pod (per-pod) - If you want each subflow pod to make at most N concurrent DB calls, limit concurrency inside the subflow itself: - Best practice: make the DB call a task, and set the subflow’s task runner max_workers. Sync example:

from prefect import flow, task
  from prefect.task_runners import ThreadPoolTaskRunner

  @task
  def query_db(item):
      # do blocking DB call here
      return do_sync_query(item)

  @flow(task_runner=ThreadPoolTaskRunner(max_workers=4))
  def subflow(items):
      futures = [query_db.submit(i) for i in items]
      return [f.result() for f in futures]

Async example with a local semaphore (if you don’t want to make it a task):

import asyncio
  from prefect import flow

  async def do_async_query(item): ...
  
  @flow
  async def subflow(items):
      sem = asyncio.Semaphore(4)  # per-pod cap
      async def guarded(item):
          async with sem:
              return await do_async_query(item)
      return await asyncio.gather(*(guarded(i) for i in items))

Notes - You can combine A and B: cap how many subflow pods run at once and cap DB concurrency within each pod. - I’d avoid global concurrency limits for your case since you asked for per-deployment/pod; GCLs are cluster-wide. If you ever need a global cap, see Global concurrency limits. Quick check to make sure we’re aligned: - Do you want to cap the number of subflow pods running concurrently, the per-pod DB concurrency, or both? If you confirm, I can give you the exact deployment/queue config snippet for your setup.