trio review
Trio 0.34.0 is an async I/O runtime whose nurseries bind every child task to an explicit parent scope. Leaving a nursery waits for its children, and one child's failure cancels siblings before the errors return as an exception group. Cancel scopes carry deadlines through nested calls. The package also supplies bounded memory channels, capacity limiters, TCP and TLS streams, subprocess support, thread bridges, and a test clock. It runs its own event loop, so asyncio drivers do not become compatible merely because both APIs use `await`. Version 0.34.0 adds peak channel-buffer statistics, Python 3.15 beta support, better nursery-start error context, and a free-threaded worker fix.
Trio 0.34.0 installed 6 packages using 4 MB in 0.5 seconds and imported in 0.63 seconds in our sandbox, but choosing it also chooses a non-asyncio event loop for the whole call path. Use it when structured task lifetime is worth auditing every driver; reusable libraries should usually expose AnyIO instead.
We installed it
| Install | ✓ · 0.5s | 6 packages on disk · 4 MB |
| Import | ✓ | import trio in 0.63s · pure Python · py.typed · requires Python >=3.10 |
| Known vulns | 0 | (pip-audit) |
Answers from our run
Does trio install cleanly?
Yes. In a fresh container with an empty cache, pip install trio finished in 0.5s, leaving 6 packages and 4 MB on disk. pip-audit reported no known vulnerabilities.
What does trio need to run?
Python >=3.10, and nothing compiled: it is pure Python. In our run import trio succeeded in 0.63s, and the package ships py.typed for type checkers.
trio or anyio: which should you use?
anyio: Choose it when library code should keep structured scopes while supporting both asyncio and Trio. Trio 0.34.0 installed 6 packages using 4 MB in 0.5 seconds and imported in 0.63 seconds in our sandbox, but choosing it also chooses a non-asyncio event loop for the whole call path.
When should you not use trio?
A required PostgreSQL, Redis, cloud, or HTTP client only runs on asyncio. A bridge or replacement would sit on every important path.
Use it if
- A network service controls its runtime and wants task lifetime plus cancellation visible in nested scopes.
- Tests contain long deadlines that a fake clock can skip once all tasks block.
- Workers need bounded channels, explicit readiness, backpressure, and sibling cancellation on failure.
- The dependency set already supports AnyIO or Trio and has no required asyncio-only driver.
- A required PostgreSQL, Redis, cloud, or HTTP client only runs on asyncio. A bridge or replacement would sit on every important path.
- The code is a reusable library for both major Python async runtimes. AnyIO avoids imposing Trio on every caller.
- ASGI operations assume asyncio-specific servers or middleware. Hypercorn has Trio support, but many deployment recipes do not.
- The team cannot handle `ExceptionGroup` and `except*`. Nursery exits may group failures even when a single child caused them.
- Policy forbids beta or pre-1.0 runtimes. Trio documents compatibility changes, but its package classifier still says beta.
Setup reality
We installed trio 0.34.0 without a cache in a clean Python 3.12 Bookworm sandbox. pip finished in 0.5 seconds and left 6 packages occupying 4 MB. The pure-Python distribution declares 7 direct dependencies, requires Python >=3.10, and includes py.typed. import trio worked in 0.63 seconds, and pip-audit returned 0 known vulnerabilities. PyPI did not expose a recognized license value, although the repository offers MIT or Apache 2.0 terms.
Trio owns the running event loop. An asyncio-only database or cloud client cannot run directly under it; select a Trio-native implementation, an AnyIO interface, or a deliberately narrow bridge. An ASGI deployment also needs a server backend that supports Trio, followed by middleware tests on that exact backend. Shared async def syntax does not imply runtime compatibility.
Cancellation is delivered at checkpoints. CPU work that never awaits can monopolize the loop and pass its deadline without noticing. Send blocking functions through trio.to_thread.run_sync or restructure them to yield. If cancellation abandons a worker thread, the Python function underneath may continue running, so do not treat cancellation as proof that an external side effect stopped.
Start children inside nurseries and use nursery.start() when a parent needs a readiness handshake. Current Python reports multiple child failures with exception groups handled through except*. Tests can use pytest-trio or call trio.run() directly. Trio's fake clock advances only when every task is blocked; a spinning task still hangs. Version 0.34.0's peak channel statistic gives tests and production telemetry a concrete buffer high-water mark.
Patterns
Tie workers to one nursery run-tasks-in-a-nursery
import trio
async def worker(name: str) -> None:
await trio.sleep(1)
print("done", name)
async def main() -> None:
async with trio.open_nursery() as nursery:
nursery.start_soon(worker, "a")
nursery.start_soon(worker, "b")
print("both finished")
trio.run(main)The context waits for every child and cancels siblings on an uncaught error. `start_soon` takes positional arguments, so bind keywords first.
Apply an outer and inner deadline timeouts-with-cancel-scopes
import trio
async def main() -> None:
with trio.move_on_after(5) as scope:
await slow_call()
if scope.cancelled_caught:
print("gave up after 5s")
with trio.fail_after(5): # raises trio.TooSlowError
await slow_call()
# extend a deadline from inside
with trio.move_on_after(1) as scope:
scope.deadline = trio.current_time() + 30
await slow_call()These are synchronous context managers around async work. Expiry is observed at a checkpoint, not during uninterrupted CPU execution.
Catch selected nursery failures handle-exception-groups
import trio
async def main() -> None:
try:
async with trio.open_nursery() as nursery:
nursery.start_soon(might_fail)
nursery.start_soon(might_also_fail)
except* ValueError as group: # Python 3.11+
for exc in group.exceptions:
print("task failed:", exc)
except* OSError:
print("network problem")Strict grouping can wrap even one child error. Python 3.10 cannot parse `except*` and needs compatibility helpers.
Backpressure producers with a channel memory-channels
import trio
async def producer(send_channel) -> None:
async with send_channel:
for i in range(100):
await send_channel.send(i)
async def consumer(receive_channel) -> None:
async with receive_channel:
async for item in receive_channel:
await handle(item)
async def main() -> None:
send, receive = trio.open_memory_channel(max_buffer_size=10)
async with trio.open_nursery() as nursery:
nursery.start_soon(producer, send)
nursery.start_soon(consumer, receive)
trio.run(main)Receiver iteration finishes after all send clones close. Close unused original endpoints or they keep the channel logically open.
Offload a blocking function run-blocking-code-in-a-thread
import functools
import trio
async def main() -> None:
rows = await trio.to_thread.run_sync(query_database, "select 1")
# keyword arguments need partial
body = await trio.to_thread.run_sync(
functools.partial(requests.get, url, timeout=10)
)
# let the thread keep running if we are cancelled
await trio.to_thread.run_sync(flush_to_disk, abandon_on_cancel=True)Thread work cannot be forcibly stopped. Normal cancellation waits; abandon mode frees the Trio task while the function continues alone.
Call async code from a worker thread call-trio-from-a-thread
import trio
def blocking_worker(paths):
for path in paths:
data = process(path)
trio.from_thread.run(report_progress, path) # await an async fn
trio.from_thread.run_sync(counter.increment) # sync fn on the loop
async def main() -> None:
await trio.to_thread.run_sync(blocking_worker, paths)A Trio-created worker inherits the required token. Arbitrary external threads need an explicit token or supported portal.
Serve TCP connections concurrently tcp-server-and-client
import trio
async def echo(stream: trio.SocketStream) -> None:
async with stream:
async for chunk in stream:
await stream.send_all(chunk)
async def server() -> None:
await trio.serve_tcp(echo, 9000)
async def client() -> None:
stream = await trio.open_tcp_stream("127.0.0.1", 9000)
async with stream:
await stream.send_all(b"hello\n")
print(await stream.receive_some(1024))Hostname dialing races candidate addresses. Keep the long-lived server in the application nursery so shutdown cancels it predictably.
Wait until a service is ready wait-for-task-startup
import trio
async def serve(port: int, *, task_status=trio.TASK_STATUS_IGNORED) -> None:
listeners = await trio.open_tcp_listeners(port)
task_status.started(listeners[0].socket.getsockname())
await trio.serve_listeners(handler, listeners)
async def main() -> None:
async with trio.open_nursery() as nursery:
address = await nursery.start(serve, 0)
print("bound to", address) # guaranteed listening nowThe parent resumes after `started()` supplies readiness. Version 0.34.0 keeps cause and context if startup fails earlier.
Give cleanup a protected timeout shield-cleanup-from-cancellation
import trio
async def handle(conn) -> None:
try:
await serve_requests(conn)
finally:
with trio.CancelScope(shield=True) as scope:
scope.deadline = trio.current_time() + 2
await conn.send_goodbye()
await conn.aclose()Shielding blocks the outer cancellation while cleanup awaits. The nested 5-second scope prevents cleanup itself from hanging shutdown.
Time-limit a subprocess run-a-subprocess
import trio
async def main() -> None:
with trio.move_on_after(30):
result = await trio.run_process(
["ffmpeg", "-i", "in.mov", "out.mp4"],
capture_stdout=True,
capture_stderr=True,
)
print(result.returncode, result.stdout[:200])`check=True` raises on a nonzero status, and cancellation terminates the process. Streaming pipes need the lower-level process API.
Cap concurrent resource users limit-concurrency
import trio
async def main(urls) -> None:
limiter = trio.CapacityLimiter(8)
async def fetch(url: str) -> None:
async with limiter:
await download(url)
async with trio.open_nursery() as nursery:
for url in urls:
nursery.start_soon(fetch, url)The limiter constrains borrowers while any number of tasks may exist. Its statistics expose both capacity and the waiting queue.
Skip idle time in a test test-with-a-fake-clock
import pytest
import trio
from trio.testing import MockClock, wait_all_tasks_blocked
@pytest.mark.trio
async def test_retry_backoff(autojump_clock):
start = trio.current_time()
await retry_with_backoff(always_fails, attempts=5)
assert trio.current_time() - start == pytest.approx(31.0)
def test_standalone():
trio.run(main, clock=MockClock(autojump_threshold=0))pytest-trio injects this clock. It jumps only after every runnable task blocks, so a checkpoint-free loop still exposes starvation by hanging.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| anyio | PyPI | Choose it when library code should keep structured scopes while supporting both asyncio and Trio. |
| trio-asyncio | PyPI | Choose it to isolate a small, unavoidable asyncio dependency inside an otherwise Trio application. |
| curio | PyPI | Choose it for a controlled project that accepts Curio's smaller runtime and ecosystem. |
| uvloop | PyPI | Choose it when compatibility keeps the app on asyncio and only the event-loop implementation should change. |
More web backend guides
urllib3 · requests · ws · anyio · httpx · undici · the whole shelf →
How this guide is made: grounded in the library's documentation, release notes, changelog, and issue history, on a fixed rubric — not a hands-on install of every release. The 50 most-downloaded entries are additionally install-verified in clean containers. Corrections: contact the desk.