Python FastAPI Patterns

Python FastAPI 实践模式

Project Structure

项目结构

src/
├── main.py           # App entry point
├── config.py         # Settings management
├── dependencies.py   # Shared dependencies
├── models/           # Pydantic models
├── routes/           # API endpoints
├── services/         # Business logic
├── repositories/     # Data access
└── utils/            # Helpers

src/
├── main.py           # App entry point
├── config.py         # Settings management
├── dependencies.py   # Shared dependencies
├── models/           # Pydantic models
├── routes/           # API endpoints
├── services/         # Business logic
├── repositories/     # Data access
└── utils/            # Helpers

Basic Setup

基础设置

python

undefined

python

undefined

main.py

from fastapi import FastAPI from contextlib import asynccontextmanager

@asynccontextmanager async def lifespan(app: FastAPI): # Startup await init_db() yield # Shutdown await close_db()

app = FastAPI( title="My API", version="1.0.0", lifespan=lifespan, )

undefined

from fastapi import FastAPI from contextlib import asynccontextmanager

@asynccontextmanager async def lifespan(app: FastAPI): # Startup await init_db() yield # Shutdown await close_db()

app = FastAPI( title="My API", version="1.0.0", lifespan=lifespan, )

undefined

Pydantic Models

Pydantic 模型

python

from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
from typing import Optional

class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(min_length=2, max_length=100)

class UserCreate(UserBase):
    password: str = Field(min_length=8)

class UserResponse(UserBase):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    name: Optional[str] = Field(None, min_length=2, max_length=100)

python

from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
from typing import Optional

class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(min_length=2, max_length=100)

class UserCreate(UserBase):
    password: str = Field(min_length=8)

class UserResponse(UserBase):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    name: Optional[str] = Field(None, min_length=2, max_length=100)

Route Organization

路由组织

python

undefined

python

undefined

routes/users.py

from fastapi import APIRouter, Depends, HTTPException, status from typing import Annotated

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/", response_model=list[UserResponse]) async def list_users( skip: int = 0, limit: int = Query(20, le=100), db: Session = Depends(get_db), ): return await user_service.get_all(db, skip=skip, limit=limit)

@router.get("/{user_id}", response_model=UserResponse) async def get_user( user_id: int, db: Session = Depends(get_db), ): user = await user_service.get_by_id(db, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED) async def create_user( user_in: UserCreate, db: Session = Depends(get_db), ): return await user_service.create(db, user_in)

undefined

from fastapi import APIRouter, Depends, HTTPException, status from typing import Annotated

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/", response_model=list[UserResponse]) async def list_users( skip: int = 0, limit: int = Query(20, le=100), db: Session = Depends(get_db), ): return await user_service.get_all(db, skip=skip, limit=limit)

@router.get("/{user_id}", response_model=UserResponse) async def get_user( user_id: int, db: Session = Depends(get_db), ): user = await user_service.get_by_id(db, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED) async def create_user( user_in: UserCreate, db: Session = Depends(get_db), ): return await user_service.create(db, user_in)

undefined

Dependency Injection

依赖注入

python

undefined

python

undefined

dependencies.py

from fastapi import Depends, HTTPException, Security from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_db(): db = SessionLocal() try: yield db finally: await db.close()

async def get_current_user( credentials: HTTPAuthorizationCredentials = Security(security), db: Session = Depends(get_db), ) -> User: token = credentials.credentials payload = decode_token(token) if not payload: raise HTTPException(status_code=401, detail="Invalid token")

user = await user_service.get_by_id(db, payload["sub"])
if not user:
    raise HTTPException(status_code=401, detail="User not found")
return user

from fastapi import Depends, HTTPException, Security from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_db(): db = SessionLocal() try: yield db finally: await db.close()

async def get_current_user( credentials: HTTPAuthorizationCredentials = Security(security), db: Session = Depends(get_db), ) -> User: token = credentials.credentials payload = decode_token(token) if not payload: raise HTTPException(status_code=401, detail="Invalid token")

user = await user_service.get_by_id(db, payload["sub"])
if not user:
    raise HTTPException(status_code=401, detail="User not found")
return user

Type alias for cleaner signatures

CurrentUser = Annotated[User, Depends(get_current_user)]

undefined

CurrentUser = Annotated[User, Depends(get_current_user)]

undefined

