Yes—use tag-based concurrency limits with mapped tasks so only N run at a time, even if you map over a huge list. Two common patterns: 1) Cap concurrent API calls with a tag (recommended) - Tag the task and set a concurrency limit. This works across all flows/workers in your workspace.

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

@task(tags=["external-api"])
def call_api(url: str):
    r = requests.get(url, timeout=30)
    r.raise_for_status()
    return r.json()

@flow(task_runner=ThreadPoolTaskRunner(max_workers=50))  # worker threads can be high
def fetch_all(urls: list[str]):
    futures = call_api.map(urls)          # submits many tasks
    return futures.result()               # but only N run at once due to the tag limit

Set the limit:

prefect concurrency-limit create external-api 5
prefect concurrency-limit inspect external-api
prefect concurrency-limit ls
prefect concurrency-limit delete external-api

- Docs: How to limit concurrent task runs with tags - CLI ref: concurrency-limit CLI - Concept: Tag-based concurrency limits Notes: - This limits simultaneous API calls (e.g., 5 at a time). If your vendor enforces requests-per-second, pair the limit with a small sleep inside the task to shape QPS. - If you only need to throttle within a single flow run, you can also set the flow’s task runner lower, e.g.

ThreadPoolTaskRunner(max_workers=5)

. Tag limits are better when multiple flows/deployments share the same API. 2) Batch very large inputs to avoid creating too many task runs - Chunk the list; each task processes a small batch sequentially (optionally with a short sleep for strict QPS).

python
from prefect import flow, task
import time, requests

def chunks(items, size):
    for i in range(0, len(items), size):
        yield items[i:i+size]

@task(tags=["external-api"])
def process_batch(batch: list[str], sleep_s: float = 0.0):
    out = []
    for url in batch:
        r = requests.get(url, timeout=30)
        r.raise_for_status()
        out.append(r.json())
        if sleep_s:
            time.sleep(sleep_s)  # shape QPS if needed
    return out

@flow
def fetch_batched(urls: list[str], batch_size: int = 100):
    batched = list(chunks(urls, batch_size))
    futures = process_batch.map(batched, sleep_s=[0.0]*len(batched))
    # flatten results
    return [item for batch in futures.result() for item in batch]

When to use which: - Need workspace-wide control and simplicity: tag-based limit on the API task. - Extremely large lists (>10k) and want to reduce orchestration overhead: batching, optionally combined with a tag limit on the batch task. - Strict requests-per-second: combine a small

sleep

with a tag limit (tag = concurrent cap; sleep = pace). If you share a bit more about your vendor’s exact rate policy (e.g., “max 5 concurrent, 100 req/min”), I can suggest concrete numbers for the limit and batch size/sleep.

Got it — you’re concerned about Prefect Cloud/Server API rate limits, not your vendor’s API. Short answer: No — setting

ProcessPoolTaskRunner(max_workers=...)

does not, by itself, prevent hitting Prefect’s API rate limit when you map over a huge list. It caps how many tasks run concurrently, but a large

.map(...)

still submits a burst of task runs and their initial orchestration events (creation/initial state transitions/logs). That burst can exceed the Prefect API rate limit before execution is throttled by

max_workers

. What to do instead - Batch your map submissions to stage task-run creation - Or wrap work in a “batch task” so you create far fewer task runs - Optional: reduce per-item logging to cut API traffic further Example: stage task creation in batches

python
from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

def chunks(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i+size]

@task
def preprocess(item):
    # your work here
    return item

@flow(task_runner=ProcessPoolTaskRunner(max_workers=PARALLEL_PREPROCESSING_JOB_LIMIT))
def main(items, submit_batch_size: int = 200):
    results = []
    for batch in chunks(items, submit_batch_size):
        futures = preprocess.map(batch)   # creates a manageable number of task runs
        results.extend(f.result() for f in futures)
    return results

Example: use a batch-processing task to reduce task-run count

python
from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

@task
def preprocess_batch(batch):
    # sequential inside the task to avoid many task runs
    return [preprocess_one(x) for x in batch]

@flow(task_runner=ProcessPoolTaskRunner(max_workers=PARALLEL_PREPROCESSING_JOB_LIMIT))
def main(items, batch_size: int = 500):
    batches = [items[i:i+batch_size] for i in range(0, len(items), batch_size)]
    futures = preprocess_batch.map(batches)
    return [y for batch in futures.result() for y in batch]

