### Install Performance Enhancements for Uvicorn

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Installs `uvloop` and `httptools` to improve the performance of Uvicorn, FastAPI's ASGI server. These libraries provide a faster event loop and HTTP parser than the defaults. Uvicorn automatically utilizes them if installed.

```bash
pip install uvloop httptools
```

--------------------------------

### Install Performance Dependencies for FastAPI

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Install uvloop and httptools to improve the performance of the asyncio event loop and HTTP parsing. These dependencies are automatically utilized by Uvicorn when present in the environment.

```bash
# Install performance dependencies
pip install uvloop httptools

# For cross-platform projects (uvloop doesn't work on Windows)
# Add to requirements.txt with environment marker:
# uvloop; sys_platform != 'win32'
# httptools
```

--------------------------------

### FastAPI Dependency Execution in Threads

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Demonstrates how non-async functions used as FastAPI dependencies are executed in a separate thread. Includes an example of monitoring thread usage and shows how making the dependency function async prevents it from running in a thread.

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from httpx import AsyncClient
from fastapi import FastAPI, Request, Depends


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, AsyncClient]]:
    async with AsyncClient() as client:
        yield {"client": client}


app = FastAPI(lifespan=lifespan)


def http_client(request: Request) -> AsyncClient:
    return request.state.client


@app.get("/")
async def read_root(client: AsyncClient = Depends(http_client)):
    return await client.get("/")
```

```python
# ...

async def http_client(request: Request) -> AsyncClient:
    return request.state.client

# ...
```

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import anyio
from anyio.to_thread import current_default_thread_limiter
from httpx import AsyncClient
from fastapi import FastAPI, Request, Depends


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, AsyncClient]]:
    async with AsyncClient() as client:
        yield {"client": client}


app = FastAPI(lifespan=lifespan)


# Change this function to be async, and rerun this application.
def http_client(request: Request) -> AsyncClient:
    return request.state.client


@app.get("/")
async def read_root(client: AsyncClient = Depends(http_client)): ...


async def monitor_thread_limiter():
    limiter = current_default_thread_limiter()
    threads_in_use = limiter.borrowed_tokens
    while True:
        if threads_in_use != limiter.borrowed_tokens:
            print(f"Threads in use: {limiter.borrowed_tokens}")
            threads_in_use = limiter.borrowed_tokens
        await anyio.sleep(0)


if __name__ == "__main__":
    import uvicorn

    config = uvicorn.Config(app="main:app")
    server = uvicorn.Server(config)

    async def main():
        async with anyio.create_task_group() as tg:
            tg.start_soon(monitor_thread_limiter)
            await server.serve()

    anyio.run(main)
```

--------------------------------

### Configure Async Tests with AnyIO

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Demonstrates how to mark asynchronous test functions using pytest.mark.anyio and how to configure the specific backend (asyncio or trio) using a pytest fixture.

```python
import pytest

@pytest.mark.anyio
async def test_async_function(): ...
```

```python
import pytest

@pytest.fixture
def anyio_backend():
    return "asyncio"  # or "trio"
```

--------------------------------

### Test FastAPI Lifespan Events with asgi-lifespan

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Demonstrates how to test FastAPI applications that utilize lifespan events for startup and shutdown procedures. It uses the `asgi-lifespan` package to manage these events during testing, ensuring proper initialization and cleanup of resources like database connections or HTTP clients.

```python
import anyio
from contextlib import asynccontextmanager
from typing import AsyncIterator

from asgi_lifespan import LifespanManager
from httpx import AsyncClient, ASGITransport
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    print("Starting app - initializing resources")
    yield
    print("Stopping app - cleaning up resources")


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def read_root():
    return {"status": "ready"}


async def test_with_lifespan():
    async with LifespanManager(app) as manager:
        async with AsyncClient(
            transport=ASGITransport(app=manager.app),
            base_url="http://test"
        ) as client:
            response = await client.get("/")
            assert response.status_code == 200


anyio.run(test_with_lifespan)
# Output:
# Starting app - initializing resources
# Stopping app - cleaning up resources
```

--------------------------------

