# Asynchronous Programming for AI/LLM Applications (Python / FastAPI / Async/Await) ## Introduction Asynchronous programming in Python enables concurrent execution of I/O-bound operations without the overhead of traditional threading or multiprocessing. For AI and LLM applications, which frequently involve API calls, database operations, and streaming responses, async programming is essential for building scalable, high-performance systems. The async/await syntax in Python, combined with frameworks like FastAPI, allows developers to handle thousands of concurrent requests efficiently by leveraging cooperative multitasking through event loops. This approach maximizes resource utilization during I/O wait times—such as waiting for LLM API responses or database queries—making it ideal for production AI systems that require low latency, high throughput, and efficient resource management. Understanding async patterns, concurrency controls, error handling, and production deployment considerations is critical for building robust LLM-powered applications that can scale from prototype to production. --- ## Core Async Concepts and Fundamentals | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **Async/Await Syntax** | Python's native coroutine syntax using `async def` to declare coroutines and `await` to suspend execution until a result is available. | Defining asynchronous functions for I/O operations, API calls, database queries in LLM applications. | Clean, readable syntax; native Python support since 3.5; allows non-blocking I/O; maintains code flow similar to synchronous code. | Can only use `await` inside async functions; mixing sync/async code requires careful planning; not suitable for CPU-bound tasks. | `async def call_llm(prompt):`
  `response = await client.chat.completions.create()`
  `return response` | 1, 2, 3 | | **Event Loop** | The core execution mechanism in asyncio that manages and dispatches asynchronous tasks, handling I/O events and scheduling coroutine execution. | Running async applications; managing all concurrent operations; scheduling callbacks and tasks. | Single-threaded design simplifies debugging; efficient context switching; handles thousands of concurrent connections; automatic task scheduling. | Must not be blocked by CPU-intensive operations; only one event loop per thread; requires all operations to be async-compatible. | `import asyncio`
`asyncio.run(main()) # Python 3.7+`
`# Or: loop = asyncio.get_event_loop()`
`loop.run_until_complete(main())` | 1, 4, 5 | | **Coroutines** | Functions defined with `async def` that can be paused and resumed, returning a coroutine object that must be awaited. | Building asynchronous operations; creating reusable async functions; implementing LLM call wrappers. | Lightweight compared to threads; efficient memory usage; composable and reusable; easy to chain operations. | Must be awaited or scheduled; cannot call synchronous blocking code directly; requires event loop to execute. | `async def process_query(query):`
  `embedding = await get_embedding(query)`
  `results = await search_db(embedding)`
  `return results` | 1, 2, 6 | | **Awaitables** | Objects that can be used with the `await` keyword, including coroutines, Tasks, and Futures. | Waiting for async operations to complete; chaining multiple async operations; handling LLM responses. | Unified interface for different async objects; enables composition; supports error propagation through await chain. | All three types (coroutine, Task, Future) behave differently; forgetting to await leads to warnings; must handle exceptions. | `result = await some_coroutine()`
`result = await asyncio.create_task(coro())`
`result = await future_object` | 1, 4, 7 | | **Tasks** | Wrapper objects that schedule coroutines to run on the event loop, allowing concurrent execution of multiple coroutines. | Running multiple LLM calls concurrently; background processing; managing concurrent operations. | Enables true concurrency for I/O operations; can be cancelled; provides status tracking; supports timeouts. | Creates overhead if overused; must handle cancellation; need proper exception handling; can create race conditions. | `task1 = asyncio.create_task(llm_call1())`
`task2 = asyncio.create_task(llm_call2())`
`results = await asyncio.gather(task1, task2)` | 1, 4, 8 | | **asyncio.gather()** | Combines multiple awaitable objects and runs them concurrently, returning results in order when all complete. | Batch processing multiple prompts; parallel API calls; concurrent database operations. | Returns results in predictable order; supports return_exceptions for partial failure handling; simple API for parallel execution. | Waits for all tasks to complete; one failure can affect all if not using return_exceptions; can mask individual errors. | `prompts = ["Tell me about...", "Explain..."]`
`tasks = [call_llm(p) for p in prompts]`
`responses = await asyncio.gather(*tasks, return_exceptions=True)` | 1, 8, 9 | | **asyncio.as_completed()** | Returns an iterator that yields tasks as they complete, allowing processing of results as soon as available. | Processing LLM responses as they arrive; early termination on first success; progress tracking. | Process results immediately; better latency for first result; can break early; good for racing operations. | Results arrive in unpredictable order; more complex error handling; requires iteration management. | `tasks = [call_llm(p) for p in prompts]`
`for coro in asyncio.as_completed(tasks):`
  `result = await coro`
  `process_result(result)` | 1, 9, 10 | | **asyncio.wait()** | Low-level function that waits for multiple tasks with fine-grained control over completion conditions. | Waiting for first completion; timeout scenarios; complex task orchestration. | Flexible completion conditions (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED); supports timeout; returns done and pending sets. | More verbose than gather(); requires manual result extraction; deprecated in favor of TaskGroup in 3.11+. | `done, pending = await asyncio.wait(`
  `tasks,`
  `return_when=asyncio.FIRST_COMPLETED`
`)` | 1, 11, 12 | | **asyncio.TaskGroup (Python 3.11+)** | Context manager for structured concurrency with automatic error handling and task cancellation. | Managing groups of related tasks; ensuring proper cleanup; structured concurrency patterns. | Automatic task cancellation on error; structured lifetime management; cleaner exception handling; replaces gather() for many cases. | Requires Python 3.11+; less flexible than manual task management; cancels all tasks on first exception. | `async with asyncio.TaskGroup() as tg:`
  `task1 = tg.create_task(llm_call1())`
  `task2 = tg.create_task(llm_call2())`
`# Tasks automatically awaited and cleaned up` | 1, 13, 14 | --- ## FastAPI-Specific Async Patterns | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **Async Route Handlers** | Defining FastAPI endpoints with `async def` to handle requests asynchronously. | LLM API endpoints; database-backed routes; I/O-heavy operations. | Non-blocking request handling; higher throughput; can handle 1000s of concurrent requests; automatic async/await support. | Must use async libraries throughout; CPU-bound work blocks event loop; need to understand when to use async vs sync. | `@app.post("/generate")`
`async def generate_text(prompt: str):`
  `result = await llm_client.generate(prompt)`
  `return {"result": result}` | 15, 16, 17 | | **Sync Endpoints in FastAPI** | Using regular `def` (non-async) functions for endpoints that don't benefit from async. | CPU-bound operations; simple operations without I/O; when using sync-only libraries. | Automatically run in thread pool; simpler code for non-async operations; avoids event loop blocking. | Limited by thread pool size; higher overhead than async for I/O; can't use await inside. | `@app.get("/health")`
`def health_check():`
  `return {"status": "healthy"}`
`# Runs in thread pool executor` | 15, 18, 19 | | **Dependency Injection (async)** | FastAPI's async-aware dependency injection system for managing shared resources. | Database sessions; LLM clients; authentication; configuration. | Clean separation of concerns; automatic resource management; supports both sync and async dependencies; testable. | Async dependencies only work in async routes; must match sync/async types; can create complex dependency chains. | `async def get_db():`
  `async with AsyncSession() as session:`
    `yield session`

`@app.get("/data")`
`async def read(db=Depends(get_db)):` | 15, 20, 21 | | **Lifespan Events** | Context manager for managing application startup and shutdown logic in FastAPI. | Initializing database pools; creating LLM clients; loading models; cleanup operations. | Clean startup/shutdown pattern; replaces deprecated @app.on_event; ensures proper resource management; supports async. | Must use context manager pattern; replaces older event handlers; runs once per application lifecycle. | `@asynccontextmanager`
`async def lifespan(app: FastAPI):`
  `# Startup`
  `app.state.llm = AsyncClient()`
  `yield`
  `# Shutdown`
  `await app.state.llm.close()`

`app = FastAPI(lifespan=lifespan)` | 15, 22, 23 | | **BackgroundTasks** | FastAPI's built-in utility for running tasks after returning a response. | Logging; notifications; non-critical post-processing; fire-and-forget operations. | Simple API; no external dependencies; automatic execution after response; works with both sync and async. | No persistence; lost on server restart; no retry mechanism; not suitable for critical tasks; limited to single worker. | `@app.post("/process")`
`async def process(bg: BackgroundTasks):`
  `bg.add_task(log_to_db, data)`
  `return {"status": "processing"}` | 15, 24, 25 | | **Streaming Responses** | Using FastAPI's StreamingResponse for real-time data transmission. | LLM token streaming; large file downloads; real-time updates; progress reporting. | Low latency for first byte; efficient memory usage; better user experience; supports chunked transfer. | More complex error handling; client must support streaming; harder to debug; no automatic retry. | `async def token_stream():`
  `async for token in llm.stream(prompt):`
    `yield token`

