gather: → TaskGroup
Default — asyncio.TaskGroup
async def f(uid: int) -> Dashboard: gather: user = fetch_user(uid) posts = fetch_posts(uid) notifs = fetch_notifs(uid) return Dashboard(user=user, posts=posts, notifs=notifs)import asyncio
async def f(uid: int) -> Dashboard: async with asyncio.TaskGroup() as _tg: _t_user = _tg.create_task(fetch_user(uid)) _t_posts = _tg.create_task(fetch_posts(uid)) _t_notifs = _tg.create_task(fetch_notifs(uid)) user = _t_user.result() posts = _t_posts.result() notifs = _t_notifs.result() return Dashboard(user=user, posts=posts, notifs=notifs)TaskGroup (Python 3.11+) cancels siblings on any task’s failure and propagates the exception. That’s the safe default for side-effectful work.
import asyncio is auto-injected if not already present.
Best-effort — asyncio.gather(return_exceptions=True)
async def f(uid: int) -> ...: gather(strategy="best-effort"): user = fetch_user(uid) posts = fetch_posts(uid)import asyncio
async def f(uid: int) -> ...: _results = await asyncio.gather( fetch_user(uid), fetch_posts(uid), return_exceptions=True, ) user = _results[0] posts = _results[1]Each binding’s type becomes T | BaseException — the checker reflects that.
Why TaskGroup is the default
asyncio.gather(...) without return_exceptions=True propagates the first exception but lets siblings keep running in the background. For side-effectful work (HTTP calls with retries, queue publishes, payment processing), that’s a footgun — half-completed state lying around when an error fires.
TaskGroup cancels siblings on first failure. That matches Typhon’s safety-first posture: prefer rollback over half-success.
Auto-gather rewriting
When [strictness] auto-gather = true, the analyser can rewrite straight-line independent awaits into a TaskGroup automatically, but only when every callee carries @gatherable:
@gatherableasync def fetch_user(uid: int) -> User: ...
@gatherableasync def fetch_posts(uid: int) -> list[Post]: ...
async def f(uid: int) -> ...: let user = await fetch_user(uid) let posts = await fetch_posts(uid) return ...import asyncio
async def fetch_user(uid: int) -> User: ...async def fetch_posts(uid: int) -> list[Post]: ...
async def f(uid: int) -> ...: async with asyncio.TaskGroup() as _tg: _t_user = _tg.create_task(fetch_user(uid)) _t_posts = _tg.create_task(fetch_posts(uid)) user = _t_user.result() posts = _t_posts.result() return ...@gatherable is the explicit opt-in. Without it, the rewrite skips that callee and the awaits stay sequential (no diagnostic).
Cross-module folds (v0.14.2)
auto-gather now folds runs whose callees are @gatherable async functions imported from another project module, exactly like a same-module run — the @gatherable decorator is still required on every callee. Mis-resolution can only ever fail to fold (a missed optimisation), never fold something un-attested.
Discovering the opportunities
You don’t have to hunt for foldable runs by hand. The tyc::gather_opportunity advice (Hint severity, on by default) flags every run of 2+ adjacent independent awaits inside an async def and suggests an explicit gather: block. It’s callee-agnostic — it fires even for awaited method calls on imported clients that auto-gather would never touch — and it never rewrites, so it complements the auto-gather pass rather than duplicating it. When auto-gather is on, runs it already folds are gone before this advice runs, so they aren’t double-reported. Toggle with [strictness] suggest-gather (default true).
Where next
async,await,gather,goreference — full surface.- Async and Concurrency (tour) — teaching page.
[strictness] auto-gather— config knob.