### Test FastAPI with AsyncClient

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Demonstrates how to use httpx AsyncClient with ASGITransport to perform asynchronous integration tests on FastAPI applications.

```python
import anyio
from httpx import AsyncClient, ASGITransport

async def main():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get("/")
        assert response.status_code == 200
        assert response.json() == {"Hello": "World"}

anyio.run(main)
```

--------------------------------

### Use pytest.mark.anyio for Async FastAPI Tests

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Recommends using `pytest.mark.anyio` for testing asynchronous FastAPI applications, leveraging the `anyio` library which is already a dependency. This avoids the need for `pytest-asyncio` and allows for configuration of the test backend (asyncio or trio) via fixtures.

```python
import pytest
from httpx import AsyncClient, ASGITransport
from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}


# Fixture to use asyncio backend only (not trio)
@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.mark.anyio
async def test_read_item(anyio_backend):
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        response = await client.get("/items/42")
        assert response.status_code == 200
        assert response.json() == {"item_id": 42}


# Run with: pytest test_app.py -v
```

--------------------------------

### Test FastAPI with Lifespan Events

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Uses the asgi-lifespan package to manage application startup and shutdown events during testing, ensuring resources are properly initialized.

```python
from contextlib import asynccontextmanager
from typing import AsyncIterator
import anyio
from asgi_lifespan import LifespanManager
from httpx import AsyncClient, ASGITransport
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    print("Starting app")
    yield
    print("Stopping app")

app = FastAPI(lifespan=lifespan)

async def main():
    async with LifespanManager(app) as manager:
        async with AsyncClient(transport=ASGITransport(app=manager.app)) as client:
            response = await client.get("/")
            assert response.status_code == 200
            assert response.json() == {"Hello": "World"}

anyio.run(main)
```

--------------------------------

### Use FastAPI Lifespan State for Shared Objects

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Illustrates the recommended approach for sharing objects, such as database connections or HTTP clients, between startup and request handlers in FastAPI. It leverages the lifespan state for type safety and improved dependency injection, contrasting it with the less preferred `app.state`.

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any, TypedDict, cast

from fastapi import FastAPI, Request
from httpx import AsyncClient


class State(TypedDict):
    client: AsyncClient


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[State]:
    async with AsyncClient() as client:
        yield {"client": client}


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def read_root(request: Request) -> dict[str, Any]:
    # Access typed state from request
    client = cast(AsyncClient, request.state.client)
    response = await client.get("https://httpbin.org/get")
    return {"external_status": response.status_code}
```

--------------------------------

### Use HTTPX `AsyncClient` for Testing FastAPI

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Recommends using `httpx.AsyncClient` for testing FastAPI applications, especially when dealing with asynchronous operations. It provides a more suitable alternative to Starlette's `TestClient` for async-native testing scenarios.

```python
from fastapi import FastAPI


app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}


# Using TestClient
from starlette.testclient import TestClient

client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"Hello": "World"}
```

--------------------------------

### Handle WebSocket Messages with `async for`

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Shows the preferred method for handling incoming WebSocket messages using `async for` over a `while True` loop. This approach is cleaner and automatically handles `WebSocketDisconnect` exceptions, simplifying resource management.

```python
from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    async for data in websocket.iter_text():
        await websocket.send_text(f"Message text was: {data}")
```

--------------------------------

### Test Async Endpoints with HTTPX AsyncClient

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Use HTTPX's AsyncClient to test asynchronous FastAPI endpoints. This method avoids context-switching overhead and ensures that the full async code path is executed during testing.

```python
import anyio
from httpx import AsyncClient, ASGITransport
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}


async def test_endpoint():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        response = await client.get("/")
        assert response.status_code == 200
        assert response.json() == {"Hello": "World"}
        print(f"Response: {response.json()}")


# Run the test
anyio.run(test_endpoint)
```

--------------------------------

### Configure FastAPI Threadpool Size

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Demonstrates how to adjust the number of threads available in FastAPI's internal thread pool. This is useful for managing the performance impact of non-async functions, which are run in a thread pool via `anyio.to_thread.run_sync`. The default is 40 threads.

```python
import anyio
from contextlib import asynccontextmanager
from typing import Iterator