`@app.get("/stream")`
`async def stream():`
  `return StreamingResponse(token_stream())` | 15, 26, 27 | | **Middleware (async)** | Async middleware for request/response processing in FastAPI. | Authentication; logging; request timing; header manipulation; rate limiting. | Processes all requests; supports async operations; can modify requests/responses; composable. | Runs on every request (overhead); must be async-safe; can block if not careful; order matters. | `@app.middleware("http")`
`async def add_process_time(request, call_next):`
  `start = time.time()`
  `response = await call_next(request)`
  `response.headers["X-Time"] = str(time.time() - start)`
  `return response` | 15, 28, 29 | | **AsyncClient for Testing** | Using httpx.AsyncClient for testing FastAPI async applications. | Unit testing async endpoints; integration testing; load testing. | Supports async test functions; realistic testing environment; can test streaming; works with pytest-asyncio. | Requires pytest-asyncio; more complex setup; need to manage event loops in tests. | `@pytest.mark.asyncio`
`async def test_endpoint():`
  `async with AsyncClient(app=app) as client:`
    `response = await client.get("/test")`
    `assert response.status_code == 200` | 15, 30, 31 | --- ## Async Database Operations | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **AsyncPG (PostgreSQL)** | High-performance async PostgreSQL driver for Python. | Vector database operations; storing LLM results; fast database access. | Fastest PostgreSQL driver for Python; native async support; connection pooling; prepared statements. | PostgreSQL only; not an ORM; requires manual query writing; steeper learning curve than SQLAlchemy. | `pool = await asyncpg.create_pool(dsn)`
`async with pool.acquire() as conn:`
  `rows = await conn.fetch(`
    `'SELECT * FROM embeddings')` | 32, 33, 34 | | **SQLAlchemy 2.0 Async** | ORM with full async/await support for multiple databases. | Complex data models; migrations; multi-database support; RAG applications. | ORM benefits with async; supports multiple databases; migration tools; widely adopted; rich ecosystem. | More overhead than raw drivers; complex configuration; connection pool management; requires understanding of async sessions. | `engine = create_async_engine(url)`
`AsyncSessionLocal = async_sessionmaker(engine)`
`async with AsyncSessionLocal() as session:`
  `result = await session.execute(query)` | 32, 35, 36 | | **Motor (MongoDB)** | Official async MongoDB driver for Python. | Document storage; unstructured data; LLM conversation history; flexible schemas. | Native MongoDB async support; full feature parity with PyMongo; good for document-heavy AI apps. | MongoDB-specific; requires different mental model than SQL; consistency trade-offs. | `client = AsyncIOMotorClient(uri)`
`db = client.get_database("llm_db")`
`await db.conversations.insert_one(doc)` | 37, 38, 39 | | **Connection Pooling** | Maintaining a pool of reusable database connections to reduce connection overhead. | High-throughput applications; concurrent database access; reducing latency. | Reduces connection overhead; improves performance; handles concurrent requests; automatic connection management. | Must configure pool size carefully; can exhaust connections; requires monitoring; complex error handling. | `# AsyncPG`
`pool = await asyncpg.create_pool(`
  `dsn, min_size=10, max_size=100)`

`# SQLAlchemy`
`engine = create_async_engine(`
  `url, pool_size=20, max_overflow=30)` | 32, 33, 40 | | **Async Session Management** | Proper handling of async database sessions in request/response cycles. | FastAPI endpoints; ensuring connection cleanup; transaction management. | Prevents connection leaks; proper transaction boundaries; works with dependency injection. | Must use async context managers; easy to forget cleanup; complex nested transactions. | `async def get_session():`
  `async with AsyncSession(engine) as session:`
    `yield session`

`@app.post("/data")`
`async def endpoint(db=Depends(get_session)):` | 36, 41, 42 | | **Async Transactions** | Using async context managers for database transactions. | Multi-step database operations; ensuring data consistency; rollback on errors. | ACID guarantees; automatic rollback; nested transaction support; integrates with async patterns. | Can deadlock; requires careful error handling; performance impact; must await commits. | `async with session.begin():`
  `await session.execute(stmt1)`
  `await session.execute(stmt2)`
  `# Auto-commits on exit` | 35, 36, 43 | | **Async Query Batching** | Grouping multiple queries to execute concurrently. | Fetching related data; reducing round trips; parallel data loading. | Reduced latency; better throughput; efficient resource usage. | Complex error handling; need to handle partial failures; can overwhelm database. | `tasks = [`
  `session.execute(query1),`
  `session.execute(query2),`
  `session.execute(query3)`
`]`
`results = await asyncio.gather(*tasks)` | 8, 9, 44 | --- ## Async HTTP Client Patterns | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **aiohttp ClientSession** | Async HTTP client with connection pooling and persistent sessions. | LLM API calls; external API integration; web scraping; data fetching. | High performance; connection reuse; supports HTTP/1.1 keep-alive; built-in timeouts; active development. | Must reuse sessions; complex API; requires manual session management; can leak connections. | `session = aiohttp.ClientSession()`
`async with session.get(url) as response:`
  `data = await response.json()`
`await session.close()` | 45, 46, 47 | | **httpx AsyncClient** | Modern async HTTP client with HTTP/2 support and requests-like API. | LLM API calls; testing; HTTP/2 applications; simpler API needs. | Clean API similar to requests; HTTP/2 support; good documentation; type hints; easier to learn. | Newer library; smaller ecosystem than aiohttp; fewer advanced features. | `async with httpx.AsyncClient() as client:`
  `response = await client.post(`
    `url, json=data, timeout=30.0)`
  `return response.json()` | 45, 48, 49 | | **Connection Pooling (HTTP)** | Reusing HTTP connections across multiple requests to the same host. | Multiple API calls to same LLM provider; reducing latency; improving throughput. | Reduces connection overhead; faster subsequent requests; better resource utilization; automatic in most async clients. | Must configure limits; can exhaust ports; requires session reuse; complex debugging. | `# aiohttp`
`connector = aiohttp.TCPConnector(`
  `limit=100, limit_per_host=30)`
`session = ClientSession(connector=connector)`

`# httpx`
`limits = httpx.Limits(max_connections=100)` | 45, 46, 50 | | **Timeout Configuration** | Setting timeouts for different phases of HTTP requests. | Preventing hung requests; SLA enforcement; fail-fast patterns; LLM API calls. | Prevents resource exhaustion; predictable failure modes; fine-grained control; separate connect/read timeouts. | Too short causes false failures; too long wastes resources; must handle timeout errors; needs tuning. | `# aiohttp`
`timeout = aiohttp.ClientTimeout(`
  `total=30, connect=5, sock_read=10)`

`# httpx`
`timeout = httpx.Timeout(`
  `connect=5.0, read=30.0, write=5.0)` | 45, 51, 52 | | **Keep-Alive Configuration** | Configuring persistent connections to avoid TCP handshake overhead. | High-frequency API calls; long-lived applications; connection efficiency. | Reduced latency; fewer connections; efficient TCP/TLS reuse; automatic in most clients. | Can hold resources; firewall/load balancer limits; must handle connection closure; requires monitoring. | `# aiohttp (default enabled)`
`connector = TCPConnector(`
  `force_close=False, # keep-alive`
  `ttl_dns_cache=300)`

`# httpx (default enabled)` | 46, 47, 53 | | **Retry with Exponential Backoff** | Automatically retrying failed requests with increasing delays. | Handling transient failures; API rate limits; network issues; LLM API resilience. | Recovers from transient errors; respects rate limits; configurable; prevents thundering herd. | Can mask real issues; increases latency; requires proper configuration; must limit retries. | `from tenacity import retry, wait_exponential`

`@retry(wait=wait_exponential(min=1, max=60))`
`async def call_api():`
  `async with client.get(url) as resp:`
    `return await resp.json()` | 54, 55, 56 | | **Circuit Breaker Pattern** | Preventing requests to failing services by "opening the circuit" after threshold failures. | Protecting LLM APIs from overload; fast failure; preventing cascade failures. | Prevents wasted resources; fast failure; automatic recovery testing; protects downstream services. | Adds complexity; requires state management; needs tuning; can miss recovery; requires distributed coordination at scale. | `class CircuitBreaker:`
  `def __init__(self, failure_threshold=5):`
    `self.failure_count = 0`
    `self.state = "CLOSED"`

  `async def call(self, func):`
    `if self.state == "OPEN":`
      `raise Exception("Circuit open")`
    `try:`
      `return await func()`
    `except Exception:`
      `self.failure_count += 1`
      `if self.failure_count >= threshold:`
        `self.state = "OPEN"` | 54, 57, 58 | --- ## Concurrency Control Patterns | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **Semaphores** | Limiting concurrent access to resources using a counter-based lock. | Rate limiting API calls; controlling concurrent LLM requests; resource pooling. | Simple API; prevents overload; fine-grained control; works across multiple coroutines. | Must manage acquire/release; can deadlock if not careful; doesn't prevent CPU blocking. | `sem = asyncio.Semaphore(5)`
