<!-- Generated from models/_canonical by scripts/build-model-variants.js. Edit the canonical source, not this file. Behavioural profile for Gemini: scripts/model-profiles.json -->

#Purpose

Rules for Python 3.12+ asyncio code. Async in Python is cooperative: one blocking call stalls every coroutine on the loop, and nothing warns you. Most "async is slow" reports are a sync call hiding inside an async def.

Framework specifics are Backend/fastapi; ORM async is Database/sqlalchemy.

---

#When to use it

Threads or multiprocessing beat asyncio for CPU; sync code beats it for simplicity. Choose it for I/O concurrency, and then go all the way — a half-async codebase gets the costs of both.

---

#Never block the loop

python
async def handler():
    data = requests.get(url).json()     # blocks the loop for the whole request
    time.sleep(1)                        # blocks every coroutine for 1s
    rows = session.execute(stmt)         # sync SQLAlchemy: blocks

async def handler():
    async with httpx.AsyncClient() as c:
        data = (await c.get(url)).json()
    await asyncio.sleep(1)
    rows = await asession.execute(stmt)
    report = await asyncio.to_thread(render_pdf, rows)   # CPU/sync → thread

• Every library call inside async def must be either awaited or wrapped in asyncio.to_thread. requests, boto3, psycopg2, open() on a network mount — all blocking.
• Enable debug mode in development: asyncio.run(main(), debug=True) or PYTHONASYNCIODEBUG=1 logs any callback that ran longer than 100 ms. This is how you find the hidden sync call.
• to_thread is bounded by the default executor (min(32, cpus+4) threads). Offloading thousands of calls to it serialises them; that is a sign to use an async driver instead.

---

#Structured concurrency: TaskGroup

python
async with asyncio.TaskGroup() as tg:
    t1 = tg.create_task(fetch_user(uid))
    t2 = tg.create_task(fetch_orders(uid))
user, orders = t1.result(), t2.result()

• TaskGroup (3.11+) cancels the siblings when one fails and re-raises as an ExceptionGroup. gather() by default lets the others keep running after one fails, and gather(return_exceptions=True) hands you exceptions as values you can forget to check.
• Fire-and-forget asyncio.create_task(coro()) without keeping a reference: the task can be garbage-collected mid-flight. Keep a reference or use a group.
• except* ValueError: to handle one member type of an ExceptionGroup.

---

#Cancellation and timeouts

python
async with asyncio.timeout(5):                 # 3.11+; raises TimeoutError
    await fetch()

try:
    await work()
except asyncio.CancelledError:
    await cleanup()                            # allowed: short, itself awaitable
    raise                                       # always re-raise

• asyncio.timeout() over wait_for(): it is a context manager, composes, and does not create an extra task.
• Every await is a cancellation point. Code that must not be interrupted (commit-then-ack) goes in asyncio.shield() or a finally block — and the finally block must be short, because cancellation can arrive again.
• Swallowing CancelledError breaks shutdown and TaskGroup semantics. Catch, clean up, re-raise.
• Timeouts on every external call, without exception. An unbounded await is a leaked connection under a network partition. → Backend/error-handling

---

#Async-native libraries

Create one client per process and reuse it; a new httpx.AsyncClient per request discards the connection pool and pays TLS every time. Close it in the application's shutdown hook.

---

#Sync boundaries

python
def cli_entry() -> None:
    asyncio.run(main())           # exactly one asyncio.run per process

# Calling async from sync code that is already inside a running loop:
# you cannot. Refactor the caller to be async, or run in a separate thread.

• asyncio.run() once, at the top. Nested run() calls raise; get_event_loop() in library code is deprecated behaviour.
• Semaphores for concurrency limits: sem = asyncio.Semaphore(20) around fan-out, or you will open 10,000 connections to a service that allows 100.
• Async generators need async with aclosing(gen) or explicit aclose(); an abandoned one holds its resources until finalised.

---

#Testing

python
# pyproject.toml → [tool.pytest.ini_options] asyncio_mode = "auto"

async def test_timeout_cancels_and_cleans_up(monkeypatch):
    async def slow(): await asyncio.sleep(10)
    with pytest.raises(TimeoutError):
        async with asyncio.timeout(0.01):
            await slow()

• pytest-asyncio in auto mode so every async def test_* just runs.
• Never await asyncio.sleep(0.5) to "let things settle" — assert on an event or a future. Real sleeps make the suite slow and still flaky on loaded CI.
• unittest.mock.AsyncMock for async dependencies; a plain Mock returns a non-awaitable and the test passes for the wrong reason. → Testing/pytest

---

#Anti-patterns

---

#Checklist

• Verify: Async chosen for I/O concurrency, not for CPU work
• Verify: No blocking call inside any async def; debug mode used to find them
• Verify: Sync work offloaded with asyncio.to_thread, sparingly
• Verify: Related tasks run under TaskGroup; no bare gather
• Verify: Every create_task result is retained
• Verify: CancelledError is cleaned up and re-raised, never swallowed
• Verify: Every external await has a timeout via asyncio.timeout()
• Verify: Non-interruptible sections use shield or a short finally
• Verify: Async-native drivers used; clients created once and closed at shutdown
• Verify: Fan-out bounded with a Semaphore
• Verify: Exactly one asyncio.run() per process
• Verify: Tests run in asyncio_mode = "auto" with AsyncMock; no real sleeps

---

#Anchors (restated last, read last)

The rules that must hold when you stop, repeated here because the end of the context is what you act on:

• Async chosen for I/O concurrency, not for CPU work
• No blocking call inside any async def; debug mode used to find them
• Sync work offloaded with asyncio.to_thread, sparingly
• Related tasks run under TaskGroup; no bare gather
• Every create_task result is retained
• CancelledError is cleaned up and re-raised, never swallowed

Before reporting done, prove the module still imports — run the line for this stack and paste its output:

bash
python -c "import <package>"          # Python: the package you changed
node -e "require('./<entry>')"       # Node CJS, or: node --input-type=module -e "import './<entry>.js'"
go build ./...                        # Go