https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I have a task that needs a concurrenc...
# ask-marvin
s
@Marvin I have a task that needs a concurrency limiting setting. My task does two things: 1. run queries against an API that has strict rate limit, 4 concurrent connections limit. 2. from api results run some image processing via https streaming of a S3-like file. Can you write up an example where I can apply different concurrency limits for 1) and 2).
m
✅ thought for 168.2 seconds
Yep — the clean Prefect 3.x way is to give the two tasks different tags and configure a separate concurrency limit for each tag. In your case: - API query task → limit to 4 concurrent - image/stream-processing task → limit to whatever is safe for your CPU/network/memory, say 2 as an example Example:
Copy code
python
from io import BytesIO

import requests
from PIL import Image
from prefect import flow, task
from prefect.futures import wait


@task(tags=["api-query"])
def fetch_metadata(item_id: str) -> dict:
    # This task will be limited by the "api-query" tag
    resp = requests.get(
        f"<https://api.example.com/items/{item_id}>",
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


@task(tags=["image-stream"])
def process_image_from_url(image_url: str) -> tuple[int, int]:
    # This task will be limited by the "image-stream" tag
    # Example: stream image over HTTPS from S3-like storage
    resp = requests.get(image_url, stream=True, timeout=60)
    resp.raise_for_status()

    data = BytesIO()
    for chunk in resp.iter_content(chunk_size=1024 * 1024):
        if chunk:
            data.write(chunk)

    data.seek(0)

    with Image.open(data) as img:
        img.load()
        img.thumbnail((512, 512))
        return img.size


@task
def extract_image_url(api_result: dict) -> str:
    return api_result["image_url"]


@flow
def pipeline(item_ids: list[str]):
    futures = []

    for item_id in item_ids:
        api_result = fetch_metadata.submit(item_id)
        image_url = extract_image_url.submit(api_result)
        processed = process_image_from_url.submit(image_url)
        futures.append(processed)

    wait(futures)


if __name__ == "__main__":
    pipeline(["a1", "a2", "a3", "a4", "a5", "a6", "a7"])
Then create the limits in Prefect:
Copy code
bash
prefect concurrency-limit create api-query 4
prefect concurrency-limit create image-stream 2
How this behaves: - Prefect can schedule many task runs overall - but only 4 
fetch_metadata
task runs can be running at once - and only 2 
process_image_from_url
task runs can be running at once A couple practical notes: - Tag-based concurrency limits are still valid in Prefect 3.x - If your image processing is mostly network-bound, you may allow a higher limit - If it is CPU/memory heavy, keep that limit lower - These limits are independent, so you can tune them separately If you want even finer control, you can also put a global concurrency limit around only part of a task using 
prefect.concurrency.sync.concurrency
, but for your two-step pipeline, task tags are usually the simplest solution. Example of that pattern:
Copy code
python
from prefect import task
from prefect.concurrency.sync import concurrency


@task
def process_image_from_url(image_url: str) -> tuple[int, int]:
    with concurrency("image-download", occupy=1):
        resp = requests.get(image_url, stream=True, timeout=60)
        resp.raise_for_status()
        data = resp.content

    with concurrency("image-cpu", occupy=1):
        with Image.open(BytesIO(data)) as img:
            img.load()
            img.thumbnail((512, 512))
            return img.size
That approach is useful if you want separate limits for: - network streaming - CPU-heavy decoding/transforms If you want, I can also give you a version using: - async tasks with 
httpx
 - mapped tasks - or global concurrency limits (
gcl
) instead of tag limits. Relevant references: - Tag-based concurrency limits - Global concurrency limits - `prefect.concurrency.sync` API
s
@Marvin yes, please do a 
glc
example using the logic of this task; so that I don't have to break my task into several tasks like in your tag-based example:
Copy code
@task()
def create_img_prod():

     api_response = 'api_response.json'

     # run stac queries (need concurrency limit to comply with API rate limits)
     shell_output = ShellOperation(commands=[f"my_search_script.py  input_filenames > {api_response}"]).run()

     # run image processing using stac items and http streaming, both a GPU and network bound process
     # need concurrency limit for efficient use of GPU and http data streaming.
     shell_output = ShellOperation(commands=[f"process_images.py --collection {api_response}"]).run()
m
✅ thought for 83.2 seconds
Absolutely — if you want to keep this as one task, use global concurrency limits and wrap each section of the task in its own 
with concurrency(...)
block. Also, small correction: the CLI is 
gcl
, not 
glc
. Here’s a version using your exact pattern with 
ShellOperation
. Single-task example with different GCLs for each phase
Copy code
python
from prefect import task
from prefect.concurrency.sync import concurrency
from prefect_shell.commands import ShellOperation


@task
def create_img_prod():
    api_response = "api_response.json"

    # Phase 1: STAC/API queries
    # Limit this section to 4 concurrent users across all runs
    with concurrency(
        "stac-api",
        occupy=1,
        lease_duration=600,
        strict=True,
    ):
        ShellOperation(
            commands=[
                f"my_search_script.py input_filenames > {api_response}"
            ]
        ).run()

    # Phase 2: image processing / HTTP streaming / GPU use
    # Separate limit from API querying
    with concurrency(
        "image-processing",
        occupy=1,
        lease_duration=3600,
        strict=True,
    ):
        ShellOperation(
            commands=[
                f"process_images.py --collection {api_response}"
            ]
        ).run()
What this gives you - While the first block is running, the task consumes one slot from 
stac-api
- As soon as that block exits, that slot is released - Then the second block acquires a slot from 
image-processing
- So you can keep one Prefect task, but still enforce different limits for different parts of it That’s the main advantage of GCLs over tag-based limits here. Create the limits For your API limit of 4:
Copy code
bash
prefect gcl create stac-api --limit 4
For image processing, choose a number based on GPU/network capacity. For example, if only 1 should run at a time per workspace:
Copy code
bash
prefect gcl create image-processing --limit 1
Or if 2 image-processing sections can safely run at once:
Copy code
bash
prefect gcl create image-processing --limit 2
You can inspect them with:
Copy code
bash
prefect gcl inspect stac-api
prefect gcl inspect image-processing
Why 
strict=True
is useful With 
strict=True
, Prefect raises an error if the named concurrency limit does not exist. Without it, a typo like 
image-procesing
could silently skip protection. Recommended version for long-running shell commands If either shell command can run for a long time, I’d make the lease duration a bit more generous:
Copy code
python
from prefect import task
from prefect.concurrency.sync import concurrency
from prefect_shell.commands import ShellOperation


@task
def create_img_prod():
    api_response = "api_response.json"

    with concurrency(
        "stac-api",
        occupy=1,
        lease_duration=900,
        strict=True,
    ):
        ShellOperation(
            commands=[
                f"my_search_script.py input_filenames > {api_response}"
            ]
        ).run()

    with concurrency(
        "image-processing",
        occupy=1,
        lease_duration=7200,
        strict=True,
    ):
        ShellOperation(
            commands=[
                f"process_images.py --collection {api_response}"
            ]
        ).run()
If GPU and network should each have their own limit If your second script is constrained by both GPU availability and network streaming capacity, you can require slots from multiple GCLs at once: ```python from prefect import task from prefect.concurrency.sync import concurrency from prefect_shell.commands import ShellOperation @task def create_img_prod(): api_response = "api_response.json" with concurrency("stac-api", occupy=1, lease_duration=900, strict=True): ShellOperation( commands=[ f"my_search_script.py input_filenames > {api_response}" ] ).run() # Require both a GPU slot and a streaming slot with concurrency(
["gpu", "http-streaming"], occupy=1, lease_duration=7200, strict=True, ): ShellOperation( commands=[ f"process_images.py --collection {api_response}" ] ).run()
Copy code
Then create those limits separately, for example:

```bash
prefect gcl create gpu --limit 1
prefect gcl create http-streaming --limit 3
That means the second phase only runs when both are available. Important caveat This controls concurrency at the Prefect orchestration level for the wrapped code block. It does not limit internal parallelism spawned inside your shell script unless your script itself respects that. For example: - if 
process_images.py
starts 16 threads internally - Prefect still sees that as one protected section - so GCL limits how many task sections enter that code, not how many sub-processes/threads the script launches internally So this works best when: - 
my_search_script.py
represents one API-consuming unit of work - 
process_images.py
represents one GPU/network-consuming unit of work If your script itself fans out heavily, you may also need to tune the script’s own concurrency flags. Best-fit setup for your described case I’d probably do: - 
stac-api
→ limit 
4
- 
gpu
→ limit 
1
or however many GPUs/workers you truly have - 
http-streaming
→ limit based on how many large image streams you want in flight Example final version:
Copy code
python
from prefect import task
from prefect.concurrency.sync import concurrency
from prefect_shell.commands import ShellOperation


@task
def create_img_prod():
    api_response = "api_response.json"

    # API-limited phase
    with concurrency("stac-api", occupy=1, lease_duration=900, strict=True):
        ShellOperation(
            commands=[
                f"my_search_script.py input_filenames > {api_response}"
            ]
        ).run()

    # GPU + network-limited phase
    with concurrency(
        ["gpu", "http-streaming"],
        occupy=1,
        lease_duration=7200,
        strict=True,
    ):
        ShellOperation(
            commands=[
                f"process_images.py --collection {api_response}"
            ]
        ).run()
And corresponding setup:
Copy code
bash
prefect gcl create stac-api --limit 4
prefect gcl create gpu --limit 1
prefect gcl create http-streaming --limit 2
If you want, I can also turn this into a full flow example that submits many 
create_img_prod
runs in parallel so you can see exactly how the limits behave across multiple task runs. Useful references: - Global concurrency limits - `prefect.concurrency.sync` API - `prefect-shell` commands
s
@Marvin Sure send me the full flow example.
m
✅ thought for 23.6 seconds
Here’s a full example that keeps your logic in one task, but lets many task runs be submitted in parallel while Prefect enforces different GCLs for each phase. Example flow with parallel task submission + GCLs inside the task
Copy code
python
from pathlib import Path

from prefect import flow, task
from prefect.concurrency.sync import concurrency
from prefect.futures import wait
from prefect_shell.commands import ShellOperation


@task
def create_img_prod(input_filename: str, run_dir: str) -> str:
    run_path = Path(run_dir)
    run_path.mkdir(parents=True, exist_ok=True)

    api_response = run_path / "api_response.json"

    # Phase 1: STAC/API queries
    # At most 4 of these blocks can run at once across all task runs
    with concurrency(
        "stac-api",
        occupy=1,
        lease_duration=900,
        strict=True,
    ):
        ShellOperation(
            commands=[
                f"my_search_script.py {input_filename} > {api_response}"
            ]
        ).run()

    # Phase 2: image processing
    # This phase requires both a GPU slot and an HTTP streaming slot
    with concurrency(
        ["gpu", "http-streaming"],
        occupy=1,
        lease_duration=7200,
        strict=True,
    ):
        ShellOperation(
            commands=[
                f"process_images.py --collection {api_response}"
            ]
        ).run()

    return str(api_response)


@flow
def image_pipeline(input_filenames: list[str], base_run_dir: str = "./work") -> list[str]:
    futures = []

    for i, input_filename in enumerate(input_filenames):
        run_dir = f"{base_run_dir}/job_{i}"
        future = create_img_prod.submit(input_filename=input_filename, run_dir=run_dir)
        futures.append(future)

    wait(futures)
    return [f.result() for f in futures]


if __name__ == "__main__":
    image_pipeline(
        input_filenames=[
            "input_001.txt",
            "input_002.txt",
            "input_003.txt",
            "input_004.txt",
            "input_005.txt",
            "input_006.txt",
            "input_007.txt",
            "input_008.txt",
        ]
    )
How this behaves Say the flow submits 8 
create_img_prod
task runs. - All 8 may be submitted quickly - but only 4 can be inside the 
stac-api
block at once - after a task finishes the search step, it releases the 
stac-api
slot - then it tries to enter the image-processing section - that second section only starts when both: - a 
gpu
slot is free - a 
http-streaming
slot is free So you get phase-specific throttling without splitting the work into multiple tasks. Create the global concurrency limits first
Copy code
bash
prefect gcl create stac-api --limit 4
prefect gcl create gpu --limit 1
prefect gcl create http-streaming --limit 2
Example interpretation: - 
stac-api=4
→ up to 4 concurrent search phases - 
gpu=1
→ only 1 image-processing phase can use GPU at a time - 
http-streaming=2
→ up to 2 processing phases may stream imagery at once - because the processing section requires both 
gpu
and 
http-streaming
, effective concurrency there is the tighter of the two limits In the example above, that means only 1 processing phase runs at once. If you had:
Copy code
bash
prefect gcl create gpu --limit 2
prefect gcl create http-streaming --limit 3
then up to 2 processing phases could run concurrently. Recommended production tweak: check shell exit behavior If you want explicit failure behavior and logging around each shell phase, wrap 
run()
in normal Python error handling: ```python from pathlib import Path from prefect import flow, task, get_run_logger from prefect.concurrency.sync import concurrency from prefect.futures import wait from prefect_shell.commands import ShellOperation @task def create_img_prod(input_filename: str, run_dir: str) -> str: logger = get_run_logger() run_path = Path(run_dir) run_path.mkdir(parents=True, exist_ok=True)
api_response = run_path / "api_response.json" logger.info("Starting STAC query phase for %s", input_filename) with concurrency("stac-api", occupy=1, lease_duration=900, strict=True): ShellOperation( commands=[ f"my_search_script.py {input_filename} > {api_response}" ] ).run() logger.info("Finished STAC query phase for %s", input_filename) logger.info("Starting image processing phase for %s", input_filename) with concurrency( ["gpu", "http-streaming"], occupy=1, lease_duration=7200, strict=True, ): ShellOperation( commands=[ f"process_images.py --collection {api_response}" ] ).run() logger.info("Finished image processing phase for %s", input_filename) return str(api_response) @flow def image_pipeline(input_filenames: list[str], base_run_dir: str = "./work") -> list[str]: futures = [] for i, input_filename in enumerate(input_filenames): run_dir = f"{base_run_dir}/job_{i}" futures.append( create_img_prod.submit(input_filename=input_filename, run_dir=run_dir) ) wait(futures) return [future.result() for future in futures] ``` One more practical note If 
process_images.py
already manages its own GPU/network concurrency internally, then you may only need one limit for that block, for example just 
gpu
. But if you want Prefect to gate access to both shared resources at orchestration time, using both names in the same 
concurrency([...])
block is a good fit. If you want, I can also provide: - a version using 
ConcurrentTaskRunner
/ threadpool tuning - a version with retries and timeouts - or a version that uses 
shell_run_command
 instead of 
ShellOperation
for a lighter example. References: - Global concurrency limits - `prefect.concurrency.sync` API - `prefect-shell` documentation
3 Views
Open in Slack
PreviousNext
Powered by