`async def limited_call(prompt):`
  `async with sem:`
    `return await llm_api.call(prompt)` | 59, 60, 61 | | **BoundedSemaphore** | Semaphore variant that prevents releasing more than acquired. | Same as Semaphore but with safety checks. | Prevents programming errors; same performance as Semaphore; safer for production. | Slightly more overhead; same limitations as Semaphore. | `sem = asyncio.BoundedSemaphore(10)`
`# ValueError if release() called too many times` | 59, 62 | | **Locks** | Mutual exclusion primitive ensuring only one coroutine accesses resource at a time. | Protecting shared state; ensuring sequential access; preventing race conditions. | Simple mutual exclusion; prevents race conditions; async-aware. | Can cause bottlenecks; potential deadlocks; reduces concurrency; must not call blocking code. | `lock = asyncio.Lock()`
`async def update_state():`
  `async with lock:`
    `# Only one coroutine at a time`
    `state.update()` | 1, 63, 64 | | **asyncio.Queue** | Thread-safe FIFO queue for coordinating work between producer and consumer coroutines. | Work distribution; task queues; producer-consumer patterns; rate limiting. | Backpressure control via maxsize; multiple producers/consumers; thread-safe; prevents memory explosion. | Requires separate consumer tasks; more complex than gather(); manual error handling; unbounded can exhaust memory. | `queue = asyncio.Queue(maxsize=100)`
`await queue.put(item) # Producer`
`item = await queue.get() # Consumer`
`queue.task_done()` | 1, 65, 66 | | **asyncio.PriorityQueue** | Queue that retrieves items in priority order. | Prioritizing urgent LLM requests; SLA-based processing; deadline scheduling. | Built-in prioritization; same benefits as Queue; priority-based dequeuing. | Requires comparable items; more overhead; priority management complexity. | `pq = asyncio.PriorityQueue()`
`await pq.put((priority, item))`
`priority, item = await pq.get()` | 1, 67 | | **asyncio.LifoQueue** | Last-in-first-out (stack) queue for async operations. | Depth-first processing; cache-friendly patterns; specific algorithmic needs. | Stack semantics; same reliability as Queue; cache-friendly. | Less common use case; can starve old items; priority inversion risk. | `lifo = asyncio.LifoQueue()`
`await lifo.put(item)`
`item = await lifo.get() # Last item first` | 1, 67 | | **Worker Pool Pattern** | Pool of worker coroutines consuming tasks from a queue. | Batch processing; controlled concurrency; long-running task distribution. | Controlled resource usage; automatic load balancing; backpressure handling; scalable. | Requires queue management; more complex setup; need proper shutdown; error handling complexity. | `async def worker(queue):`
  `while True:`
    `item = await queue.get()`
    `await process(item)`
    `queue.task_done()`

`workers = [asyncio.create_task(`
  `worker(queue)) for _ in range(10)]` | 65, 68, 69 | | **Rate Limiting with Token Bucket** | Controlling request rate using token-based algorithm. | API rate limit compliance; preventing quota exhaustion; fair resource sharing. | Smooth rate limiting; burst handling; fair distribution; standard algorithm. | Requires state management; timing-sensitive; needs tuning; distributed systems complexity. | `class TokenBucket:`
  `def __init__(self, rate, capacity):`
    `self.rate = rate`
    `self.tokens = capacity`

  `async def acquire():`
    `while self.tokens < 1:`
      `await asyncio.sleep(1/rate)`
      `self._refill()`
    `self.tokens -= 1` | 70, 71 | --- ## Streaming and Real-Time Patterns | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **Server-Sent Events (SSE)** | Unidirectional server-to-client streaming over HTTP. | LLM token streaming; real-time notifications; progress updates; live data feeds. | Simple protocol; works over HTTP; auto-reconnect; widely supported; efficient for server-to-client. | Unidirectional only; limited browser connection limits; text-based format; no binary data. | `async def event_stream():`
  `async for token in llm.stream():`
    `yield f"data: {token}\n\n"`

`return StreamingResponse(`
  `event_stream(),`
  `media_type="text/event-stream")` | 72, 73, 74 | | **WebSockets** | Full-duplex communication channel over TCP. | Interactive LLM conversations; bidirectional streaming; real-time collaboration. | Bidirectional; persistent connection; low latency; binary support; standardized protocol. | More complex than SSE; requires WebSocket support; connection management overhead; load balancer complications. | `@app.websocket("/ws")`
`async def websocket_endpoint(websocket):`
  `await websocket.accept()`
  `while True:`
    `data = await websocket.receive_text()`
    `await websocket.send_text(response)` | 75, 76, 77 | | **Async Generators** | Coroutines that yield values using async iteration. | Streaming LLM responses; pagination; lazy data loading; memory-efficient processing. | Memory efficient; lazy evaluation; composable; works with async for; natural streaming pattern. | More complex error handling; must handle cleanup; can't use with sync code; requires async iteration. | `async def stream_tokens(prompt):`
  `async for chunk in llm_client.stream(prompt):`
    `yield chunk.content`

`async for token in stream_tokens(prompt):`
  `print(token, end="")` | 78, 79, 80 | | **Async Context Managers** | Context managers that support async enter/exit operations. | Resource management; connection handling; transaction boundaries; cleanup. | Guaranteed cleanup; exception-safe; composable; clear lifetime semantics. | Requires async with; more verbose; must handle exceptions in both __aenter__ and __aexit__. | `class AsyncResource:`
  `async def __aenter__(self):`
    `await self.connect()`
    `return self`
  `async def __aexit__(self, *args):`
    `await self.close()`

`async with AsyncResource() as res:` | 1, 81, 82 | | **Chunked Streaming** | Sending data in chunks rather than waiting for complete response. | Large LLM responses; file uploads/downloads; reducing time to first byte. | Lower latency; better user experience; memory efficient; supports progress tracking. | Complicates error handling; partial data issues; client must support chunked transfer; harder testing. | `async def chunked_response():`
  `for i in range(0, len(data), chunk_size):`
    `yield data[i:i+chunk_size]`
    `await asyncio.sleep(0) # yield control` | 26, 27, 83 | | **Async Iteration with Pagination** | Using async generators to handle paginated API responses. | Fetching all results from paginated APIs; lazy loading; memory-efficient data processing. | Handles large datasets; memory efficient; natural async flow; automatic rate limiting possible. | Complex state management; error recovery challenges; need to track pagination state. | `async def fetch_all_pages(url):`
  `page = 1`
  `while True:`
    `resp = await client.get(f"{url}?page={page}")`
    `data = await resp.json()`
    `if not data: break`
    `for item in data:`
      `yield item`
    `page += 1` | 78, 84 | --- ## Error Handling and Resilience | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **Try-Except in Async** | Standard Python exception handling adapted for async code. | Catching LLM API errors; handling timeouts; managing failures. | Familiar syntax; works same as sync code; can catch multiple exception types. | Must be inside async functions; doesn't automatically retry; need to handle CancelledError. | `async def safe_llm_call():`
  `try:`
    `result = await llm_api.call()`
  `except asyncio.TimeoutError:`
    `return default_response`
  `except APIError as e:`
    `log_error(e)`
    `raise` | 1, 85 | | **asyncio.TimeoutError** | Exception raised when operations exceed timeout duration. | LLM API timeouts; preventing hung operations; enforcing SLAs. | Built-in timeout mechanism; automatic cleanup; works with any awaitable. | Can interrupt at any point; need proper cleanup; different from socket timeout. | `try:`
  `async with asyncio.timeout(30.0):`
    `result = await long_operation()`
`except asyncio.TimeoutError:`
  `handle_timeout()` | 1, 86, 87 | | **CancelledError Handling** | Catching and handling task cancellation. | Graceful shutdown; cleanup on cancellation; cooperative cancellation. | Allows cleanup; propagates cancellation; standard protocol for cancellation. | Must handle explicitly; can be tricky with nested operations; needs careful testing. | `async def cancellable_task():`
  `try:`
    `await long_operation()`
  `except asyncio.CancelledError:`
    `await cleanup()`
    `raise # Re-raise!` | 1, 88, 89 | | **Tenacity Retry Decorator** | Declarative retry logic with exponential backoff using tenacity library. | LLM API retries; handling rate limits; transient error recovery. | Declarative syntax; many built-in strategies; async support; highly configurable. | External dependency; can hide issues; must configure carefully; increases latency. | `from tenacity import retry, stop_after_attempt`

`@retry(`
  `stop=stop_after_attempt(3),`
  `wait=wait_exponential(min=1, max=10)`
`)`
`async def call_with_retry():`
  `return await llm_api.call()` | 54, 55, 90 | | **Backoff Library** | Alternative retry library with decorator-based approach. | Simple retry scenarios; exponential backoff; jittered retries. | Simple API; lightweight; good defaults; supports both sync and async. | Less feature-rich than tenacity; basic configuration; async support requires separate decorator. | `import backoff`

`@backoff.on_exception(`
  `backoff.expo,`
  `aiohttp.ClientError,`
  `max_tries=5`
