Python structured concurrency with anyio

import anyio
from anyio import create_task_group, move_on_after, fail_after
from anyio.abc import TaskGroup

# Concurrent tasks with automatic cancellation
async def fetch_trace_data(trace_id: str) -> dict:
    async with create_task_group() as tg:
        results = {}

        async def fetch_trace():
            results['trace'] = await db.get_trace(trace_id)

        async def fetch_votes():
            results['votes'] = await db.get_votes(trace_id)

        async def fetch_tags():
            results['tags'] = await db.get_tags(trace_id)

        tg.start_soon(fetch_trace)
        tg.start_soon(fetch_votes)
        tg.start_soon(fetch_tags)

    return results

# Timeout with move_on_after (gives up, returns None-equivalent)
async def try_embed_with_timeout(text: str) -> list[float] | None:
    result = None
    with move_on_after(5.0):  # Gives up after 5 seconds
        result = await embed_text(text)
    return result

# Timeout that raises on expiry
async def embed_or_fail(text: str) -> list[float]:
    with fail_after(10.0):
        return await embed_text(text)

# Run from sync code
if __name__ == '__main__':
    # Run with asyncio (default)
    anyio.run(main)

    # Run with trio
    anyio.run(main, backend='trio')

# Library code that works with both
async def anyio_compatible_function() -> None:
    # Uses anyio primitives, not asyncio directly
    await anyio.sleep(1)
    async with anyio.open_file('data.txt') as f:
        content = await f.read()