fastapi-tips

repository·main·Indexed 25 days ago

https://github.com/kludex/fastapi-tips

A collection of performance tips, best practices, and advanced usage patterns for FastAPI developers. Covers topics such as optimizing thread pool size, using uvloop and httptools, implementing Pure ASGI middleware, managing lifespan state, and testing async functions with httpx.AsyncClient and pytest.mark.anyio.

Tokens
2.3K
Snippets
9
Records
10
Agent score
36%

What's inside fastapi-tips

  1. Use `pytest.mark.anyio` for testing async functions

    main

    Since FastAPI depends on anyio, it is recommended to use pytest.mark.anyio instead of pytest.mark.asyncio for testing.

    By default, anyio runs tests twice (once for trio and once for asyncio). If you are testing an application rather than a library, you should restrict the backend to a single one using an anyio_backend fixture.

    import pytest
    
    @pytest.fixture
    def anyio_backend():
        return "asyncio"  # or "trio"
    
    @pytest.mark.anyio
    async def test_async_function():
        ...
  2. Use HTTPX `AsyncClient` for testing instead of `TestClient`

    main

    For applications using async functions, using httpx.AsyncClient is preferred over Starlette's TestClient.

    When testing applications that use lifespan events (on_startup, on_shutdown, or the lifespan parameter), use the asgi-lifespan package to ensure these events are triggered during your tests.

    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)
    
    
    @app.get("/")
    async def read_root():
        return {"Hello": "World"}
    
    
    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)
  3. Understand how non-async dependencies run in threads

    main

    In FastAPI, if a dependency function is defined using def instead of async def, it will be executed in a separate thread rather than the main event loop. To ensure a dependency runs directly in the event loop, define it with async def.

    If you need to verify if functions are consuming threads, you can monitor the anyio.to_thread.current_default_thread_limiter().

    # This runs in a thread
    def http_client(request: Request) -> AsyncClient:
        return request.state.client
    
    # This runs in the event loop
    async def http_client(request: Request) -> AsyncClient:
        return request.state.client
  4. Use Lifespan State instead of `app.state`

    main

    FastAPI recommends using the lifespan state to manage objects created at startup (like database connections or HTTP clients) that need to be accessed during the request-response cycle. Avoid using app.state directly.

    To implement this, define a TypedDict for your state and yield it from your lifespan context manager. You can then access these objects via request.state.

    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()
  5. Handle `WebSocketDisconnect` in WebSocket loops

    main

    If you choose to use the while True pattern for WebSockets instead of async for, you must manually catch the WebSocketDisconnect exception to prevent errors when a client disconnects.

    In recent FastAPI versions, all methods (both receive and send) will raise WebSocketDisconnect. Ensure your try/except block covers the necessary operations.

    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
  6. Enable AsyncIO debug mode to find blocking endpoints

    main

    To identify endpoints that are blocking the event loop (e.g., using time.sleep() instead of await asyncio.sleep()), enable AsyncIO debug mode. When enabled, Python will print a warning if a task takes longer than 100ms to execute.

    Run your application with the PYTHONASYNCIODEBUG=1 environment variable.

    PYTHONASYNCIODEBUG=1 python main.py
  7. Use `async for` for WebSocket message iteration

    main

    Instead of using a while True loop to receive messages from a WebSocket, use the async for notation with websocket.iter_text(). This is cleaner and automatically handles the connection closing logic.

    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}")
  8. Install `uvloop` and `httptools` for performance

    main

    By default, Uvicorn does not include uvloop and httptools, which are faster than the default asyncio event loop and HTTP parser. Installing them allows Uvicorn to use them automatically.

    Note: uvloop cannot be installed on Windows. If you are developing on Windows but deploying to Linux, use an environment marker in your dependencies to avoid installation errors on Windows.

    pip install uvloop httptools
  9. Optimize thread pool size for non-async functions

    main

    When you use non-async functions in FastAPI, they are executed in a thread pool using run_in_threadpool (which uses anyio.to_thread.run_sync). By default, there are only 40 threads available. If all threads are occupied, your application will be blocked.

    You can increase the number of available threads by configuring the anyio thread limiter within a lifespan handler.

    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)