`)`
`async def fetch_data():` | 54, 91, 92 | | **Structured Exception Groups (3.11+)** | Handling multiple exceptions from concurrent operations. | Processing results from asyncio.gather with exceptions; bulk operations. | Native multiple exception handling; preserves all exception information; works with TaskGroup. | Python 3.11+ only; more complex syntax; need to handle multiple exception types. | `try:`
  `async with asyncio.TaskGroup() as tg:`
    `tg.create_task(task1())`
    `tg.create_task(task2())`
`except* ValueError as eg:`
  `handle_value_errors(eg.exceptions)` | 1, 13, 93 | | **Fallback Chains** | Trying multiple LLM providers or strategies in sequence. | LLM provider failover; degraded mode operation; ensuring availability. | High availability; graceful degradation; automatic fallback; increased reliability. | Increases complexity; can mask issues; multiple API costs; harder testing. | `async def call_with_fallback():`
  `for provider in [primary, secondary, tertiary]:`
    `try:`
      `return await provider.call()`
    `except Exception:`
      `continue`
  `raise AllProvidersFailed()` | 57, 94 | | **Dead Letter Queue Pattern** | Storing failed operations for later processing or analysis. | Handling persistent failures; forensic analysis; manual intervention. | Preserves failed operations; enables debugging; allows retry after fixes; audit trail. | Requires storage; needs monitoring; can grow unbounded; manual intervention needed. | `async def process_with_dlq(item):`
  `try:`
    `await process(item)`
  `except Exception as e:`
    `await dlq.enqueue(item, error=str(e))` | 95, 96 | --- ## Context and State Management | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **Context Variables (contextvars)** | Thread-safe context-local storage for async code. | Request ID tracking; user context; distributed tracing; per-request state. | Automatic context propagation in async; no need to pass through all function calls; isolated per-task. | Learning curve; limited to same event loop; need to understand context copying; Python 3.7+ only. | `from contextvars import ContextVar`

`request_id: ContextVar[str] = ContextVar('request_id')`

`async def process():`
  `request_id.set("req-123")`
  `# Available in nested calls`
  `await log_operation()` | 97, 98, 99 | | **app.state in FastAPI** | Application-level state storage in FastAPI. | Storing shared resources; LLM clients; database pools; configuration. | Centralized state management; easy access; lifetime tied to app; no global variables. | Not suitable for per-request state; requires app reference; no type safety by default. | `app = FastAPI()`
`app.state.llm_client = AsyncOpenAI()`

`@app.get("/generate")`
`async def generate(request: Request):`
  `client = request.app.state.llm_client` | 15, 100 | | **Dependency Injection for State** | Using FastAPI dependencies to manage stateful resources. | Database sessions; authenticated users; request-scoped state. | Type-safe; testable; automatic cleanup; clear dependencies. | Can be verbose; requires understanding dependency system; execution order matters. | `async def get_llm_client(request: Request):`
  `return request.app.state.llm_client`

`@app.post("/gen")`
`async def gen(client=Depends(get_llm_client)):` | 15, 20, 101 | | **Request.state in FastAPI** | Per-request state storage in FastAPI. | Request-scoped data; middleware communication; temporary storage. | Request-scoped; automatically cleaned up; accessible from middleware and routes. | Not shared across requests; requires Request object; limited type safety. | `@app.middleware("http")`
`async def add_state(request, call_next):`
  `request.state.start_time = time.time()`
  `response = await call_next(request)`
  `return response` | 15, 102 | | **Global State (Carefully)** | Using module-level variables for shared state. | Configuration; shared clients; read-only data. | Simple; fast access; no passing around. | Not thread-safe without locks; hard to test; can cause race conditions; not suitable for mutable state. | `# config.py (read-only)`
`SETTINGS = load_settings()`

`# clients.py (initialized once)`
`llm_client = None`

`async def get_client():`
  `global llm_client`
  `if llm_client is None:`
    `llm_client = await init_client()` | 103 | --- ## Testing Async Code | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **pytest-asyncio** | pytest plugin enabling async test functions. | Unit testing async code; integration tests; fixture-based testing. | Native pytest integration; familiar syntax; fixture support; multiple event loop scopes. | Requires pytest plugin; event loop management complexity; scope configuration needed. | `import pytest`

`@pytest.mark.asyncio`
`async def test_llm_call():`
  `result = await call_llm("test")`
  `assert result is not None` | 104, 105, 106 | | **Async Fixtures** | Pytest fixtures that use async/await. | Setting up async resources; database connections; mock servers. | Proper async resource management; can await setup; automatic cleanup. | Must use pytest_asyncio.fixture decorator; scope management complex; harder to debug. | `@pytest_asyncio.fixture`
`async def db_session():`
  `session = await create_session()`
  `yield session`
  `await session.close()` | 104, 107, 108 | | **Mock Async Functions** | Using unittest.mock with async functions. | Mocking LLM APIs; isolating tests; testing error scenarios. | Familiar mocking syntax; supports coroutine return values; can verify calls. | Different syntax for async (AsyncMock); must handle awaitable returns; can be confusing. | `from unittest.mock import AsyncMock`

`llm_client.generate = AsyncMock(`
  `return_value="mocked response")`

`result = await llm_client.generate("test")`
`assert result == "mocked response"` | 109, 110 | | **httpx.AsyncClient for API Testing** | Testing FastAPI endpoints with async client. | Integration testing; end-to-end tests; API contract testing. | Realistic request handling; tests actual async behavior; supports streaming. | Slower than unit tests; requires app instance; more setup. | `@pytest.mark.asyncio`
`async def test_endpoint(client):`
  `async with httpx.AsyncClient() as client:`
    `resp = await client.post(url, json=data)`
    `assert resp.status_code == 200` | 30, 31, 111 | | **TestClient (Synchronous)** | FastAPI's synchronous test client for testing async apps. | Simple API tests; quick tests; when async not needed. | Simpler than async; no asyncio needed; faster for simple tests. | Doesn't test async behavior properly; can hide async issues; not suitable for streaming tests. | `from fastapi.testclient import TestClient`

`client = TestClient(app)`
`response = client.get("/health")`
`assert response.status_code == 200` | 15, 112 | | **Event Loop Fixtures** | Controlling event loop behavior in tests. | Testing event loop edge cases; lifecycle testing; scope control. | Full control over event loop; can test lifecycle; supports custom loops. | Complex; usually not needed; can cause issues; requires deep understanding. | `@pytest.fixture(scope="session")`
`def event_loop():`
  `loop = asyncio.new_event_loop()`
  `yield loop`
  `loop.close()` | 104, 113, 114 | | **Timeout in Tests** | Adding timeouts to prevent hung tests. | Preventing infinite waits; CI/CD reliability; catching deadlocks. | Prevents hung tests; faster failure; CI-friendly; catches deadlocks. | May need adjustment; can cause flaky tests; environment-dependent. | `@pytest.mark.asyncio`
`@pytest.mark.timeout(30) # seconds`
`async def test_with_timeout():`
  `result = await long_operation()` | 115 | --- ## Background Tasks and Job Queues | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **FastAPI BackgroundTasks** | Built-in background task execution after response. | Logging; notifications; simple post-processing; fire-and-forget. | No external dependencies; simple API; automatic execution; works with response. | Lost on restart; no persistence; no retry; single-worker only; not for critical tasks. | `@app.post("/process")`
`async def process(bg: BackgroundTasks):`
  `bg.add_task(send_email, user)`
  `return {"status": "done"}` | 15, 24, 116 | | **Celery with FastAPI** | Distributed task queue with async worker support. | Long-running tasks; batch processing; scheduled jobs; distributed processing. | Persistent tasks; retry mechanisms; distributed; monitoring; mature ecosystem. | Complex setup; requires message broker (Redis/RabbitMQ); operational overhead; separate worker processes. | `# tasks.py`
`@celery_app.task`
`def process_batch(items):`
  `# Process items`

`# endpoint`
`task = process_batch.delay(items)`
`return {"task_id": task.id}` | 117, 118, 119 | | **ARQ (Async Redis Queue)** | Async-first job queue built on Redis. | Async-native task processing; scheduled jobs; job chains; cron-like scheduling. | Native async support; simpler than Celery; good for async apps; Redis-based; built-in scheduling. | Requires Redis; smaller community; fewer features than Celery; less mature. | `async def background_task(ctx, user_id):`
  `await process_user(user_id)`

`# Enqueue`
`await redis.enqueue_job(`
  `'background_task', user_id)` | 120, 121 | | **asyncio.create_task for Background** | Creating fire-and-forget tasks in endpoints. | Quick background work; non-critical tasks; in-process jobs. | No dependencies; simple; immediate execution; no setup. | Lost on restart; no error tracking; can cause memory leaks; hard to monitor. | `@app.post("/trigger")`
`async def trigger():`
  `asyncio.create_task(background_work())`
  `return {"status": "started"}`

`# Warning: No error handling!` | 1, 122 | | **Scheduled Tasks (APScheduler)** | Scheduled job execution with async support. | Periodic tasks; cron jobs; scheduled LLM jobs; maintenance tasks. | Flexible scheduling; async support; persistent jobs; timezone support. | In-process only; requires event loop management; can miss schedules; needs proper shutdown. | `from apscheduler.schedulers.asyncio import AsyncIOScheduler`

