FastAPI Best Practices

repository·master·Indexed 12 days ago

https://github.com/zhanymkanov/fastapi-best-practices

A collection of opinionated architectural patterns and best practices for building production-ready FastAPI applications. Covers domain-driven project structure, I/O and CPU task optimization, advanced Pydantic validation, dependency chaining, and production configuration for scalability and maintainability.

Tokens
11.4K
Snippets
30
Records
37
Agent score
47%

What's inside FastAPI Best Practices

  1. Implement a Custom Base Model for global behavior

    master

    Create a global CustomModel inheriting from BaseModel to enforce application-wide standards. Common use cases include:

    • Standardizing datetime serialization (e.g., forcing UTC and specific string formats) using @field_serializer.
    • Adding utility methods like serializable_dict() to return dictionaries containing only JSON-serializable fields using jsonable_encoder.
    from datetime import datetime
    from typing import Any
    from zoneinfo import ZoneInfo
    from fastapi.encoders import jsonable_encoder
    from pydantic import BaseModel, ConfigDict, field_serializer
    
    class CustomModel(BaseModel):
        model_config = ConfigDict(populate_by_name=True)
    
        @field_serializer("*", when_used="json", check_fields=False)
        def _serialize_datetimes(self, value: Any) -> Any:
            if isinstance(value, datetime):
                if value.tzinfo is None:
                    value = value.replace(tzinfo=ZoneInfo("UTC"))
                return value.strftime("%Y-%m-%dT%H:%M:%S%z")
            return value
    
        def serializable_dict(self, **kwargs):
            """Returns a dictionary containing only serializable fields."""
            default_dict = self.model_dump()
            return jsonable_encoder(default_dict)
  2. Handle ValueError in Pydantic models

    master

    When you raise a ValueError inside a Pydantic @field_validator or @model_validator, FastAPI automatically catches it and returns a detailed 422 Unprocessable Entity response to the client, including the error message. This is useful for client-facing validation logic.

    class ProfileCreate(BaseModel):
        username: str
        
        @field_validator("password", mode="after")
        @classmethod
        def valid_password(cls, password: str) -> str:
            if not re.match(STRONG_PASSWORD_PATTERN, password):
                raise ValueError("Password must contain...)
            return password
  3. Optimize dependencies with caching and async preference

    master

    FastAPI caches the result of a dependency within the scope of a single request. If multiple dependencies in a single route call the same sub-dependency (e.g., parse_jwt_data), it is only executed once.

    Best Practices:

    • Split dependencies: Break them into smaller, granular functions to maximize reuse and caching benefits.
    • Prefer async dependencies: Even for non-I/O tasks, async dependencies avoid the overhead of running in a thread pool, which is required for synchronous dependencies.
  4. Optimize CPU intensive tasks

    master

    Standard async/await and threadpools are designed for I/O-bound tasks. They do not solve performance issues for CPU-bound tasks (e.g., heavy math, data processing, video transcoding).

    Why they fail for CPU tasks:

    • Async/Await: Awaiting a CPU task provides no benefit because the CPU must actively work to complete the calculation; it's not waiting on an external resource.
    • Threads: Due to Python's Global Interpreter Lock (GIL), only one thread can execute Python bytecode at a time. Running CPU-intensive work in multiple threads will not provide true parallelism.

    The Solution: To optimize CPU-intensive tasks, offload them to worker processes using:

    • The multiprocessing module.
    • A dedicated task queue like Celery.
  5. Optimize CPU-intensive tasks using worker processes

    master

    For CPU-intensive tasks (e.g., heavy mathematical computations, data processing, or video transcoding), neither async def nor standard thread pools are effective due to Python's Global Interpreter Lock (GIL). The GIL ensures only one thread can execute Python bytecode at a time, meaning threads cannot provide true parallelism for CPU-bound work.

    Best Practice: Instead of running CPU-intensive tasks within the FastAPI application process, offload them to worker nodes in separate processes (e.g., using Celery, RQ, or a dedicated microservice).

  6. Handle I/O intensive tasks correctly in FastAPI

    master

    FastAPI handles async and sync routes differently. Choosing the wrong one for I/O operations can block your entire server.

    Guidelines for I/O (DB calls, API requests, File I/O):

    1. Use async def with non-blocking calls: This is the most efficient method. Use await with asynchronous libraries (e.g., asyncio.sleep(), httpx, motor). The event loop remains free to handle other requests.
    2. Use def (sync) for blocking calls: If you must use a synchronous library that blocks (e.g., time.sleep(), requests), define the route with def instead of async def. FastAPI will automatically run these in a separate threadpool, preventing the main event loop from being blocked.
    3. Avoid async def with blocking calls: Never use time.sleep() or synchronous I/O inside an async def function. This blocks the entire event loop, meaning the server cannot accept any new requests until the operation finishes.

    Comparison Summary:

    Route TypeImplementationBehavior
    Perfectasync def + awaitNon-blocking; event loop stays free.
    Gooddef + blocking callOffloaded to threadpool; event loop stays free.
    Terribleasync def + blocking callBlocks the entire event loop; server stops responding.
    import asyncio
    import time
    from fastapi import APIRouter
    
    router = APIRouter()
    
    # BAD: Blocks the entire event loop
    @router.get("/terrible-ping")
    async def terrible_ping():
        time.sleep(10) 
        return {"pong": True}
    
    # GOOD: Runs in a separate threadpool
    @router.get("/good-ping")
    def good_ping():
        time.sleep(10) 
        return {"pong": True}
    
    # PERFECT: Non-blocking awaitable
    @router.get("/perfect-ping")
    async def perfect_ping():
        await asyncio.sleep(10) 
        return {"pong": True}
  7. Handle Pydantic Validation Errors

    master

    When you raise a ValueError inside a Pydantic field_validator, FastAPI automatically catches it and returns a detailed 422 Unprocessable Entity validation error response to the client. This is a clean way to enforce business logic constraints (like password strength) within your schemas.

    from pydantic import BaseModel, field_validator
    import re
    
    class ProfileCreate(BaseModel):
        username: str
        password: str
        
        @field_validator("password", mode="after")
        @classmethod
        def valid_password(cls, password: str) -> str:
            if not re.match(STRONG_PASSWORD_PATTERN, password):
                # This ValueError results in a FastAPI 422 response
                raise ValueError("Password must contain at least one lower, one upper, digit, or special symbol")
            return password
  8. Chain and reuse Dependencies

    master

    Dependencies can depend on other dependencies, allowing you to build complex validation logic from simple, reusable building blocks.

    Key Benefit: Caching. FastAPI caches the result of a dependency within a single request's scope. If multiple dependencies in a single route call the same underlying dependency (e.g., parse_jwt_data), that dependency is executed only once.

    # dependencies.py
    async def parse_jwt_data(token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))) -> dict:
        # ... logic to decode JWT
        return {"user_id": payload["id"]}
    
    async def valid_owned_post(
        post: Mapping = Depends(valid_post_id), 
        token_data: dict = Depends(parse_jwt_data),
    ) -> Mapping:
        if post["creator_id"] != token_data["user_id"]:
            raise UserNotOwner()
        return post
    
    # router.py
    @router.get("/users/{user_id}/posts/{post_id}")
    async def get_user_post(
        post: Mapping = Depends(valid_owned_post),
        user: Mapping = Depends(valid_active_creator),
    ):
        # parse_jwt_data is called only once here, even though used by both dependencies
        return post
  9. Choose between BackgroundTasks and Task Queues

    master

    FastAPI's BackgroundTasks runs tasks after the response is sent, in the same worker process. Use it only for short, non-critical tasks where failure can be silently dropped. For anything else, use a dedicated task queue like Celery, Arq, or RQ.

    FeatureBackgroundTasksCelery / Arq / RQ
    Task DurationShort (< 1 second)Seconds to minutes
    ReliabilityLost if worker diesRetries & dead-letter handling
    WorkloadIn-process (email, logging)CPU-heavy or separate workers
    CapabilitiesNo scheduling/rate limitingCron, ETA, rate limiting
    from fastapi import BackgroundTasks
    
    @router.post("/signup")
    async def signup(data: SignupIn, bg: BackgroundTasks):
        user = await service.create_user(data)
        # Use for fire-and-forget, in-process tasks
        bg.add_task(send_welcome_email, user.email) 
        return user
  10. Handle I/O-intensive tasks with async or sync routes

    master

    FastAPI handles synchronous and asynchronous routes differently. Choosing the right one is critical for performance:

    1. async def (The Perfect Way): Use this for non-blocking I/O operations (e.g., using await asyncio.sleep() or an async database driver). This allows the event loop to process other requests while waiting.
    2. def (The Safe Way for Sync I/O): If you must use a blocking synchronous library (e.g., time.sleep() or a synchronous SDK), define the route with def instead of async def. FastAPI will run these in a separate thread pool, preventing the main event loop from being blocked.
    3. async def with Blocking Code (The Terrible Way): Never perform blocking I/O (like time.sleep()) inside an async def route. This will block the entire event loop, preventing the server from handling any other incoming requests until the operation completes.

    Summary Table:

    Route TypeImplementationBehaviorBest For
    async defawait async_call()Non-blocking; event loop continuesAsync I/O (DB, APIs)
    defsync_call()Runs in a thread pool; event loop continuesSync I/O (Legacy SDKs)
    async defsync_call()Blocks the entire event loopDO NOT USE
    import asyncio
    import time
    from fastapi import APIRouter
    
    router = APIRouter()
    
    # BAD: Blocks the entire process
    @router.get("/terrible-ping")
    async def terrible_ping():
        time.sleep(10)
        return {"pong": True}
    
    # GOOD: Runs in a separate thread pool
    @router.get("/good-ping")
    def good_ping():
        time.sleep(10)
        return {"pong": True}
    
    # PERFECT: Non-blocking I/O
    @router.get("/perfect-ping")
    async def perfect_ping():
        await asyncio.sleep(10)
        return {"pong": True}
  11. Reuse dependencies by following REST conventions

    master

    To maximize dependency reuse across different routes, use consistent variable names in your path parameters. For example, if multiple endpoints validate a profile, use {profile_id} in all of them. This allows you to chain dependencies: a valid_creator_id dependency can depend on a valid_profile_id dependency.

    # src.profiles.dependencies
    async def valid_profile_id(profile_id: UUID4) -> Mapping:
        profile = await service.get_by_id(profile_id)
        if not profile:
            raise ProfileNotFound()
        return profile
    
    # src.creators.dependencies
    async def valid_creator_id(profile: Mapping = Depends(valid_profile_id)) -> Mapping:
        if not profile["is_creator"]:
           raise ProfileNotCreator()
        return profile
    
    # src.creators.router.py
    @router.get("/creators/{profile_id}", response_model=ProfileResponse)
    async def get_user_profile_by_id(creator_profile: Mapping = Depends(valid_creator_id)):
        return creator_profile
  12. Understand FastAPI response serialization overhead

    master

    Be aware that returning a Pydantic model from a route that has a response_model defined causes the model to be instantiated twice. First, you create it to return from the function; second, FastAPI implicitly creates it again to validate the data against the response_model. For high-performance paths, consider returning raw dictionaries or optimized data structures to avoid this double instantiation.

    @app.get("/", response_model=ProfileResponse)
    async def root():
        # This creates the model once
        return ProfileResponse()
        # FastAPI then creates it a second time for validation