from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> Iterator[None]:
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100
    yield

app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Handle WebSockets with Async Iteration

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Use the async for loop with websocket.iter_text() for cleaner WebSocket communication. This approach automatically handles WebSocketDisconnect exceptions, reducing boilerplate code.

```python
from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    # Clean async iteration - auto-handles disconnect
    async for data in websocket.iter_text():
        await websocket.send_text(f"Echo: {data}")
    # Connection closed cleanly when loop exits
```

--------------------------------

### Manage Lifespan State

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Compares the deprecated app.state approach with the recommended Lifespan State pattern for sharing objects like HTTP clients across the request-response cycle.

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any, TypedDict, cast
from fastapi import FastAPI, Request
from httpx import AsyncClient

class State(TypedDict):
    client: AsyncClient

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[State]:
    async with AsyncClient(app=app) as client:
        yield {"client": client}

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def read_root(request: Request) -> dict[str, Any]:
    client = cast(AsyncClient, request.state.client)
    response = await client.get("/")
    return response.json()
```

--------------------------------

### Enable AsyncIO Debug Mode for Blocking Call Detection

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Explains how to activate AsyncIO debug mode to pinpoint endpoints that cause blocking issues within the event loop. When enabled, Python issues warnings for tasks exceeding a 100ms execution time, aiding in the identification of problematic calls like `time.sleep()` in asynchronous functions.

```python
import time
import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    time.sleep(1)  # BAD: Blocking call in async function!
    return {"Hello": "World"}


if __name__ == "__main__":
    uvicorn.run(app, loop="uvloop")

# Run with: PYTHONASYNCIODEBUG=1 python main.py
# When endpoint is called, you'll see:
# Executing <Task finished name='Task-3' ...> took 1.009 seconds
```

--------------------------------

### Make FastAPI Dependencies Async to Avoid Thread Pool

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Highlights the importance of defining dependencies as `async def` functions in FastAPI to ensure they run on the event loop rather than the thread pool. This practice is crucial for performance, especially for dependencies that do not involve blocking I/O operations, preventing unnecessary consumption of limited threads.

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from httpx import AsyncClient
from fastapi import FastAPI, Request, Depends


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, AsyncClient]]:
    async with AsyncClient() as client:
        yield {"client": client}


app = FastAPI(lifespan=lifespan)


# GOOD: async dependency runs on event loop
async def http_client(request: Request) -> AsyncClient:
    return request.state.client


# BAD: sync dependency uses thread pool
# def http_client(request: Request) -> AsyncClient:
#     return request.state.client


@app.get("/")
async def read_root(client: AsyncClient = Depends(http_client)):
    response = await client.get("https://httpbin.org/get")
    return {"status": response.status_code}
```

--------------------------------

### Handle WebSocket Disconnection with while True

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Implement the traditional while True loop for WebSocket communication while explicitly catching WebSocketDisconnect. This allows for graceful resource cleanup when a client disconnects.

```python
from fastapi import FastAPI
from starlette.websockets import WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        # Clean up resources here (close DB connections, etc.)
        print("Client disconnected")
```

--------------------------------

### Handle WebSocket Disconnect Exception

Source: https://github.com/kludex/fastapi-tips/blob/main/README.md

Illustrates how to explicitly catch the `WebSocketDisconnect` exception when using a `while True` loop for WebSocket communication. This is necessary for performing cleanup actions when a client disconnects. Newer FastAPI versions raise this exception for both send and receive operations.

```python
from fastapi import FastAPI
from starlette.websockets import WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        pass
```

--------------------------------

### Configure Thread Pool Size in FastAPI Lifespan

Source: https://context7.com/kludex/fastapi-tips/llms.txt

Increase the default thread pool size for synchronous endpoints to prevent blocking under high concurrency. This is achieved by modifying the anyio thread limiter within the FastAPI lifespan context.

```python
import anyio
from contextlib import asynccontextmanager
from typing import Iterator

from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> Iterator[None]:
    # Increase thread pool from default 40 to 100
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/")
def sync_endpoint():
    # This runs in thread pool - ensure adequate threads available
    return {"status": "processed"}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.