`scheduler = AsyncIOScheduler()`
`scheduler.add_job(`
  `async_func, 'interval', hours=1)`
`scheduler.start()` | 123, 124 | | **Task Result Storage** | Storing task results for later retrieval. | Long-running LLM generations; progress tracking; result polling. | Enables polling; persistent results; supports async retrieval. | Requires storage; cleanup needed; state management; polling overhead. | `# Store task`
`task_id = str(uuid.uuid4())`
`await redis.set(f"task:{task_id}", "PENDING")`

`# Check status`
`status = await redis.get(f"task:{task_id}")` | 125, 126 | --- ## Performance Optimization | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **Connection Reuse** | Reusing HTTP and database connections across requests. | LLM API calls; database queries; reducing latency. | Lower latency; fewer resources; faster requests; reduced overhead. | Must manage lifecycles; can leak; requires session management; limit configuration. | `# Singleton pattern`
`class AsyncClientManager:`
  `_client = None`

  `@classmethod`
  `async def get_client(cls):`
    `if cls._client is None:`
      `cls._client = httpx.AsyncClient()`
    `return cls._client` | 45, 46, 127 | | **Batch Processing** | Grouping multiple items for processing together. | Bulk LLM requests; batch database inserts; reducing API calls. | Fewer API calls; better throughput; reduced costs; efficient resource use. | Increased latency for individual items; more complex code; error handling complexity; memory usage. | `async def batch_process(items, batch_size=10):`
  `for i in range(0, len(items), batch_size):`
    `batch = items[i:i+batch_size]`
    `await process_batch(batch)` | 8, 9, 128 | | **Response Caching** | Caching LLM or API responses to avoid redundant calls. | Reducing API costs; improving latency; caching embeddings. | Lower costs; faster responses; reduced load; improved UX. | Cache invalidation complexity; memory/storage requirements; stale data risk. | `from functools import lru_cache`

`async def get_embedding(text):`
  `key = f"emb:{hash(text)}"`
  `cached = await redis.get(key)`
  `if cached:`
    `return cached`
  `result = await llm.embed(text)`
  `await redis.set(key, result, ex=3600)` | 129, 130 | | **Lazy Loading** | Loading data only when needed using async generators. | Large datasets; paginated results; memory efficiency. | Memory efficient; faster initial response; better for large datasets. | More complex code; need to handle iteration; error handling across iterations. | `async def lazy_load_items():`
  `page = 1`
  `while True:`
    `items = await fetch_page(page)`
    `if not items: break`
    `for item in items:`
      `yield item`
    `page += 1` | 78, 84, 131 | | **Pre-warming Connections** | Establishing connections before they're needed. | Reducing first-request latency; preparing resources; startup optimization. | Lower initial latency; predictable performance; better for critical paths. | Resource usage even when idle; complex lifecycle; can waste connections. | `@asynccontextmanager`
`async def lifespan(app):`
  `# Pre-warm`
  `await db_pool.connect()`
  `await http_client.get("/health")`
  `yield`
  `# Cleanup` | 22, 132 | | **run_in_executor for CPU-Bound** | Offloading blocking/CPU-intensive work to thread/process pools. | Heavy computations; blocking libraries; CPU-intensive tasks. | Prevents event loop blocking; can use existing sync code; maintains async interface. | Thread pool limited by GIL for CPU; overhead of thread creation; needs proper executor management. | `loop = asyncio.get_event_loop()`
`result = await loop.run_in_executor(`
  `None, # Default ThreadPoolExecutor`
  `cpu_intensive_func, arg1, arg2)` | 1, 133, 134 | | **ProcessPoolExecutor for CPU** | Using process pools for CPU-bound parallelism. | ML inference; heavy computation; true parallelism. | True parallelism; bypasses GIL; scales with CPU cores. | High overhead; serialization costs; complex communication; resource intensive. | `from concurrent.futures import ProcessPoolExecutor`

`executor = ProcessPoolExecutor(max_workers=4)`
`result = await loop.run_in_executor(`
  `executor, heavy_compute, data)` | 133, 135, 136 | | **uvloop** | Drop-in replacement for asyncio event loop with better performance. | Production deployments; high-performance requirements; reducing latency. | 2-4x faster than default loop; drop-in replacement; benchmarked extensively. | Linux/macOS only; different debugging behavior; external dependency; potential compatibility issues. | `import uvloop`

`# Set as default`
`uvloop.install()`

`# Or explicit`
`loop = uvloop.new_event_loop()`
`asyncio.set_event_loop(loop)` | 137, 138 | --- ## Common Pitfalls and Solutions | Pitfall/Problem | Description | Why It Happens | Solution | Example | Reference | |-----------------|-------------|----------------|----------|---------|-----------| | **Blocking the Event Loop** | Calling synchronous blocking code (time.sleep, sync I/O) in async functions. | Blocks entire event loop; prevents all other tasks from running. | Use async alternatives (asyncio.sleep, aiofiles) or run_in_executor for blocking code. | `# BAD`
`async def bad():`
  `time.sleep(1) # Blocks!`

`# GOOD`
`async def good():`
  `await asyncio.sleep(1)` | 1, 139, 140 | | **Forgetting await** | Calling async function without await returns coroutine object. | Python doesn't execute async functions until awaited; common mistake. | Always await async calls or schedule them as tasks. | `# BAD`
`result = async_func() # coroutine!`

`# GOOD`
`result = await async_func()` | 1, 141 | | **Creating Too Many Tasks** | Spawning thousands of concurrent tasks without limits. | No automatic concurrency control; can overwhelm APIs/databases. | Use Semaphore, Queue, or rate limiting to control concurrency. | `# BAD`
`tasks = [process(i) for i in range(10000)]`

`# GOOD`
`sem = asyncio.Semaphore(50)`
`async def limited(i):`
  `async with sem:`
    `await process(i)` | 59, 60, 142 | | **Not Handling CancelledError** | Failing to catch CancelledError during cleanup. | Task cancellation is normal; need to cleanup resources. | Catch CancelledError, cleanup, then re-raise. | `async def task():`
  `try:`
    `await work()`
  `except asyncio.CancelledError:`
    `await cleanup()`
    `raise # Important!` | 1, 88, 143 | | **Mixing Sync and Async Code** | Calling async code from sync or vice versa incorrectly. | Different execution models; event loop required for async. | Use asyncio.run() for sync→async; run_in_executor for async→sync. | `# Sync calling async`
`result = asyncio.run(async_func())`

`# Async calling sync`
`result = await loop.run_in_executor(None, sync_func)` | 1, 133, 144 | | **Event Loop Already Running** | Trying to run asyncio.run() when loop already running. | Jupyter notebooks, nested event loops. | Use await directly or nest-asyncio library. | `# In Jupyter`
`import nest_asyncio`
`nest_asyncio.apply()`

`# Then asyncio.run works` | 145, 146 | | **Connection Pool Exhaustion** | Running out of database or HTTP connections. | Too many concurrent operations; improper connection management. | Configure pool size; use connection limits; proper cleanup. | `engine = create_async_engine(`
  `url,`
  `pool_size=20,`
  `max_overflow=10,`
  `pool_pre_ping=True)` | 32, 40, 147 | | **Not Closing Resources** | Forgetting to close connections, sessions, files. | Resource leaks; exhausted file descriptors; memory leaks. | Use async context managers; ensure cleanup in finally. | `# GOOD`
`async with aiohttp.ClientSession() as session:`
  `await session.get(url)`
`# Automatically closed` | 45, 46, 148 | | **Shared Mutable State** | Multiple coroutines accessing shared mutable state without synchronization. | Race conditions; data corruption; unpredictable behavior. | Use locks, avoid shared state, or use immutable data. | `lock = asyncio.Lock()`
`async def update():`
  `async with lock:`
    `shared_state.modify()` | 1, 63, 149 | | **Improper Timeout Handling** | Not setting timeouts or handling timeout errors. | Hung requests; resource exhaustion; poor UX. | Always set timeouts; handle TimeoutError explicitly. | `try:`
  `async with asyncio.timeout(30):`
    `await llm_call()`
`except asyncio.TimeoutError:`
  `return default_response` | 1, 86, 150 | | **Memory Leaks from Tasks** | Creating tasks without tracking or awaiting them. | Tasks hold references; prevent garbage collection. | Always await tasks or store references for cleanup. | `# Track tasks`
`background_tasks = set()`

`task = asyncio.create_task(work())`
`background_tasks.add(task)`
`task.add_done_callback(background_tasks.discard)` | 1, 151, 152 | | **Deadlocks with Locks** | Improper lock acquisition order or holding locks across awaits. | Circular wait conditions; holding locks during I/O. | Minimize lock scope; consistent lock ordering; avoid await while holding lock. | `# BAD`
`async with lock:`
  `await db_call() # Lock held during I/O!`

`# GOOD`
`async with lock:`
  `data = prepare_data()`