Async Patterns

异步模式

python

import asyncio
from httpx import AsyncClient

python

import asyncio
from httpx import AsyncClient

Parallel requests

async def fetch_all_data(user_id: int): async with AsyncClient() as client: tasks = [ client.get(f"/api/user/{user_id}"), client.get(f"/api/user/{user_id}/posts"), client.get(f"/api/user/{user_id}/stats"), ] results = await asyncio.gather(*tasks) return { "user": results[0].json(), "posts": results[1].json(), "stats": results[2].json(), }

Background tasks

from fastapi import BackgroundTasks

@router.post("/notify") async def send_notification( email: str, background_tasks: BackgroundTasks, ): background_tasks.add_task(send_email, email) return {"message": "Notification queued"}

undefined

from fastapi import BackgroundTasks

@router.post("/notify") async def send_notification( email: str, background_tasks: BackgroundTasks, ): background_tasks.add_task(send_email, email) return {"message": "Notification queued"}

undefined

Error Handling

错误处理

python

from fastapi import Request
from fastapi.responses import JSONResponse

class AppException(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": {"code": exc.code, "message": exc.message}
        },
    )

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "error": {"code": "INTERNAL_ERROR", "message": "Something went wrong"}
        },
    )

python

from fastapi import Request
from fastapi.responses import JSONResponse

class AppException(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": {"code": exc.code, "message": exc.message}
        },
    )

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "error": {"code": "INTERNAL_ERROR", "message": "Something went wrong"}
        },
    )

Settings Management

配置管理

python

undefined

python

undefined

config.py

from pydantic_settings import BaseSettings from functools import lru_cache

class Settings(BaseSettings): app_name: str = "My API" debug: bool = False database_url: str secret_key: str

model_config = {"env_file": ".env"}

@lru_cache def get_settings() -> Settings: return Settings()

undefined

from pydantic_settings import BaseSettings from functools import lru_cache

class Settings(BaseSettings): app_name: str = "My API" debug: bool = False database_url: str secret_key: str

model_config = {"env_file": ".env"}

@lru_cache def get_settings() -> Settings: return Settings()

undefined

Middleware

中间件

python

from fastapi import Request
from time import time
import logging

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time()
    response = await call_next(request)
    duration = time() - start

    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} duration={duration:.3f}s"
    )
    return response

python

from fastapi import Request
from time import time
import logging

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time()
    response = await call_next(request)
    duration = time() - start

    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} duration={duration:.3f}s"
    )
    return response

Testing

测试

python

import pytest
from httpx import AsyncClient
from main import app

@pytest.fixture
async def client():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post(
        "/users/",
        json={"email": "test@example.com", "name": "Test", "password": "password123"}
    )
    assert response.status_code == 201
    assert response.json()["email"] == "test@example.com"

python

import pytest
from httpx import AsyncClient
from main import app

@pytest.fixture
async def client():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post(
        "/users/",
        json={"email": "test@example.com", "name": "Test", "password": "password123"}
    )
    assert response.status_code == 201
    assert response.json()["email"] == "test@example.com"

Best Practices

最佳实践

Use type hints everywhere for auto-documentation
Validate at boundaries with Pydantic models
Inject dependencies for testability
Handle errors consistently with custom exceptions
Use async/await for I/O operations
Keep routes thin - business logic in services
Document with OpenAPI annotations

处处使用类型提示 以实现自动文档生成
在边界处进行验证，使用Pydantic模型
注入依赖项 以提升可测试性
使用自定义异常 实现一致的错误处理
针对I/O操作使用async/await
保持路由轻量化 - 业务逻辑放在服务层
使用OpenAPI注解 进行文档编写

python-fastapi-patterns

Original

Translation

Python FastAPI Patterns

Python FastAPI 实践模式

Project Structure

项目结构

Basic Setup

基础设置

main.py

main.py

Pydantic Models

Pydantic 模型

Route Organization

路由组织

routes/users.py

routes/users.py

Dependency Injection

依赖注入

dependencies.py

dependencies.py

Type alias for cleaner signatures

Type alias for cleaner signatures

Async Patterns

异步模式

Parallel requests

Parallel requests

Background tasks

Background tasks

Error Handling

错误处理

Settings Management

配置管理

config.py

config.py

Middleware

中间件

Testing

测试

Best Practices

最佳实践