Notes - Tag-based concurrency limits won’t solve this specific issue either — they cap how many tasks run at once, but they don’t stop a large map from generating lots of orchestration events quickly. - If you’re connected to Prefect Cloud, lowering logging verbosity (e.g.,

PREFECT_LOGGING_LEVEL=WARNING

) can help reduce API traffic from logs, but batching is the main fix. Helpful refs: - Run work concurrently - Task runners If you can share: - Roughly how many items you map at once - Whether you’re on Cloud or a self-hosted server - The exact error (e.g., 429s) and when they appear (on submit vs. start/finish) …I can help size a good batch size and submission cadence for your case.

Yes. Within a single flow run, all calls to

.map()

use the same task runner instance defined on that flow. So if your flow is declared with

ProcessPoolTaskRunner(...)

, every mapped task submission in that flow will run in processes managed by that same pool. A few clarifications: - One task runner per flow run: Prefect duplicates and starts the flow’s configured task runner at flow start, and reuses it for all task submissions within that flow run. - Multiple

.map()

calls share that runner: they all submit work to the same pool (same processes/threads). - Subflows get their own runner: a nested

@flow

call creates a new task runner instance for that subflow (it does not inherit the parent’s pool), unless you explicitly configure the subflow with the same runner. Docs for background: Task runners

however prefect actually has two rate limits, one for tasks/flows etc, and another one that they call their "orchestration limit" which task tag validation uses. this limit isn't documented, and you can't get usage from it, and is what caused us major slowdowns when submitting a lot of tasks

I’ve created us a wrapper internally to use:

def batch(size: int, sleep_time: float):
    """
    Decorator to add batched execution to a Prefect task.

    Adds a .batch() method to the task that internally calls .map()
    for each batch with a sleep between batches.

    Args:
        size: Number of items to process in each batch.
        sleep_time: Seconds to sleep between batches. Set to 0 for no sleep.

    The decorated task will have a .batch() method that:
    - Batches the iterable using iterate_batch()
    - Calls .map() for each batch
    - Collects all futures into a single PrefectFutureList
    - Sleeps for sleep_time seconds between batches
    - Returns PrefectFutureList[R] (same type as .map())

    Example:
        @batch(batch_size=50, sleep_time=5)
        @task
        def process_item(item: str) -> str:
            return item.upper()

        futures = process.batch(items, unmapped(other_arg))
    """

    def decorator(task_obj: Task) -> Task:
        @wraps(task_obj)
        def f(
            iterable,
            *args,
            **kwargs,
        ) -> PrefectFutureList[R]:
            # Construct a type that has `.result()` and so on
            all_futures: PrefectFutureList[R] = PrefectFutureList[R]()

            for batch in iterate_batch(iterable, size):
                batch_futures = task_obj.map(
                    batch,
                    # These will be parameters for the @task and
                    # iterable.
                    *args,
                    **kwargs,
                )
                all_futures.extend(batch_futures)

                # Give the Prefect API a small rest
                time.sleep(sleep_time)

            return all_futures

        # Make it available to call
        setattr(task_obj, "batch", f)

        return task_obj

    return decorator

and swapped from

.map

to

.batch

as the tag lookup would get 429'd by prefect, then wait, and try again!

Prefect claim that they're finishing work on updates to rate limits + observability to make it easier to see these limits. let's see if/when that happens

Haha I know what you mean. I was asking about observability for this beyond the webapp, and here’s some help I received:

It seems in the meantime you can view or download your rate limit usage with the following endpoints:

•

/api/accounts/{account_id}/rate-limits/usage

•

/api/accounts/{account_id}/rate-limits/usage/download

The

keys

parameter should be

logs-and-events

Link to the api: https://app.prefect.cloud/api/docs#get-/api/accounts/{account_id}/rate-limits/usage

in my understanding here, the main "caller" that is important is the prefect client library anyway. so I expect that it would "always" work?

it does feel a bit weird to have this rate limit anyway, if that's the case. I guess it catches things like if you have a dashboard or other external tool which makes requests to get status of flows/tasks

I agree. I kind of get it since someone could accidentally or maliciously DoS with tasks submission. What we wished was that they’d go into a queue, but then, it’s being handled so that would be a hit? It’s an odd balancing act

Again, really appreciate your help