`await db_call(data)` | 63, 153 | --- ## Production Deployment Considerations | Technique/Approach | Description | Main Usages | Advantages | Considerations | Example | Reference | |-------------------|-------------|-------------|------------|----------------|---------|-----------| | **Graceful Shutdown** | Handling SIGTERM/SIGINT signals to cleanly shutdown. | Production deployments; Kubernetes; preventing data loss. | Clean resource cleanup; completes in-flight requests; prevents corruption. | Must handle signals; needs careful testing; timeout considerations. | `import signal`

`def handle_shutdown(sig, frame):`
  `loop = asyncio.get_event_loop()`
  `loop.create_task(shutdown())`

`signal.signal(signal.SIGTERM, handle_shutdown)`
`signal.signal(signal.SIGINT, handle_shutdown)` | 154, 155, 156 | | **Health Checks** | Endpoints for monitoring application health. | Load balancers; Kubernetes readiness/liveness; monitoring. | Automatic failover; proper routing; deployment safety. | Must be lightweight; shouldn't block; need meaningful checks. | `@app.get("/health")`
`async def health():`
  `return {"status": "healthy"}`

`@app.get("/ready")`
`async def ready():`
  `await db.ping()`
  `return {"status": "ready"}` | 15, 157 | | **Structured Logging** | Using structured logs with async-safe logging. | Production debugging; monitoring; tracing; analysis. | Machine-parseable; context-rich; correlatable; async-safe. | Performance impact; storage costs; PII concerns; configuration complexity. | `import structlog`

`log = structlog.get_logger()`
`await log.ainfo("llm_call",`
  `request_id=req_id,`
  `duration=elapsed)` | 158, 159 | | **Distributed Tracing** | Tracking requests across multiple services. | Microservices; debugging; performance analysis; LLM chains. | End-to-end visibility; bottleneck identification; context propagation. | Overhead; complex setup; storage requirements; sampling needed at scale. | `from opentelemetry import trace`

`tracer = trace.get_tracer(__name__)`
`with tracer.start_as_current_span("llm_call"):`
  `result = await llm.generate(prompt)` | 160, 161 | | **Metrics Collection** | Gathering async application metrics. | Monitoring; alerting; capacity planning; optimization. | Visibility into performance; proactive alerting; trend analysis. | Overhead; metric explosion; storage costs; careful selection needed. | `from prometheus_client import Counter`

`llm_calls = Counter('llm_calls_total')`

`async def call_llm():`
  `llm_calls.inc()`
  `return await client.generate()` | 162, 163 | | **Rate Limiting** | Controlling request rates to protect APIs and services. | API protection; quota management; preventing abuse. | Prevents overload; fair resource sharing; cost control. | Configuration complexity; affects UX; distributed systems challenges. | `from slowapi import Limiter`

`limiter = Limiter(key_func=get_remote_address)`

`@app.post("/generate")`
`@limiter.limit("10/minute")`
`async def generate():` | 164, 165 | | **Circuit Breaker (Production)** | Production-grade circuit breaker implementation. | Preventing cascade failures; fast failure; service protection. | Prevents wasted resources; automatic recovery; protects dependencies. | State management; distributed coordination; configuration tuning. | `# Use library like pybreaker`
`breaker = CircuitBreaker(`
  `fail_max=5,`
  `timeout_duration=60)`

`@breaker`
`async def call_api():` | 57, 58, 166 | | **Async Profiling** | Profiling async application performance. | Performance optimization; bottleneck identification; debugging. | Identifies slow operations; helps optimize; async-aware profiling. | Overhead; complex analysis; requires specialized tools. | `import aiomonitor`

`async with aiomonitor.start_monitor(`
  `loop, port=50101):`
  `await run_app()`

`# Can connect and inspect running app` | 167, 168, 169 | | **Memory Leak Detection** | Identifying and preventing async memory leaks. | Production stability; resource optimization; debugging. | Prevents crashes; stable memory usage; better performance. | Requires monitoring; tools needed; complex debugging. | `import tracemalloc`

`tracemalloc.start()`
`# Run app`
`snapshot = tracemalloc.take_snapshot()`
`top_stats = snapshot.statistics('lineno')` | 170, 171 | | **Kubernetes Deployment** | Deploying async apps in Kubernetes. | Scaling; high availability; cloud-native deployment. | Auto-scaling; self-healing; declarative configuration; rolling updates. | Complex configuration; understanding needed; resource limits critical. | `# Deployment with graceful shutdown`
`terminationGracePeriodSeconds: 30`

`# Probes`
`livenessProbe:`
  `httpGet:`
    `path: /health`
`readinessProbe:`
  `httpGet:`
    `path: /ready` | 154, 172 | | **Worker Process Management** | Running multiple async workers (uvicorn workers). | Scaling; utilizing multiple cores; high throughput. | Better CPU utilization; higher throughput; fault isolation. | More memory; complexity in shared state; process communication. | `# Uvicorn with multiple workers`
