Task groups and one shared concurrency limit

pythonapi-clientsautomationindependent-practice

Saved on this device

Full lesson · 4 learning steps Open any exercise for focused practice

Step 1 of 4 · intro card

Task groups and one shared concurrency limit

Concept and contract

Run cooperative jobs under one shared semaphore while retaining input order and propagating an incomplete group failure.

Asynchronous work is useful when jobs spend time waiting, but async syntax alone does not bound the number of active operations. This lesson uses cooperative fictional jobs with asyncio.sleep(0), not real HTTP calls.

One shared semaphore limits active jobs and a TaskGroup owns their lifetime. The synchronous wrapper uses asyncio.run only at a top-level caller boundary; an existing event loop would instead await the async operation directly. The input also caps the number of synthetic jobs at twenty, separately from the active-work limit.

Mechanism and worked example

Each job acquires the shared semaphore, increments an observed active count and yields once.

Each job acquires the shared semaphore, increments an observed active count and yields once. A finally block decrements the count even on failure.

Tasks are retained in input order and their results are read after the group exits successfully, so completion timing does not reorder the returned labels. A controlled FAIL marker raises inside a task, allowing TaskGroup to propagate a grouped failure rather than returning an apparently complete partial result.

Complete worked example

python

import asyncio

def concurrent_labels(items):
    if not isinstance(items, list) or len(items) > 20 or any(not isinstance(item, str) for item in items):
        raise ValueError("text list required")
    async def run():
        semaphore = asyncio.Semaphore(2)
        active, peak = 0, 0
        async def job(item):
            nonlocal active, peak
            async with semaphore:
                active += 1
                peak = max(peak, active)
                try:
                    await asyncio.sleep(0)
                    if item == "FAIL":
                        raise ValueError("controlled job failure")
                    return item.upper()
                finally:
                    active -= 1
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(job(item)) for item in items]
        return {"labels": [task.result() for task in tasks], "peak": peak}
    return asyncio.run(run())

Reasoning: Run cooperative jobs under one shared semaphore while retaining input order and propagating an incomplete group failure. Compare the stated values and failure boundary before changing the implementation.

Counterexamples and common mistakes

The counterexample creates a new semaphore inside each job.

The counterexample creates a new semaphore inside each job. Every job then owns a different allowance and the combined active count exceeds the intended limit.

Placing a blocking network client inside an async function is another common mistake: it still blocks the event loop. Do not swallow cancellation to force a success response. This fixture measures cooperative active work, not CPU parallelism or the behavior of an arbitrary production HTTP library.

Counterexample: a runnable policy defect

python

import asyncio

def concurrent_labels(items):
    if not isinstance(items, list) or len(items) > 20 or any(not isinstance(item, str) for item in items):
        raise ValueError("text list required")
    async def run():
        semaphore = asyncio.Semaphore(2)
        active, peak = 0, 0
        async def job(item):
            nonlocal active, peak
            async with asyncio.Semaphore(2):
                active += 1
                peak = max(peak, active)
                try:
                    await asyncio.sleep(0)
                    if item == "FAIL":
                        raise ValueError("controlled job failure")
                    return item.upper()
                finally:
                    active -= 1
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(job(item)) for item in items]
        return {"labels": [task.result() for task in tasks], "peak": peak}
    return asyncio.run(run())

Reasoning: This runnable counterexample violates the declared contract; the companion lab requires it to fail.

Independent practice: API client task

Add a limit parameter defaulting to two, accepting exact integers from one through four.

Add a limit parameter defaulting to two, accepting exact integers from one through four. Apply it to the one shared semaphore.

Begin with three labels under a limit of one, then three and finally an invalid boolean. Keep input order, failure propagation and the finally cleanup intact. You are exposing one bounded policy input rather than building a general concurrent crawler or retry scheduler.

Step 1: Expose a validated concurrency limit parameter. Step 2: Use it for the one semaphore shared by all jobs.

Practice starter: solve before reading the solution

python

import asyncio

def concurrent_labels(items):
    # Remove this placeholder, then adapt the retained baseline:
    # Expose a validated concurrency limit parameter.
    raise NotImplementedError
    if not isinstance(items, list) or len(items) > 20 or any(not isinstance(item, str) for item in items):
        raise ValueError("text list required")
    async def run():
        semaphore = asyncio.Semaphore(2)
        active, peak = 0, 0
        async def job(item):
            nonlocal active, peak
            async with semaphore:
                active += 1
                peak = max(peak, active)
                try:
                    await asyncio.sleep(0)
                    if item == "FAIL":
                        raise ValueError("controlled job failure")
                    return item.upper()
                finally:
                    active -= 1
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(job(item)) for item in items]
        return {"labels": [task.result() for task in tasks], "peak": peak}
    return asyncio.run(run())

Reasoning: Expose a validated concurrency limit parameter. Use it for the one semaphore shared by all jobs.

Hint 1: Create the semaphore once per run, outside job.
Hint 2: A boolean is not an accepted concurrency count.

Solution, tests and trade-offs

The fixture deliberately yields inside every acquired slot, making the expected peak active count observable for this controlled workload.

The fixture deliberately yields inside every acquired slot, making the expected peak active count observable for this controlled workload. Cases include empty input, three jobs, malformed input and a failing task.

The practice checks both sequential and wider allowed concurrency. A successful test proves these controlled scheduling and result-order properties; it does not establish a wall-clock speedup, a global rate limit or permission to issue many source requests.

Practice solution and acceptance checks

python

import asyncio

def concurrent_labels(items, limit=2):
    if not isinstance(items, list) or len(items) > 20 or any(not isinstance(item, str) for item in items):
        raise ValueError("text list required")
    if type(limit) is not int or not 1 <= limit <= 4:
        raise ValueError("limit must be 1..4")
    async def run():
        semaphore = asyncio.Semaphore(limit)
        active, peak = 0, 0
        async def job(item):
            nonlocal active, peak
            async with semaphore:
                active += 1
                peak = max(peak, active)
                try:
                    await asyncio.sleep(0)
                    if item == "FAIL":
                        raise ValueError("controlled job failure")
                    return item.upper()
                finally:
                    active -= 1
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(job(item)) for item in items]
        return {"labels": [task.result() for task in tasks], "peak": peak}
    return asyncio.run(run())

Reasoning: Create the semaphore once per run, outside job. A boolean is not an accepted concurrency count.

Interview reasoning and next step

Explain the differences among concurrency, request rate and parallel CPU execution.

Explain the differences among concurrency, request rate and parallel CPU execution. A semaphore bounds active work but can still permit a high rate of very short requests, so a real feed client also needs pauses and backoff.

TaskGroup supplies a structured lifetime and cancellation behavior; per-request timeouts and total deadlines remain separate policies. Keep synchronous file or SQLite operations outside the event loop unless their blocking behavior is deliberately isolated.

Optional stretch — after the core task works: Separate a request-rate budget from the active-work semaphore.

Computer practice: https://applydjinn.com/typorithm/python-automation-lab-20261105.zip. Run verify.py using your selected Python interpreter, then adapt the retained exercise baseline and check it with --practice followed by the lesson id. The standard-library lab uses owned temporary 127.0.0.1 loopback servers, fictional pages and disposable files; it never scans production feeds or opens the working ApplyDjinn database.

Run cooperative jobs under one shared semaphore while retaining input order and propagating an incomplete group failure.