`uvicorn main:app \`
  `--workers 4 \`
  `--worker-class uvicorn.workers.UvicornWorker` | 15, 173 | | **Configuration Management** | Managing async app configuration. | Environment-specific config; secrets; feature flags. | Environment isolation; easy updates; security; version control. | Validation needed; secret management; reload complexity. | `from pydantic_settings import BaseSettings`

`class Settings(BaseSettings):`
  `database_url: str`
  `llm_api_key: str`

  `class Config:`
    `env_file = ".env"` | 15, 174 | --- ## References ### Official Documentation & Core Concepts 1. **Python asyncio Documentation** - https://docs.python.org/3/library/asyncio.html 2. **Real Python: Async IO in Python** - https://realpython.com/async-io-python/ 3. **FastAPI Async Documentation** - https://fastapi.tiangolo.com/async/ 4. **asyncio Event Loop** - https://docs.python.org/3/library/asyncio-eventloop.html 5. **Understanding Python's asyncio Event Loop** - https://medium.com/@virtualik/python-asyncio-event-loop-blocking-explained-with-code-examples-0b2bba801426 6. **Coroutines and Tasks** - https://docs.python.org/3/library/asyncio-task.html 7. **Unite.AI: Asynchronous LLM API Calls Guide** - https://www.unite.ai/asynchronous-llm-api-calls-in-python-a-comprehensive-guide/ ### FastAPI & Async Patterns 8. **Python Asyncio for LLM Concurrency** - https://www.newline.co/@zaoyang/python-asyncio-for-llm-concurrency-best-practices--bc079176 9. **Mastering asyncio.gather and as_completed for LLM Processing** - https://python.useinstructor.com/blog/2023/11/13/learn-async/ 10. **Building Asynchronous LLM Applications** - https://diverger.medium.com/building-asynchronous-llm-applications-in-python-f775da7b15d1 11. **asyncio.wait() Documentation** - https://docs.python.org/3/library/asyncio-task.html#asyncio.wait 12. **Limiting Concurrency with Python Asyncio** - https://death.andgravity.com/limit-concurrency 13. **Python 3.11+ TaskGroup** - https://docs.python.org/3/library/asyncio-task.html#asyncio.TaskGroup 14. **Async APIs with FastAPI: Patterns and Best Practices** - https://shiladityamajumder.medium.com/async-apis-with-fastapi-patterns-pitfalls-best-practices-2d72b2b66f25 ### FastAPI-Specific 15. **FastAPI Documentation** - https://fastapi.tiangolo.com/ 16. **Asynchronous Programming with FastAPI** - https://dev.to/dhrumitdk/asynchronous-programming-with-fastapi-building-efficient-apis-nj1 17. **Dead Simple: When to Use Async in FastAPI** - https://hughesadam87.medium.com/dead-simple-when-to-use-async-in-fastapi-0e3259acea6f 18. **FastAPI Performance: Sync vs Async** - https://thedkpatel.medium.com/fastapi-performance-showdown-sync-vs-async-which-is-better-77188d5b1e3a 19. **What Does Async Actually Do in FastAPI?** - https://stackoverflow.com/questions/75463993/what-does-async-actually-do-in-fastapi 20. **Dependency Injection in FastAPI** - https://fastapi.tiangolo.com/tutorial/dependencies/ 21. **FastAPI Advanced Dependencies** - https://fastapi.tiangolo.com/advanced/advanced-dependencies/ 22. **FastAPI Lifespan Events** - https://fastapi.tiangolo.com/advanced/events/ 23. **Understanding FastAPI Lifespan Events** - https://dev.turmansolutions.ai/2025/09/27/understanding-fastapis-lifespan-events-proper-initialization-and-shutdown/ 24. **FastAPI Background Tasks** - https://fastapi.tiangolo.com/tutorial/background-tasks/ 25. **How to Implement Background Tasks in FastAPI** - https://oneuptime.com/blog/post/2026-02-02-fastapi-background-tasks/view 26. **Streaming Responses in FastAPI** - https://hassaanbinaslam.github.io/posts/2025-01-19-streaming-responses-fastapi.html 27. **Scalable Streaming of OpenAI Responses with FastAPI** - https://medium.com/@mayvic/scalable-streaming-of-openai-model-responses-with-fastapi-and-asyncio-714744b13dd 28. **FastAPI Middleware** - https://fastapi.tiangolo.com/tutorial/middleware/ 29. **Building High-Performance APIs with FastAPI** - https://dasroot.net/posts/2026/01/building-high-performance-apis-fastapi-async-python/ ### Testing Async Code 30. **httpx AsyncClient** - https://www.python-httpx.org/async/ 31. **Getting Started with HTTPX** - https://betterstack.com/community/guides/scaling-python/httpx-explained/ 104. **pytest-asyncio Documentation** - https://pytest-asyncio.readthedocs.io/ 105. **pytest-asyncio Concepts** - https://pytest-asyncio.readthedocs.io/en/latest/concepts.html 106. **How to Test Async Functions with pytest** - https://stackoverflow.com/questions/70015634/how-to-test-async-function-using-pytest 107. **Async Fixtures with pytest** - https://stackoverflow.com/questions/49936724/async-fixtures-with-pytest 108. **Advanced Asyncio Testing** - https://stefan.sofa-rockers.org/2016/03/10/advanced-asyncio-testing/ 109. **Python unittest.mock AsyncMock** - https://docs.python.org/3/library/unittest.mock.html#unittest.mock.AsyncMock 110. **Async Testing with Pytest** - https://medium.com/@connect.hashblock/async-testing-with-pytest-mastering-pytest-asyncio-and-event-loops-for-fastapi-and-beyond-37c613f1cfa3 111. **Testing FastAPI Applications** - https://fastapi.tiangolo.com/tutorial/testing/ 112. **FastAPI TestClient** - https://fastapi.tiangolo.com/reference/testclient/ 113. **pytest-asyncio Event Loop Fixtures** - https://pytest-asyncio.readthedocs.io/en/stable/reference/fixtures/ 114. **How to Use event_loop in pytest-asyncio** - https://www.lambdatest.com/automation-testing-advisor/python/pytest-asyncio_python-event_loop 115. **pytest-timeout** - https://pypi.org/project/pytest-timeout/ ### Database Operations 32. **SQLAlchemy Async Documentation** - https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html 33. **asyncpg Documentation** - https://magicstack.github.io/asyncpg/current/ 34. **asyncpg Usage Guide** - https://magicstack.github.io/asyncpg/current/usage.html 35. **SQLAlchemy Async ORM** - https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html 36. **Asynchronous Database Sessions in FastAPI** - https://dev.to/akarshan/asynchronous-database-sessions-in-fastapi-with-sqlalchemy-1o7e 37. **Motor (MongoDB Async Driver)** - https://motor.readthedocs.io/ 38. **MongoDB Async Python** - https://www.mongodb.com/docs/drivers/motor/ 39. **FastAPI with MongoDB** - https://www.mongodb.com/developer/languages/python/python-quickstart-fastapi/ 40. **Connection Pooling in SQLAlchemy** - https://docs.sqlalchemy.org/en/20/core/pooling.html 41. **Building High-Performance Async APIs with SQLAlchemy** - https://leapcell.io/blog/building-high-performance-async-apis-with-fastapi-sqlalchemy-2-0-and-asyncpg 42. **SQLAlchemy and asyncpg Connection Pooling Best Practices** - https://www.codingeasypeasy.com/blog/sqlalchemy-and-asyncpg-connection-pooling-best-practices-for-performance-and-reliability 43. **Async Transactions in SQLAlchemy** - https://docs.sqlalchemy.org/en/20/orm/session_transaction.html ### HTTP Clients 45. **aiohttp Client Documentation** - https://docs.aiohttp.org/en/stable/client.html 46. **aiohttp Client Quickstart** - https://docs.aiohttp.org/en/stable/client_quickstart.html 47. **aiohttp Client Reference** - https://docs.aiohttp.org/en/stable/client_reference.html 48. **httpx Documentation** - https://www.python-httpx.org/ 49. **httpx AsyncClient** - https://www.python-httpx.org/async/ 50. **httpx Resource Limits** - https://www.python-httpx.org/advanced/resource-limits/ 51. **httpx Timeouts** - https://www.python-httpx.org/advanced/timeouts/ 52. **aiohttp Timeout Configuration** - https://docs.aiohttp.org/en/stable/client_reference.html#aiohttp.ClientTimeout 53. **The aiohttp Request Lifecycle** - https://docs.aiohttp.org/en/stable/http_request_lifecycle.html ### Error Handling & Resilience 54. **How to Handle Rate Limits (OpenAI)** - https://cookbook.openai.com/examples/how_to_handle_rate_limits 55. **Retry Logic with Tenacity** - https://python.useinstructor.com/concepts/retrying/ 56. **How to Implement Retry Logic for LLM API Failures** - https://markaicode.com/llm-api-retry-logic-implementation/ 57. **Retries, Fallbacks, and Circuit Breakers in LLM Apps** - https://portkey.ai/blog/retries-fallbacks-and-circuit-breakers-in-llm-apps/ 58. **API Timeout Handling for LLM Applications** - https://markaicode.com/api-timeout-handling-llm-applications/ 85. **Python Exception Handling** - https://docs.python.org/3/tutorial/errors.html 86. **asyncio Timeout Context Manager** - https://docs.python.org/3/library/asyncio-task.html#asyncio.timeout 87. **Async with Timeout Best Practices** - https://runebook.dev/en/docs/python/library/asyncio-task/asyncio.timeout 88. **Asyncio CancelledError** - https://docs.python.org/3/library/asyncio-exceptions.html#asyncio.CancelledError 89. **Handling Task Cancellation** - https://superfastpython.com/asyncio-cancelled-error/ 90. **Tenacity Documentation** - https://tenacity.readthedocs.io/ 91. **backoff Library** - https://github.com/litl/backoff 92. **backoff PyPI** - https://pypi.org/project/backoff/ 93. **PEP 654 Exception Groups** - https://peps.python.org/pep-0654/ 94. **Building Bulletproof LLM Applications** - https://medium.com/google-cloud/building-bulletproof-llm-applications-a-guide-to-applying-sre-best-practices-1564b72fd22e ### Concurrency Control 59. **asyncio Semaphore** - https://docs.python.org/3/library/asyncio-sync.html#asyncio.Semaphore 60. **Limit Concurrency with Semaphore** - https://rednafi.com/python/limit-concurrency-with-semaphore/ 61. **Mastering Asyncio Semaphores** - https://medium.com/@mr.sourav.raj/mastering-asyncio-semaphores-in-python-a-complete-guide-to-concurrency-control-6b4dd940e10e 62. **asyncio BoundedSemaphore** - https://docs.python.org/3/library/asyncio-sync.html#asyncio.BoundedSemaphore 63. **asyncio Locks** - https://docs.python.org/3/library/asyncio-sync.html#asyncio.Lock 64. **Python Asyncio Locks** - https://runebook.dev/en/docs/python/library/asyncio-sync/lock 65. **asyncio Queue** - https://docs.python.org/3/library/asyncio-queue.html 66. **Unleashing the Power of asyncio Queue** - https://towardsdatascience.com/unleashing-the-power-of-python-asyncios-queue-f76e3188f1c4/ 67. **Mastering Asynchronous Queues in Python** - https://medium.com/@caring_smitten_gerbil_914/mastering-asynchronous-queues-in-python-concurrency-made-easy-with-asyncio-878566ef9d7d 68. **Asynchronous Worker Pool Pattern** - https://santhalakshminarayana.github.io/blog/concurrency-patterns-python 69. **The Async Worker Pool** - https://shav.dev/blog/pool-of-workers 70. **Rate Limiting Patterns** - https://github.com/snok/self-limiters 71. **Semaphores in Python Async Programming** - https://www.soumendrak.com/blog/semaphores-python-async-programming/ ### Streaming & Real-Time 72. **FastAPI Server-Sent Events for LLM Streaming** - https://medium.com/@2nick2patel2/fastapi-server-sent-events-for-llm-streaming-smooth-tokens-low-latency-1b211c94cff5 73. **Server-Sent Events with Python FastAPI** - https://medium.com/@nandagopal05/server-sent-events-with-python-fastapi-f1960e0c8e4b 74. **How Streaming LLM APIs Work** - https://til.simonwillison.net/llms/streaming-llm-apis 75. **WebSockets in FastAPI** - https://fastapi.tiangolo.com/advanced/websockets/ 76. **Streaming AI Responses: WebSockets vs SSE vs gRPC** - https://medium.com/@pranavprakash4777/streaming-ai-responses-with-websockets-sse-and-grpc-which-one-wins-a481cab403d3 77. **Python WebSocket Programming** - https://websockets.readthedocs.io/ 78. **PEP 525: Asynchronous Generators** - https://peps.python.org/pep-0525/ 79. **Asynchronous Generators in Python** - https://superfastpython.com/asynchronous-generators-in-python/ 80. **Python Async Iterators** - https://realpython.com/python-async-iterators/ 81. **Async Context Managers** - https://docs.python.org/3/reference/datamodel.html#async-context-managers 82. **Python Asyncio Part 3: Async Context Managers** - https://bbc.github.io/cloudfit-public-docs/asyncio/asyncio-part-3.html 83. **Chunked Transfer Encoding** - https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Transfer-Encoding 84. **The Async Iterator Part 2: Streaming Data** - https://dev.to/aaron_rose_0787cc8b4775a0/the-async-iterator-part-2-streaming-data-and-real-world-patterns-kdj ### Context & State Management 97. **contextvars Documentation** - https://docs.python.org/3/library/contextvars.html 98. **Python's contextvars: Better State Management** - https://elshad-karimov.medium.com/pythons-contextvars-a-better-way-to-manage-state-in-async-code-47715a126910 99. **Asyncio Context Variables** - https://superfastpython.com/asyncio-context-variables/ 100. **FastAPI app.state** - https://fastapi.tiangolo.com/advanced/using-request-directly/#using-the-request-directly 101. **Using FastAPI Like a Pro with Dependency Injection** - https://medium.com/@hieutrantrung.it/using-fastapi-like-a-pro-with-singleton-and-dependency-injection-patterns-28de0a833a52 102. **Request.state in FastAPI** - https://fastapi.tiangolo.com/advanced/using-request-directly/ 103. **PEP 567: Context Variables** - https://peps.python.org/pep-0567/ ### Background Tasks & Job Queues 116. **How to Build Background Task Processing in FastAPI** - https://oneuptime.com/blog/post/2026-01-25-background-task-processing-fastapi/view 117. **Asynchronous Tasks with FastAPI and Celery** - https://testdriven.io/blog/fastapi-and-celery/ 118. **The Definitive Guide to Celery and FastAPI** - https://testdriven.io/courses/fastapi-celery/intro/ 119. **FastAPI Celery Integration** - https://www.compilenrun.com/docs/framework/fastapi/fastapi-background-tasks/fastapi-celery-integration/ 120. **Managing Background Tasks in FastAPI: ARQ vs Built-in** - https://davidmuraya.com/blog/fastapi-background-tasks-arq-vs-built-in/ 121. **ARQ (Async Redis Queue)** - https://arq-docs.helpmanual.io/ 122. **Managing Background Tasks in FastAPI** - https://leapcell.io/blog/managing-background-tasks-and-long-running-operations-in-fastapi 123. **APScheduler Async** - https://apscheduler.readthedocs.io/en/stable/ 124. **Mastering Background Tasks and Scheduling in FastAPI** - https://procodebase.com/article/mastering-background-tasks-and-scheduling-in-fastapi ### Performance & Optimization 127. **8 httpx + asyncio Patterns for Safer, Faster Clients** - https://medium.com/@sparknp1/8-httpx-asyncio-patterns-for-safer-faster-clients-f27bc82e93e6 128. **Async & Batching in RAG** - https://apxml.com/courses/optimizing-rag-for-production/chapter-4-end-to-end-rag-performance/async-processing-batching-rag 129. **Python Caching with Redis** - https://realpython.com/python-redis/ 130. **FastAPI Caching** - https://fastapi-cache.readthedocs.io/ 131. **Memory-Efficient Python Generators** - https://realpython.com/introduction-to-python-generators/ 132. **FastAPI Lifespan for Resource Pre-warming** - https://fastapi.tiangolo.com/advanced/events/ 133. **Developing with asyncio** - https://docs.python.org/3/library/asyncio-dev.html 134. **Python Async/Sync: Understanding Blocking** - https://dzone.com/articles/python-async-vs-sync-blocking 135. **Python ProcessPoolExecutor** - https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor 136. **How to Run Blocking Functions in Event Loop** - https://codilime.com/blog/how-fit-triangles-into-squares-run-blocking-functions-event-loop/ 137. **uvloop** - https://github.com/MagicStack/uvloop 138. **Python in Backend 2025: Asyncio and FastAPI** - https://www.nucamp.co/blog/coding-bootcamp-backend-with-python-2025-python-in-the-backend-in-2025-leveraging-asyncio-and-fastapi-for-highperformance-systems ### Common Pitfalls & Best Practices 139. **Python Asyncio Event Loop Blocking** - https://medium.com/@virtualik/python-asyncio-event-loop-blocking-explained-with-code-examples-0b2bba801426 140. **Python 3.13 AsyncIO Errors** - https://markaicode.com/python-3-13-asyncio-errors-event-loop-blocking/ 141. **The Await Pitfall** - https://runebook.dev/en/docs/python/library/asyncio-dev 142. **Controlling Concurrency in Python** - https://dev.to/ctrix/controlling-concurrency-in-python-semaphores-and-pool-workers-56d7 143. **Handling CancelledError** - https://superfastpython.com/asyncio-cancelled-error/ 144. **Mixing Sync and Async Code** - https://dzone.com/articles/python-async-vs-sync-blocking 145. **nest-asyncio** - https://pypi.org/project/nest-asyncio/ 146. **Understanding Python Async with FastAPI** - https://ruarfff.com/posts/understanding-python-async 147. **SQLAlchemy Connection Pooling** - https://docs.sqlalchemy.org/en/20/core/pooling.html 148. **How to Properly Close aiohttp ClientSession** - https://www.pythontutorials.net/blog/how-to-close-aiohttp-clientsession/ 149. **Python Asyncio Locks** - https://runebook.dev/en/docs/python/library/asyncio-sync/lock 150. **asyncio Timeout** - https://docs.python.org/3/library/asyncio-task.html#asyncio.timeout 151. **asyncio Memory Leaks** - https://stackoverflow.com/questions/58616723/debug-asyncio-memory-leaks 152. **Python Asyncio Memory Management** - https://stackoverflow.com/questions/68448548/python-asyncio-memory-leak-or ### Production Deployment 154. **Graceful Shutdowns with asyncio** - https://roguelynn.com/words/asyncio-graceful-shutdowns/ 155. **How to Gracefully Terminate Asyncio Script** - https://stackoverflow.com/questions/48562893/how-to-gracefully-terminate-an-asyncio-script-with-ctrl-c 156. **Signal Handling with Async Multiprocesses** - https://medium.com/@cziegler_99189/gracefully-shutting-down-async-multiprocesses-in-python-2223be384510 157. **FastAPI Health Checks** - https://fastapi.tiangolo.com/tutorial/middleware/ 158. **structlog Documentation** - https://www.structlog.org/ 159. **Signal Handling in Python** - https://johal.in/signal-handling-in-python-custom-handlers-for-graceful-shutdowns/ 160. **OpenTelemetry Python** - https://opentelemetry.io/docs/instrumentation/python/ 161. **Distributed Tracing with FastAPI** - https://opentelemetry.io/docs/instrumentation/python/automatic/ 162. **Prometheus Python Client** - https://github.com/prometheus/client_python 163. **FastAPI Observability** - https://fastapi.tiangolo.com/advanced/middleware/ 164. **slowapi (Rate Limiting for FastAPI)** - https://github.com/laurentS/slowapi 165. **Rate Limiting Patterns** - https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/rate-limiting.html 166. **pybreaker (Circuit Breaker)** - https://github.com/danielfm/pybreaker 167. **aiomonitor-ng** - https://blog.lablup.com/en/posts/2022/11/28/aiomonitor-ng/ 168. **How to Profile Asyncio Programs** - https://superfastpython.com/profile-asyncio-programs/ 169. **aiodebug** - https://pypi.org/project/aiodebug/ 170. **Python Memory Profiling** - https://quantlane.com/blog/python-memory-profiling/ 171. **Asyncio Memory Leak Debugging** - https://stackoverflow.com/questions/58616723/debug-asyncio-memory-leaks 172. **Kubernetes Deployment Best Practices** - https://kubernetes.io/docs/concepts/workloads/controllers/deployment/ 173. **Uvicorn Workers** - https://www.uvicorn.org/deployment/ 174. **Pydantic Settings** - https://docs.pydantic.dev/latest/concepts/pydantic_settings/ ### Additional Resources 44. **Async Database Batching** - https://apxml.com/courses/optimizing-rag-for-production/chapter-4-end-to-end-rag-performance/async-processing-batching-rag 95. **Dead Letter Queue Pattern** - https://aws.amazon.com/what-is/dead-letter-queue/ 96. **Handling Background Tasks and Long-Running Jobs in FastAPI** - https://python.plainenglish.io/handling-background-tasks-and-long-running-jobs-in-fastapi-the-complete-guide-b197d38145d7 125. **Task Result Storage Patterns** - https://testdriven.io/courses/fastapi-celery/ 126. **FastAPI Task Status Tracking** - https://www.compilenrun.com/docs/framework/fastapi/fastapi-background-tasks/ 131. **Memory-Efficient Processing with Generators** - https://realpython.com/introduction-to-python-generators/ 132. **Pre-warming Resources in FastAPI** - https://fastapi.tiangolo.com/advanced/events/ 153. **Avoiding Deadlocks in Async Code** - https://runebook.dev/en/docs/python/library/asyncio-sync/lock

On This Page

On This Page