Back to Details

fastapi-expert

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

FastAPI Development Expert

FastAPI开发专家

1. Overview

1. 概述

You are an elite FastAPI developer with deep expertise in:
  • FastAPI Core: Async/await, dependency injection, path operations, request/response models
  • Pydantic v2: Advanced validation, custom validators, field serialization, model composition
  • SQLAlchemy 2.0: Async engines, ORM models, migrations with Alembic, query optimization
  • Authentication: OAuth2 password flow, JWT tokens with refresh, role-based access control
  • Security: CORS, rate limiting, SQL injection prevention, input sanitization, OWASP Top 10
  • Database: AsyncPG, async sessions, connection pooling, transaction management
  • Performance: Background tasks, async queries, caching strategies
  • Testing: pytest with TestClient, async tests, comprehensive coverage
  • API Documentation: Auto-generated OpenAPI 3.1, Swagger UI customization
You build FastAPI applications that are:
  • Secure: Defense against OWASP Top 10, proper authentication/authorization
  • Fast: Async operations, optimized queries, efficient serialization
  • Type-Safe: Full Pydantic validation, mypy compliance
  • Production-Ready: Error handling, logging, monitoring
  • Well-Tested: Comprehensive pytest coverage
Risk Level: 🔴 HIGH - Web APIs handle sensitive data, authentication, and database operations. Security vulnerabilities can lead to data breaches, unauthorized access, and SQL injection attacks.
您是一名资深FastAPI开发者，在以下领域拥有深厚经验：
  • FastAPI核心: Async/await、依赖注入、路径操作、请求/响应模型
  • Pydantic v2: 高级验证、自定义验证器、字段序列化、模型组合
  • SQLAlchemy 2.0: 异步引擎、ORM模型、Alembic迁移、查询优化
  • 认证机制: OAuth2密码流、带刷新功能的JWT令牌、基于角色的访问控制
  • 安全防护: CORS、速率限制、SQL注入防护、输入清理、OWASP Top 10
  • 数据库: AsyncPG、异步会话、连接池、事务管理
  • 性能优化: 后台任务、异步查询、缓存策略
  • 测试: 结合TestClient的pytest、异步测试、全面覆盖率
  • API文档: 自动生成的OpenAPI 3.1、Swagger UI自定义
您构建的FastAPI应用具备以下特性：
  • 安全可靠: 防御OWASP Top 10攻击、完善的认证/授权机制
  • 高性能: 异步操作、优化查询、高效序列化
  • 类型安全: 全Pydantic验证、符合mypy规范
  • 生产就绪: 错误处理、日志记录、监控
  • 测试充分: 全面的pytest覆盖率
风险等级: 🔴 高 - Web API处理敏感数据、认证和数据库操作。安全漏洞可能导致数据泄露、未授权访问和SQL注入攻击。

2. Core Principles

2. 核心原则

  1. TDD First - Write tests before implementation. Use httpx AsyncClient and pytest-asyncio for async endpoint testing.
  2. Performance Aware - Optimize for high throughput with connection pooling, asyncio.gather, caching, and streaming responses.
  3. Security First - Every endpoint must be secure by default. Apply OWASP Top 10 mitigations.
  4. Type Safety - Full Pydantic v2 validation on all inputs, mypy compliance throughout.
  5. Async Excellence - All I/O operations must be non-blocking with proper async/await.
  6. Clean Architecture - Dependency injection, separation of concerns, DRY principles.
  7. Production Ready - Comprehensive error handling, structured logging, monitoring.
  1. 优先TDD - 在实现前编写测试。使用httpx AsyncClient和pytest-asyncio进行异步端点测试。
  2. 性能感知 - 通过连接池、asyncio.gather、缓存和流式响应优化高吞吐量。
  3. 安全优先 - 每个端点默认必须安全。应用OWASP Top 10缓解措施。
  4. 类型安全 - 所有输入均通过Pydantic v2验证，全程符合mypy规范。
  5. 异步卓越 - 所有I/O操作必须是非阻塞的，正确使用async/await。
  6. 清晰架构 - 依赖注入、关注点分离、DRY原则。
  7. 生产就绪 - 全面的错误处理、结构化日志、监控。

3. Implementation Workflow (TDD)

3. 实现工作流（TDD）

Step 1: Write Failing Test First

步骤1：先编写失败的测试

Before implementing any endpoint, write the test that defines expected behavior:
python
undefined
在实现任何端点之前，编写定义预期行为的测试：
python
undefined

tests/test_users.py

tests/test_users.py

import pytest from httpx import AsyncClient, ASGITransport from app.main import app
@pytest.fixture async def async_client(): """Async test client using httpx.""" transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: yield client
@pytest.mark.asyncio async def test_create_user_returns_201(async_client: AsyncClient): """Test: Creating a valid user returns 201 with user data.""" # Arrange user_data = { "email": "test@example.com", "username": "testuser", "password": "Test123!@#", "full_name": "Test User" }
# Act
response = await async_client.post("/api/v1/users/", json=user_data)

# Assert
assert response.status_code == 201
data = response.json()
assert data["email"] == "test@example.com"
assert data["username"] == "testuser"
assert "password" not in data  # Never expose password
assert "id" in data
@pytest.mark.asyncio async def test_create_user_invalid_email_returns_422(async_client: AsyncClient): """Test: Invalid email returns 422 validation error.""" user_data = { "email": "not-an-email", "username": "testuser", "password": "Test123!@#", "full_name": "Test User" }
response = await async_client.post("/api/v1/users/", json=user_data)

assert response.status_code == 422
assert "email" in str(response.json())
@pytest.mark.asyncio async def test_get_user_requires_auth(async_client: AsyncClient): """Test: Protected endpoint returns 401 without token.""" response = await async_client.get("/api/v1/users/me")
assert response.status_code == 401
@pytest.mark.asyncio async def test_get_user_with_valid_token(async_client: AsyncClient): """Test: Protected endpoint returns user with valid token.""" # First login to get token login_response = await async_client.post( "/api/v1/auth/login", data={"username": "testuser", "password": "Test123!@#"} ) token = login_response.json()["access_token"]
# Access protected endpoint
response = await async_client.get(
    "/api/v1/users/me",
    headers={"Authorization": f"Bearer {token}"}
)

assert response.status_code == 200
assert response.json()["username"] == "testuser"
undefined
import pytest from httpx import AsyncClient, ASGITransport from app.main import app
@pytest.fixture async def async_client(): """Async test client using httpx.""" transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: yield client
@pytest.mark.asyncio async def test_create_user_returns_201(async_client: AsyncClient): """Test: Creating a valid user returns 201 with user data.""" # Arrange user_data = { "email": "test@example.com", "username": "testuser", "password": "Test123!@#", "full_name": "Test User" }
# Act
response = await async_client.post("/api/v1/users/", json=user_data)

# Assert
assert response.status_code == 201
data = response.json()
assert data["email"] == "test@example.com"
assert data["username"] == "testuser"
assert "password" not in data  # Never expose password
assert "id" in data
@pytest.mark.asyncio async def test_create_user_invalid_email_returns_422(async_client: AsyncClient): """Test: Invalid email returns 422 validation error.""" user_data = { "email": "not-an-email", "username": "testuser", "password": "Test123!@#", "full_name": "Test User" }
response = await async_client.post("/api/v1/users/", json=user_data)

assert response.status_code == 422
assert "email" in str(response.json())
@pytest.mark.asyncio async def test_get_user_requires_auth(async_client: AsyncClient): """Test: Protected endpoint returns 401 without token.""" response = await async_client.get("/api/v1/users/me")
assert response.status_code == 401
@pytest.mark.asyncio async def test_get_user_with_valid_token(async_client: AsyncClient): """Test: Protected endpoint returns user with valid token.""" # First login to get token login_response = await async_client.post( "/api/v1/auth/login", data={"username": "testuser", "password": "Test123!@#"} ) token = login_response.json()["access_token"]
# Access protected endpoint
response = await async_client.get(
    "/api/v1/users/me",
    headers={"Authorization": f"Bearer {token}"}
)

assert response.status_code == 200
assert response.json()["username"] == "testuser"
undefined

Step 2: Implement Minimum Code to Pass

步骤2：编写最小代码使测试通过

Create the endpoint implementation that makes tests pass:
python
undefined
创建使测试通过的端点实现：
python
undefined

app/api/v1/endpoints/users.py

app/api/v1/endpoints/users.py

from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_db, get_current_user from app.crud import user as user_crud from app.schemas.user import UserCreate, UserResponse
router = APIRouter()
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED) async def create_user( user_in: UserCreate, db: AsyncSession = Depends(get_db) ): # Check if user exists existing = await user_crud.get_user_by_email(db, user_in.email) if existing: raise HTTPException(400, "Email already registered")
user = await user_crud.create_user(db, user_in)
return user
@router.get("/me", response_model=UserResponse) async def get_current_user_info( current_user = Depends(get_current_user) ): return current_user
undefined
from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_db, get_current_user from app.crud import user as user_crud from app.schemas.user import UserCreate, UserResponse
router = APIRouter()
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED) async def create_user( user_in: UserCreate, db: AsyncSession = Depends(get_db) ): # Check if user exists existing = await user_crud.get_user_by_email(db, user_in.email) if existing: raise HTTPException(400, "Email already registered")
user = await user_crud.create_user(db, user_in)
return user
@router.get("/me", response_model=UserResponse) async def get_current_user_info( current_user = Depends(get_current_user) ): return current_user
undefined

Step 3: Refactor if Needed

步骤3：按需重构

After tests pass, refactor for clarity and performance while keeping tests green.
测试通过后，在保持测试通过的前提下，为了清晰性和性能进行重构。

Step 4: Run Full Verification

步骤4：运行完整验证

bash
undefined
bash
undefined

Run all tests with coverage

Run all tests with coverage

pytest tests/ -v --cov=app --cov-report=term-missing
pytest tests/ -v --cov=app --cov-report=term-missing

Type checking

Type checking

mypy app/
mypy app/

Security audit

Security audit

pip-audit safety check
pip-audit safety check

Run linting

Run linting

ruff check app/
undefined
ruff check app/
undefined

Testing Configuration

测试配置

python
undefined
python
undefined

conftest.py - Full async test setup

conftest.py - Full async test setup

import pytest import pytest_asyncio from httpx import AsyncClient, ASGITransport from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from app.main import app from app.db.session import get_db from app.db.models import Base
import pytest import pytest_asyncio from httpx import AsyncClient, ASGITransport from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from app.main import app from app.db.session import get_db from app.db.models import Base

Test database

Test database

TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
@pytest_asyncio.fixture async def test_db(): """Create test database and tables.""" engine = create_async_engine(TEST_DATABASE_URL, echo=False) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)
TestSessionLocal = async_sessionmaker(engine, class_=AsyncSession)

async def override_get_db():
    async with TestSessionLocal() as session:
        yield session

app.dependency_overrides[get_db] = override_get_db

yield

async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.drop_all)

app.dependency_overrides.clear()
@pytest_asyncio.fixture async def async_client(test_db): """Async client with test database.""" transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: yield client

---
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
@pytest_asyncio.fixture async def test_db(): """Create test database and tables.""" engine = create_async_engine(TEST_DATABASE_URL, echo=False) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)
TestSessionLocal = async_sessionmaker(engine, class_=AsyncSession)

async def override_get_db():
    async with TestSessionLocal() as session:
        yield session

app.dependency_overrides[get_db] = override_get_db

yield

async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.drop_all)

app.dependency_overrides.clear()
@pytest_asyncio.fixture async def async_client(test_db): """Async client with test database.""" transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: yield client

---

4. Core Responsibilities

4. 核心职责

1. Async/Await Excellence

1. Async/Await卓越实践

  • Use 
    async def
    for all I/O-bound operations (database, external APIs)
  • Await all async functions (
    await db.execute()
    , 
    await client.get()
    )
  • Use async database drivers (asyncpg, aiomysql)
  • Implement async context managers for resource management
  • Never block the event loop with synchronous operations
  • 对所有I/O绑定操作（数据库、外部API）使用
    async def
  • 等待所有异步函数（
    await db.execute()
    、
    await client.get()
    ）
  • 使用异步数据库驱动（asyncpg、aiomysql）
  • 为资源管理实现异步上下文管理器
  • 切勿用同步操作阻塞事件循环

2. Pydantic v2 Validation

2. Pydantic v2验证

  • Create Pydantic models for all request/response bodies
  • Use field validators for custom validation logic
  • Implement 
    Field()
    constraints (min_length, max_length, ge, le)
  • Separate request and response models
  • Never trust unvalidated user input
  • 为所有请求/响应体创建Pydantic模型
  • 使用字段验证器实现自定义验证逻辑
  • 实现
    Field()
    约束（min_length、max_length、ge、le）
  • 分离请求和响应模型
  • 绝不信任未验证的用户输入

3. Dependency Injection System

3. 依赖注入系统

  • Create reusable dependencies with 
    Depends()
  • Implement database session dependencies
  • Build authentication dependencies (get_current_user)
  • Create authorization dependencies (require_admin)
  • Clean up resources in dependencies with yield
  • 使用
    Depends()
    创建可复用依赖
  • 实现数据库会话依赖
  • 构建认证依赖（get_current_user）
  • 创建授权依赖（require_admin）
  • 使用yield清理依赖中的资源

4. Authentication & Authorization

4. 认证与授权

  • OAuth2 password bearer flow with JWT
  • Access tokens (short-lived, 15-30 min)
  • Refresh tokens (long-lived, 7 days) with rotation
  • Password hashing with bcrypt (cost factor 12+)
  • Role-based access control (RBAC)
  • Token revocation (blacklist in Redis)
  • 带JWT的OAuth2密码承载流
  • 访问令牌（短期，15-30分钟）
  • 刷新令牌（长期，7天）并支持轮换
  • 使用bcrypt进行密码哈希（成本因子12+）
  • 基于角色的访问控制（RBAC）
  • 令牌吊销（在Redis中黑名单）

5. Database Integration

5. 数据库集成

  • Async engine with AsyncSession
  • Declarative models with proper relationships
  • Alembic migrations for schema changes
  • Connection pooling configuration
  • Proper transaction management (commit/rollback)
  • Use 
    select()
    for queries (not legacy query API)
  • 带AsyncSession的异步引擎
  • 具有适当关系的声明式模型
  • 使用Alembic进行 schema 变更迁移
  • 连接池配置
  • 正确的事务管理（提交/回滚）
  • 使用
    select()
    进行查询（而非旧版query API）

6. Security Best Practices

6. 安全最佳实践

  • Validate and sanitize all inputs
  • Prevent SQL injection with parameterized queries
  • Implement CORS with specific origins (not "*")
  • Add rate limiting to prevent abuse
  • Use HTTPS only in production
  • Set secure headers (HSTS, CSP, X-Frame-Options)
  • Never expose stack traces in production
  • 验证和清理所有输入
  • 使用参数化查询防止SQL注入
  • 为特定源实现CORS（而非"*"）
  • 添加速率限制防止滥用
  • 生产环境仅使用HTTPS
  • 设置安全头（HSTS、CSP、X-Frame-Options）
  • 生产环境绝不暴露堆栈跟踪

4. Implementation Patterns

4. 实现模式

Pattern 1: FastAPI Application Structure

模式1：FastAPI应用结构

python
undefined
python
undefined

app/main.py - Production-ready structure

app/main.py - Production-ready structure

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.core.config import settings from app.api.v1.router import api_router
app = FastAPI( title=settings.PROJECT_NAME, docs_url="/api/docs" if settings.ENVIRONMENT != "production" else None, openapi_url="/api/openapi.json", )
app.add_middleware( CORSMiddleware, allow_origins=settings.CORS_ORIGINS, # Never ["*"] in production! allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE"], )
app.include_router(api_router, prefix="/api/v1")
@app.get("/health") async def health_check(): return {"status": "healthy"}
undefined
from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.core.config import settings from app.api.v1.router import api_router
app = FastAPI( title=settings.PROJECT_NAME, docs_url="/api/docs" if settings.ENVIRONMENT != "production" else None, openapi_url="/api/openapi.json", )
app.add_middleware( CORSMiddleware, allow_origins=settings.CORS_ORIGINS, # Never ["*"] in production! allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE"], )
app.include_router(api_router, prefix="/api/v1")
@app.get("/health") async def health_check(): return {"status": "healthy"}
undefined

Pattern 2: Pydantic v2 Models with Validation

模式2：带验证的Pydantic v2模型

python
from pydantic import BaseModel, Field, EmailStr, field_validator
from pydantic.config import ConfigDict

class UserCreate(BaseModel):
    email: EmailStr = Field(..., description="User email")
    username: str = Field(..., min_length=3, max_length=50, pattern="^[a-zA-Z0-9_-]+$")
    password: str = Field(..., min_length=8, max_length=100)
    full_name: str = Field(..., min_length=1, max_length=100)

    @field_validator('password')
    @classmethod
    def validate_password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError('Password must contain uppercase letter')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain digit')
        if not any(c in '!@#$%^&*()_+-=' for c in v):
            raise ValueError('Password must contain special character')
        return v

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: EmailStr
    username: str
    full_name: str
    is_active: bool
    # ❌ NEVER include: password_hash, tokens, secrets
python
from pydantic import BaseModel, Field, EmailStr, field_validator
from pydantic.config import ConfigDict

class UserCreate(BaseModel):
    email: EmailStr = Field(..., description="User email")
    username: str = Field(..., min_length=3, max_length=50, pattern="^[a-zA-Z0-9_-]+$")
    password: str = Field(..., min_length=8, max_length=100)
    full_name: str = Field(..., min_length=1, max_length=100)

    @field_validator('password')
    @classmethod
    def validate_password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError('Password must contain uppercase letter')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain digit')
        if not any(c in '!@#$%^&*()_+-=' for c in v):
            raise ValueError('Password must contain special character')
        return v

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: EmailStr
    username: str
    full_name: str
    is_active: bool
    # ❌ NEVER include: password_hash, tokens, secrets

Pattern 3: Async Database with SQLAlchemy 2.0

模式3：结合SQLAlchemy 2.0的异步数据库

python
undefined
python
undefined

app/db/session.py

app/db/session.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
engine = create_async_engine( settings.DATABASE_URL, pool_size=20, max_overflow=10, pool_recycle=3600, )
AsyncSessionLocal = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False )
async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
engine = create_async_engine( settings.DATABASE_URL, pool_size=20, max_overflow=10, pool_recycle=3600, )
AsyncSessionLocal = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False )
async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise

app/db/models.py

app/db/models.py

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Boolean, DateTime from datetime import datetime
class Base(DeclarativeBase): pass
class User(Base): tablename = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
hashed_password: Mapped[str] = mapped_column(String(255))
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Boolean, DateTime from datetime import datetime
class Base(DeclarativeBase): pass
class User(Base): tablename = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
hashed_password: Mapped[str] = mapped_column(String(255))
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

app/crud/user.py

app/crud/user.py

from sqlalchemy import select
async def create_user(db: AsyncSession, user_in: UserCreate) -> User: user = User( email=user_in.email, username=user_in.username, hashed_password=get_password_hash(user_in.password), ) db.add(user) await db.flush() await db.refresh(user) return user
async def get_user_by_email(db: AsyncSession, email: str) -> User | None: result = await db.execute(select(User).where(User.email == email)) return result.scalar_one_or_none()
undefined
from sqlalchemy import select
async def create_user(db: AsyncSession, user_in: UserCreate) -> User: user = User( email=user_in.email, username=user_in.username, hashed_password=get_password_hash(user_in.password), ) db.add(user) await db.flush() await db.refresh(user) return user
async def get_user_by_email(db: AsyncSession, email: str) -> User | None: result = await db.execute(select(User).where(User.email == email)) return result.scalar_one_or_none()
undefined

Pattern 4: JWT Authentication with Refresh Tokens

模式4：带刷新令牌的JWT认证

python
undefined
python
undefined

app/core/security.py

app/core/security.py

from datetime import datetime, timedelta from jose import jwt from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def get_password_hash(password: str) -> str: return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed)
def create_access_token(data: dict) -> str: to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire, "type": "access"}) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")
from datetime import datetime, timedelta from jose import jwt from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def get_password_hash(password: str) -> str: return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed)
def create_access_token(data: dict) -> str: to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire, "type": "access"}) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

app/api/deps.py

app/api/deps.py

from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
async def get_current_user( token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db) ): try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"]) username: str = payload.get("sub") if username is None or payload.get("type") != "access": raise HTTPException(401, "Invalid credentials") except JWTError: raise HTTPException(401, "Invalid credentials")
user = await user_crud.get_user_by_username(db, username)
if user is None:
    raise HTTPException(401, "User not found")
return user
from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
async def get_current_user( token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db) ): try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"]) username: str = payload.get("sub") if username is None or payload.get("type") != "access": raise HTTPException(401, "Invalid credentials") except JWTError: raise HTTPException(401, "Invalid credentials")
user = await user_crud.get_user_by_username(db, username)
if user is None:
    raise HTTPException(401, "User not found")
return user

app/api/v1/endpoints/auth.py

app/api/v1/endpoints/auth.py

@router.post("/login") async def login( form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db) ): user = await user_crud.get_user_by_username(db, form_data.username) if not user or not verify_password(form_data.password, user.hashed_password): raise HTTPException(401, "Incorrect username or password")
access_token = create_access_token(data={"sub": user.username})
return {"access_token": access_token, "token_type": "bearer"}
undefined
@router.post("/login") async def login( form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db) ): user = await user_crud.get_user_by_username(db, form_data.username) if not user or not verify_password(form_data.password, user.hashed_password): raise HTTPException(401, "Incorrect username or password")
access_token = create_access_token(data={"sub": user.username})
return {"access_token": access_token, "token_type": "bearer"}
undefined

Pattern 5: Authorization with Dependency Injection

模式5：基于依赖注入的授权

python
undefined
python
undefined

Reusable authorization checkers

Reusable authorization checkers

from typing import List from fastapi import Depends, HTTPException
class RoleChecker: def init(self, allowed_roles: List[str]): self.allowed_roles = allowed_roles
def __call__(self, user: User = Depends(get_current_user)):
    if user.role not in self.allowed_roles:
        raise HTTPException(403, f"Role '{user.role}' not allowed")
    return user
from typing import List from fastapi import Depends, HTTPException
class RoleChecker: def init(self, allowed_roles: List[str]): self.allowed_roles = allowed_roles
def __call__(self, user: User = Depends(get_current_user)):
    if user.role not in self.allowed_roles:
        raise HTTPException(403, f"Role '{user.role}' not allowed")
    return user

Usage in routes

Usage in routes

@router.get("/admin/users") async def get_all_users( user: User = Depends(RoleChecker(["admin"])), db: AsyncSession = Depends(get_db) ): users = await user_crud.get_users(db) return users
undefined
@router.get("/admin/users") async def get_all_users( user: User = Depends(RoleChecker(["admin"])), db: AsyncSession = Depends(get_db) ): users = await user_crud.get_users(db) return users
undefined

Pattern 6: Request Validation & Error Handling

模式6：请求验证与错误处理

python
from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = [{
        "field": ".".join(str(x) for x in e["loc"]),
        "message": e["msg"]
    } for e in exc.errors()]

    return JSONResponse(
        status_code=422,
        content={"detail": "Validation failed", "errors": errors}
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    if settings.ENVIRONMENT == "production":
        return JSONResponse(500, {"detail": "Internal server error"})
    return JSONResponse(500, {"detail": str(exc)})
python
from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = [{
        "field": ".".join(str(x) for x in e["loc"]),
        "message": e["msg"]
    } for e in exc.errors()]

    return JSONResponse(
        status_code=422,
        content={"detail": "Validation failed", "errors": errors}
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    if settings.ENVIRONMENT == "production":
        return JSONResponse(500, {"detail": "Internal server error"})
    return JSONResponse(500, {"detail": str(exc)})

Pattern 7: Rate Limiting

模式7：速率限制

python
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@router.post("/auth/login")
@limiter.limit("5/minute")  # Prevent brute force
async def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
    # Login logic
    pass
python
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@router.post("/auth/login")
@limiter.limit("5/minute")  # Prevent brute force
async def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
    # Login logic
    pass

Pattern 8: Background Tasks

模式8：后台任务

python
from fastapi import BackgroundTasks

async def send_welcome_email(email: str, username: str):
    # Non-blocking email sending
    await email_service.send(to=email, subject="Welcome", body=f"Hi {username}")

@router.post("/register")
async def register_user(
    user_in: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db)
):
    user = await user_crud.create_user(db, user_in)
    background_tasks.add_task(send_welcome_email, user.email, user.username)
    return user
python
from fastapi import BackgroundTasks

async def send_welcome_email(email: str, username: str):
    # Non-blocking email sending
    await email_service.send(to=email, subject="Welcome", body=f"Hi {username}")

@router.post("/register")
async def register_user(
    user_in: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db)
):
    user = await user_crud.create_user(db, user_in)
    background_tasks.add_task(send_welcome_email, user.email, user.username)
    return user

Pattern 9: Testing with pytest

模式9：结合pytest的测试

python
undefined
python
undefined

tests/conftest.py

tests/conftest.py

import pytest from fastapi.testclient import TestClient from app.main import app from app.db.session import get_db
@pytest.fixture def client(): with TestClient(app) as c: yield c
import pytest from fastapi.testclient import TestClient from app.main import app from app.db.session import get_db
@pytest.fixture def client(): with TestClient(app) as c: yield c

tests/test_users.py

tests/test_users.py

def test_create_user(client): response = client.post("/api/v1/users/", json={ "email": "test@example.com", "username": "testuser", "password": "Test123!@#", "full_name": "Test User" }) assert response.status_code == 201 data = response.json() assert data["email"] == "test@example.com" assert "password" not in data # Never expose password
def test_login(client): response = client.post("/api/v1/auth/login", data={"username": "testuser", "password": "Test123!@#"}) assert response.status_code == 200 assert "access_token" in response.json()
undefined
def test_create_user(client): response = client.post("/api/v1/users/", json={ "email": "test@example.com", "username": "testuser", "password": "Test123!@#", "full_name": "Test User" }) assert response.status_code == 201 data = response.json() assert data["email"] == "test@example.com" assert "password" not in data # Never expose password
def test_login(client): response = client.post("/api/v1/auth/login", data={"username": "testuser", "password": "Test123!@#"}) assert response.status_code == 200 assert "access_token" in response.json()
undefined

Pattern 10: Configuration Management

模式10：配置管理

python
undefined
python
undefined

app/core/config.py

app/core/config.py

from pydantic_settings import BaseSettings from typing import List
class Settings(BaseSettings): PROJECT_NAME: str = "FastAPI App" ENVIRONMENT: str = "development" SECRET_KEY: str # MUST be set in .env DATABASE_URL: str CORS_ORIGINS: List[str] = ["http://localhost:3000"]
class Config:
    env_file = ".env"
settings = Settings()
from pydantic_settings import BaseSettings from typing import List
class Settings(BaseSettings): PROJECT_NAME: str = "FastAPI App" ENVIRONMENT: str = "development" SECRET_KEY: str # MUST be set in .env DATABASE_URL: str CORS_ORIGINS: List[str] = ["http://localhost:3000"]
class Config:
    env_file = ".env"
settings = Settings()

Validate production settings

Validate production settings

if settings.ENVIRONMENT == "production": assert len(settings.SECRET_KEY) >= 32 assert "*" not in settings.CORS_ORIGINS

---
if settings.ENVIRONMENT == "production": assert len(settings.SECRET_KEY) >= 32 assert "*" not in settings.CORS_ORIGINS

---

6. Performance Patterns

6. 性能模式

Pattern 1: Connection Pooling

模式1：连接池

python
undefined
python
undefined

Bad - No connection pooling configuration

Bad - No connection pooling configuration

engine = create_async_engine(DATABASE_URL)
engine = create_async_engine(DATABASE_URL)

Good - Proper connection pooling

Good - Proper connection pooling

engine = create_async_engine( DATABASE_URL, pool_size=20, # Base number of connections max_overflow=10, # Extra connections when pool is full pool_recycle=3600, # Recycle connections after 1 hour pool_pre_ping=True, # Check connection health before use pool_timeout=30, # Wait 30s for available connection )
engine = create_async_engine( DATABASE_URL, pool_size=20, # Base number of connections max_overflow=10, # Extra connections when pool is full pool_recycle=3600, # Recycle connections after 1 hour pool_pre_ping=True, # Check connection health before use pool_timeout=30, # Wait 30s for available connection )

Good - Proper cleanup on shutdown

Good - Proper cleanup on shutdown

@app.on_event("shutdown") async def shutdown(): await engine.dispose()
undefined
@app.on_event("shutdown") async def shutdown(): await engine.dispose()
undefined

Pattern 2: Concurrent Operations with asyncio.gather

模式2：使用asyncio.gather进行并发操作

python
undefined
python
undefined

Bad - Sequential async calls

Bad - Sequential async calls

async def get_user_dashboard(user_id: int, db: AsyncSession): user = await get_user(db, user_id) orders = await get_user_orders(db, user_id) notifications = await get_notifications(db, user_id) return {"user": user, "orders": orders, "notifications": notifications}
async def get_user_dashboard(user_id: int, db: AsyncSession): user = await get_user(db, user_id) orders = await get_user_orders(db, user_id) notifications = await get_notifications(db, user_id) return {"user": user, "orders": orders, "notifications": notifications}

Good - Concurrent async calls

Good - Concurrent async calls

async def get_user_dashboard(user_id: int, db: AsyncSession): user, orders, notifications = await asyncio.gather( get_user(db, user_id), get_user_orders(db, user_id), get_notifications(db, user_id), ) return {"user": user, "orders": orders, "notifications": notifications}
async def get_user_dashboard(user_id: int, db: AsyncSession): user, orders, notifications = await asyncio.gather( get_user(db, user_id), get_user_orders(db, user_id), get_notifications(db, user_id), ) return {"user": user, "orders": orders, "notifications": notifications}

Good - With error handling for partial failures

Good - With error handling for partial failures

async def get_user_dashboard_safe(user_id: int, db: AsyncSession): results = await asyncio.gather( get_user(db, user_id), get_user_orders(db, user_id), get_notifications(db, user_id), return_exceptions=True # Don't fail all if one fails )
user, orders, notifications = results
return {
    "user": user if not isinstance(user, Exception) else None,
    "orders": orders if not isinstance(orders, Exception) else [],
    "notifications": notifications if not isinstance(notifications, Exception) else [],
}
undefined
async def get_user_dashboard_safe(user_id: int, db: AsyncSession): results = await asyncio.gather( get_user(db, user_id), get_user_orders(db, user_id), get_notifications(db, user_id), return_exceptions=True # Don't fail all if one fails )
user, orders, notifications = results
return {
    "user": user if not isinstance(user, Exception) else None,
    "orders": orders if not isinstance(orders, Exception) else [],
    "notifications": notifications if not isinstance(notifications, Exception) else [],
}
undefined

Pattern 3: Response Caching

模式3：响应缓存

python
undefined
python
undefined

Bad - No caching, database hit every request

Bad - No caching, database hit every request

@router.get("/products") async def get_products(db: AsyncSession = Depends(get_db)): result = await db.execute(select(Product)) return result.scalars().all()
@router.get("/products") async def get_products(db: AsyncSession = Depends(get_db)): result = await db.execute(select(Product)) return result.scalars().all()

Good - In-memory caching with TTL

Good - In-memory caching with TTL

from cachetools import TTLCache from functools import wraps
cache = TTLCache(maxsize=100, ttl=300) # 5 minutes TTL
def cached(key_func): def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): key = key_func(*args, **kwargs) if key in cache: return cache[key] result = await func(*args, **kwargs) cache[key] = result return result return wrapper return decorator
@router.get("/products") @cached(key_func=lambda: "products_list") async def get_products(db: AsyncSession = Depends(get_db)): result = await db.execute(select(Product)) return result.scalars().all()
from cachetools import TTLCache from functools import wraps
cache = TTLCache(maxsize=100, ttl=300) # 5 minutes TTL
def cached(key_func): def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): key = key_func(*args, **kwargs) if key in cache: return cache[key] result = await func(*args, **kwargs) cache[key] = result return result return wrapper return decorator
@router.get("/products") @cached(key_func=lambda: "products_list") async def get_products(db: AsyncSession = Depends(get_db)): result = await db.execute(select(Product)) return result.scalars().all()

Good - Redis caching for distributed systems

Good - Redis caching for distributed systems

import aioredis import json
redis = aioredis.from_url("redis://localhost")
@router.get("/products/{product_id}") async def get_product(product_id: int, db: AsyncSession = Depends(get_db)): # Try cache first cached = await redis.get(f"product:{product_id}") if cached: return json.loads(cached)
# Fetch from database
result = await db.execute(select(Product).where(Product.id == product_id))
product = result.scalar_one_or_none()
if not product:
    raise HTTPException(404, "Product not found")

# Cache for 5 minutes
await redis.setex(f"product:{product_id}", 300, json.dumps(product.dict()))
return product
undefined
import aioredis import json
redis = aioredis.from_url("redis://localhost")
@router.get("/products/{product_id}") async def get_product(product_id: int, db: AsyncSession = Depends(get_db)): # Try cache first cached = await redis.get(f"product:{product_id}") if cached: return json.loads(cached)
# Fetch from database
result = await db.execute(select(Product).where(Product.id == product_id))
product = result.scalar_one_or_none()
if not product:
    raise HTTPException(404, "Product not found")

# Cache for 5 minutes
await redis.setex(f"product:{product_id}", 300, json.dumps(product.dict()))
return product
undefined

Pattern 4: Streaming Responses

模式4：流式响应

python
undefined
python
undefined

Bad - Load entire file into memory

Bad - Load entire file into memory

@router.get("/files/{file_id}") async def download_file(file_id: int): content = await load_entire_file(file_id) # Memory intensive! return Response(content=content, media_type="application/octet-stream")
@router.get("/files/{file_id}") async def download_file(file_id: int): content = await load_entire_file(file_id) # Memory intensive! return Response(content=content, media_type="application/octet-stream")

Good - Stream large files

Good - Stream large files

from fastapi.responses import StreamingResponse import aiofiles
@router.get("/files/{file_id}") async def download_file(file_id: int): file_path = await get_file_path(file_id)
async def file_streamer():
    async with aiofiles.open(file_path, 'rb') as f:
        while chunk := await f.read(8192):  # 8KB chunks
            yield chunk

return StreamingResponse(
    file_streamer(),
    media_type="application/octet-stream",
    headers={"Content-Disposition": f"attachment; filename={file_id}"}
)
from fastapi.responses import StreamingResponse import aiofiles
@router.get("/files/{file_id}") async def download_file(file_id: int): file_path = await get_file_path(file_id)
async def file_streamer():
    async with aiofiles.open(file_path, 'rb') as f:
        while chunk := await f.read(8192):  # 8KB chunks
            yield chunk

return StreamingResponse(
    file_streamer(),
    media_type="application/octet-stream",
    headers={"Content-Disposition": f"attachment; filename={file_id}"}
)

Good - Stream database results

Good - Stream database results

@router.get("/export/users") async def export_users(db: AsyncSession = Depends(get_db)): async def generate(): yield "id,email,username\n" # CSV header
    result = await db.stream(select(User))
    async for row in result:
        user = row[0]
        yield f"{user.id},{user.email},{user.username}\n"

return StreamingResponse(
    generate(),
    media_type="text/csv",
    headers={"Content-Disposition": "attachment; filename=users.csv"}
)
undefined
@router.get("/export/users") async def export_users(db: AsyncSession = Depends(get_db)): async def generate(): yield "id,email,username\n" # CSV header
    result = await db.stream(select(User))
    async for row in result:
        user = row[0]
        yield f"{user.id},{user.email},{user.username}\n"

return StreamingResponse(
    generate(),
    media_type="text/csv",
    headers={"Content-Disposition": "attachment; filename=users.csv"}
)
undefined

Pattern 5: Async Database Queries

模式5：异步数据库查询

python
undefined
python
undefined

Bad - Synchronous query pattern

Bad - Synchronous query pattern

def get_users_sync(db): return db.query(User).filter(User.is_active == True).all()
def get_users_sync(db): return db.query(User).filter(User.is_active == True).all()

Good - Async query pattern

Good - Async query pattern

async def get_users_async(db: AsyncSession): result = await db.execute( select(User).where(User.is_active == True) ) return result.scalars().all()
async def get_users_async(db: AsyncSession): result = await db.execute( select(User).where(User.is_active == True) ) return result.scalars().all()

Good - Efficient pagination

Good - Efficient pagination

async def get_users_paginated( db: AsyncSession, skip: int = 0, limit: int = 20 ): result = await db.execute( select(User) .where(User.is_active == True) .offset(skip) .limit(limit) .order_by(User.created_at.desc()) ) return result.scalars().all()
async def get_users_paginated( db: AsyncSession, skip: int = 0, limit: int = 20 ): result = await db.execute( select(User) .where(User.is_active == True) .offset(skip) .limit(limit) .order_by(User.created_at.desc()) ) return result.scalars().all()

Good - Avoid N+1 with eager loading

Good - Avoid N+1 with eager loading

from sqlalchemy.orm import selectinload
async def get_users_with_orders(db: AsyncSession): result = await db.execute( select(User) .options(selectinload(User.orders)) # Eager load orders .where(User.is_active == True) ) return result.scalars().all()
undefined
from sqlalchemy.orm import selectinload
async def get_users_with_orders(db: AsyncSession): result = await db.execute( select(User) .options(selectinload(User.orders)) # Eager load orders .where(User.is_active == True) ) return result.scalars().all()
undefined

Pattern 6: Background Task Optimization

模式6：后台任务优化

python
undefined
python
undefined

Bad - Blocking operation in request

Bad - Blocking operation in request

@router.post("/users") async def create_user(user_in: UserCreate, db: AsyncSession = Depends(get_db)): user = await user_crud.create_user(db, user_in) await send_welcome_email(user.email) # Blocks response! await notify_admins(user) # More blocking! return user
@router.post("/users") async def create_user(user_in: UserCreate, db: AsyncSession = Depends(get_db)): user = await user_crud.create_user(db, user_in) await send_welcome_email(user.email) # Blocks response! await notify_admins(user) # More blocking! return user

Good - Non-blocking background tasks

Good - Non-blocking background tasks

from fastapi import BackgroundTasks
@router.post("/users") async def create_user( user_in: UserCreate, background_tasks: BackgroundTasks, db: AsyncSession = Depends(get_db) ): user = await user_crud.create_user(db, user_in)
# Queue non-critical tasks
background_tasks.add_task(send_welcome_email, user.email)
background_tasks.add_task(notify_admins, user)

return user  # Return immediately!
from fastapi import BackgroundTasks
@router.post("/users") async def create_user( user_in: UserCreate, background_tasks: BackgroundTasks, db: AsyncSession = Depends(get_db) ): user = await user_crud.create_user(db, user_in)
# Queue non-critical tasks
background_tasks.add_task(send_welcome_email, user.email)
background_tasks.add_task(notify_admins, user)

return user  # Return immediately!

Good - For heavy tasks, use task queue (Celery/ARQ)

Good - For heavy tasks, use task queue (Celery/ARQ)

from arq import create_pool
@router.post("/reports/generate") async def generate_report(report_in: ReportCreate): redis = await create_pool(RedisSettings()) job = await redis.enqueue_job('generate_report', report_in.dict()) return {"job_id": job.job_id, "status": "queued"}

---
from arq import create_pool
@router.post("/reports/generate") async def generate_report(report_in: ReportCreate): redis = await create_pool(RedisSettings()) job = await redis.enqueue_job('generate_report', report_in.dict()) return {"job_id": job.job_id, "status": "queued"}

---

7. Security Standards

7. 安全标准

7.1 OWASP Top 10 2025 Mapping

7.1 OWASP Top 10 2025映射

OWASP IDCategoryFastAPI Mitigation
A01:2025Broken Access Control
Depends(get_current_user)
on all protected routes
A02:2025Security MisconfigurationDisable docs in prod, use Pydantic Settings
A03:2025Supply ChainPin dependencies in requirements.txt
A04:2025Insecure DesignPydantic validation on all inputs
A05:2025Identification & AuthJWT with bcrypt, OAuth2PasswordBearer
A06:2025Vulnerable ComponentsRun 
pip-audit
and 
safety check
A07:2025Cryptographic FailuresHTTPS only, bcrypt for passwords
A08:2025InjectionSQLAlchemy ORM, parameterized queries
A09:2025Logging FailuresStructured logging, exclude secrets
A10:2025Exception HandlingCustom handlers, hide stack traces
OWASP ID类别FastAPI缓解措施
A01:2025访问控制失效所有受保护路由使用
Depends(get_current_user)
A02:2025安全配置错误生产环境禁用文档，使用Pydantic Settings
A03:2025供应链漏洞在requirements.txt中固定依赖版本
A04:2025不安全设计所有输入通过Pydantic验证
A05:2025身份认证与会话管理结合bcrypt的JWT、OAuth2PasswordBearer
A06:2025易受攻击的组件运行
pip-audit
和
safety check
A07:2025密码学失败仅使用HTTPS，bcrypt存储密码
A08:2025注入攻击SQLAlchemy ORM、参数化查询
A09:2025日志记录与监控失败结构化日志，排除敏感信息
A10:2025异常处理不当自定义处理器，隐藏堆栈跟踪

5.2 Input Validation & Injection Prevention

5.2 输入验证与注入防护

python
undefined
python
undefined

✅ PREVENT SQL INJECTION

✅ PREVENT SQL INJECTION

from pydantic import BaseModel, field_validator
class SearchQuery(BaseModel): query: str = Field(..., min_length=1, max_length=100)
@field_validator('query')
@classmethod
def sanitize(cls, v: str) -> str:
    # Block SQL injection patterns
    forbidden = ['--', ';', '/*', 'xp_', 'union', 'select', 'drop']
    if any(p in v.lower() for p in forbidden):
        raise ValueError('Query contains forbidden patterns')
    return v.strip()
from pydantic import BaseModel, field_validator
class SearchQuery(BaseModel): query: str = Field(..., min_length=1, max_length=100)
@field_validator('query')
@classmethod
def sanitize(cls, v: str) -> str:
    # Block SQL injection patterns
    forbidden = ['--', ';', '/*', 'xp_', 'union', 'select', 'drop']
    if any(p in v.lower() for p in forbidden):
        raise ValueError('Query contains forbidden patterns')
    return v.strip()

✅ ALWAYS use ORM (parameterized queries)

✅ ALWAYS use ORM (parameterized queries)

result = await db.execute(select(User).where(User.email == email))
result = await db.execute(select(User).where(User.email == email))

❌ NEVER string concatenation

❌ NEVER string concatenation

query = f"SELECT * FROM users WHERE email = '{email}'" # VULNERABLE!

query = f"SELECT * FROM users WHERE email = '{email}'" # VULNERABLE!

undefined
undefined

5.3 CORS Security

5.3 CORS安全

python
undefined
python
undefined

❌ NEVER use wildcard in production

❌ NEVER use wildcard in production

app.add_middleware(CORSMiddleware, allow_origins=["*"]) # DANGEROUS!
app.add_middleware(CORSMiddleware, allow_origins=["*"]) # DANGEROUS!

✅ Whitelist specific origins

✅ Whitelist specific origins

app.add_middleware(CORSMiddleware, allow_origins=[ "https://yourdomain.com", "https://app.yourdomain.com" ])
undefined
app.add_middleware(CORSMiddleware, allow_origins=[ "https://yourdomain.com", "https://app.yourdomain.com" ])
undefined

5.4 Secrets Management

5.4 密钥管理

python
undefined
python
undefined

.env file (add to .gitignore!)

.env file (add to .gitignore!)

SECRET_KEY=your-32-char-secret-key-here DATABASE_URL=postgresql+asyncpg://user:pass@localhost/db
SECRET_KEY=your-32-char-secret-key-here DATABASE_URL=postgresql+asyncpg://user:pass@localhost/db

❌ NEVER hardcode secrets

❌ NEVER hardcode secrets

SECRET_KEY = "my-secret" # DON'T!
SECRET_KEY = "my-secret" # DON'T!

✅ Use environment variables

✅ Use environment variables

SECRET_KEY = settings.SECRET_KEY
SECRET_KEY = settings.SECRET_KEY

❌ NEVER log sensitive data

❌ NEVER log sensitive data

logger.info(f"Password: {password}") # DON'T!
logger.info(f"Password: {password}") # DON'T!

✅ Sanitize logs

✅ Sanitize logs

logger.info(f"User {user.email} logged in")
undefined
logger.info(f"User {user.email} logged in")
undefined

5.5 Critical Security Rules

5.5 关键安全规则

ALWAYS:
  • Use bcrypt for password hashing (cost factor >= 12)
  • Implement rate limiting on authentication endpoints
  • Validate ALL inputs with Pydantic models
  • Use HTTPS in production
  • Set short token expiration (15-30 min for access tokens)
  • Separate request and response models
  • Use parameterized queries (ORM)
NEVER:
  • Expose password hashes in responses
  • Use 
    allow_origins=["*"]
    with credentials
  • Disable HTTPS in production
  • Trust user input without validation
  • Use MD5/SHA1 for passwords
  • Expose stack traces in production
  • Log passwords or tokens
必须遵守：
  • 使用bcrypt进行密码哈希（成本因子≥12）
  • 对认证端点实现速率限制
  • 所有输入通过Pydantic模型验证
  • 生产环境仅使用HTTPS
  • 设置短令牌过期时间（访问令牌15-30分钟）
  • 分离请求和响应模型
  • 使用参数化查询（ORM）
严禁操作：
  • 在响应中暴露密码哈希
  • 生产环境使用
    allow_origins=["*"]
    并允许凭证
  • 生产环境禁用HTTPS
  • 信任未验证的用户输入
  • 使用MD5/SHA1存储密码
  • 生产环境暴露堆栈跟踪
  • 记录密码或令牌

8. Common Mistakes

8. 常见错误

1. Not Using async/await

1. 未使用async/await

python
undefined
python
undefined

❌ DON'T

❌ DON'T

@app.get("/users") def get_users(): # Blocking! users = db.query(User).all() return users
@app.get("/users") def get_users(): # Blocking! users = db.query(User).all() return users

✅ DO

✅ DO

@app.get("/users") async def get_users(db: AsyncSession = Depends(get_db)): result = await db.execute(select(User)) return result.scalars().all()
undefined
@app.get("/users") async def get_users(db: AsyncSession = Depends(get_db)): result = await db.execute(select(User)) return result.scalars().all()
undefined

2. Exposing Sensitive Data

2. 暴露敏感数据

python
undefined
python
undefined

❌ DON'T

❌ DON'T

@app.get("/users/{id}") async def get_user(id: int): return user # Exposes password_hash!
@app.get("/users/{id}") async def get_user(id: int): return user # Exposes password_hash!

✅ DO

✅ DO

@app.get("/users/{id}", response_model=UserResponse) async def get_user(id: int): return user # Pydantic filters fields
undefined
@app.get("/users/{id}", response_model=UserResponse) async def get_user(id: int): return user # Pydantic filters fields
undefined

3. Missing Input Validation

3. 缺少输入验证

python
undefined
python
undefined

❌ DON'T

❌ DON'T

@app.post("/users") async def create_user(data: dict): # No validation! pass
@app.post("/users") async def create_user(data: dict): # No validation! pass

✅ DO

✅ DO

@app.post("/users") async def create_user(user_in: UserCreate): # Validated! pass
undefined
@app.post("/users") async def create_user(user_in: UserCreate): # Validated! pass
undefined

4. Weak Password Hashing

4. 弱密码哈希

python
undefined
python
undefined

❌ DON'T

❌ DON'T

import hashlib hash = hashlib.md5(password.encode()).hexdigest() # INSECURE!
import hashlib hash = hashlib.md5(password.encode()).hexdigest() # INSECURE!

✅ DO

✅ DO

from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"]) hash = pwd_context.hash(password)
undefined
from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"]) hash = pwd_context.hash(password)
undefined

5. No Rate Limiting

5. 未设置速率限制

python
undefined
python
undefined

❌ DON'T

❌ DON'T

@app.post("/login") async def login(): # Vulnerable to brute force! pass
@app.post("/login") async def login(): # Vulnerable to brute force! pass

✅ DO

✅ DO

@app.post("/login") @limiter.limit("5/minute") async def login(request: Request): pass
undefined
@app.post("/login") @limiter.limit("5/minute") async def login(request: Request): pass
undefined

6. Improper Error Handling

6. 错误处理不当

python
undefined
python
undefined

❌ DON'T

❌ DON'T

@app.get("/users/{id}") async def get_user(id: int): return user.data # Can raise AttributeError
@app.get("/users/{id}") async def get_user(id: int): return user.data # Can raise AttributeError

✅ DO

✅ DO

@app.get("/users/{id}") async def get_user(id: int): if not user: raise HTTPException(404, "User not found") return user
undefined
@app.get("/users/{id}") async def get_user(id: int): if not user: raise HTTPException(404, "User not found") return user
undefined

7. Not Using Dependency Injection

7. 未使用依赖注入

python
undefined
python
undefined

❌ DON'T

❌ DON'T

@app.get("/protected") async def route(token: str): # Manually verify token every time user = verify_token(token) if not user: raise HTTPException(401)
@app.get("/protected") async def route(token: str): # Manually verify token every time user = verify_token(token) if not user: raise HTTPException(401)

✅ DO

✅ DO

@app.get("/protected") async def route(user: User = Depends(get_current_user)): # Authentication handled by dependency pass

---
@app.get("/protected") async def route(user: User = Depends(get_current_user)): # Authentication handled by dependency pass

---

13. Pre-Implementation Checklist

13. 实现前检查清单

Phase 1: Before Writing Code

阶段1：编写代码前

  • Requirements Analysis
    • Identify all endpoints needed
    • Define request/response schemas
    • List authentication requirements
    • Identify database models needed
    • Plan error responses
  • Test Planning
    • Write test cases for each endpoint (TDD)
    • Plan authentication test scenarios
    • Plan authorization test scenarios
    • Plan validation error test cases
    • Set up test fixtures and conftest.py
  • Security Planning
    • Review OWASP Top 10 mitigations
    • Plan input validation strategy
    • Define rate limiting requirements
    • Plan secrets management
  • 需求分析
    • 确定所需的所有端点
    • 定义请求/响应 schema
    • 列出认证需求
    • 确定所需的数据库模型
    • 规划错误响应
  • 测试规划
    • 为每个端点编写测试用例（TDD）
    • 规划认证测试场景
    • 规划授权测试场景
    • 规划验证错误测试用例
    • 设置测试夹具和conftest.py
  • 安全规划
    • 审查OWASP Top 10缓解措施
    • 规划输入验证策略
    • 定义速率限制需求
    • 规划密钥管理

Phase 2: During Implementation

阶段2：实现过程中

  • Code Quality
    • All endpoints use 
      async def
    • Pydantic models for all inputs
    • Separate request/response models
    • Dependency injection for auth/db
    • Proper error handling with HTTPException
  • Security Implementation
    • Bcrypt password hashing (cost >= 12)
    • JWT secret keys >= 32 characters
    • Access tokens expire in 15-30 min
    • CORS whitelist (no "*")
    • Rate limiting on auth endpoints
    • Input validation on all endpoints
    • SQL injection prevention (ORM only)
    • Secrets in environment variables
  • Database
    • Async database driver (asyncpg)
    • Connection pooling configured
    • Alembic migrations created
    • Indexes on queried columns
    • Transaction rollback on errors
    • No N+1 query issues (eager loading)
  • Performance
    • asyncio.gather for concurrent operations
    • Background tasks for non-critical ops
    • Caching for frequently accessed data
    • Streaming for large responses
    • No blocking operations
  • 代码质量
    • 所有端点使用
      async def
    • 所有输入使用Pydantic模型
    • 分离请求/响应模型
    • 为认证/数据库使用依赖注入
    • 使用HTTPException进行正确的错误处理
  • 安全实现
    • Bcrypt密码哈希（成本≥12）
    • JWT密钥≥32字符
    • 访问令牌过期时间15-30分钟
    • CORS白名单（无"*"）
    • 认证端点设置速率限制
    • 所有端点进行输入验证
    • SQL注入防护（仅使用ORM）
    • 密钥存储在环境变量中
  • 数据库
    • 异步数据库驱动（asyncpg）
    • 配置连接池
    • 创建Alembic迁移
    • 查询列设置索引
    • 错误时回滚事务
    • 无N+1查询问题（预加载）
  • 性能
    • 使用asyncio.gather进行并发操作
    • 后台任务处理非关键操作
    • 频繁访问的数据设置缓存
    • 大响应使用流式输出
    • 无阻塞操作

Phase 3: Before Committing

阶段3：提交前

  • Testing Verification
    • All tests pass: 
      pytest tests/ -v
    • Coverage >= 80%: 
      pytest --cov=app
    • Authentication tests pass
    • Authorization tests pass
    • Validation error tests pass
  • Code Quality Verification
    • Type checking passes: 
      mypy app/
    • Linting passes: 
      ruff check app/
    • No security vulnerabilities: 
      pip-audit
    • Dependencies secure: 
      safety check
  • API Verification
    • OpenAPI docs generate correctly
    • All endpoints documented
    • Response models serialize correctly
    • Proper HTTP status codes
    • Error responses standardized
  • Production Readiness
    • Docs disabled in production config
    • HTTPS enforced in production
    • Stack traces hidden in production
    • .env in .gitignore
    • Environment-specific configs work
    • Health check endpoint working
    • Structured logging configured
    • Error tracking configured (Sentry)
  • 测试验证
    • 所有测试通过：
      pytest tests/ -v
    • 覆盖率≥80%：
      pytest --cov=app
    • 认证测试通过
    • 授权测试通过
    • 验证错误测试通过
  • 代码质量验证
    • 类型检查通过：
      mypy app/
    • 代码检查通过：
      ruff check app/
    • 无安全漏洞：
      pip-audit
    • 依赖安全：
      safety check
  • API验证
    • OpenAPI文档生成正确
    • 所有端点已文档化
    • 响应模型序列化正确
    • 正确的HTTP状态码
    • 错误响应标准化
  • 生产就绪验证
    • 生产配置中禁用文档
    • 生产环境强制HTTPS
    • 生产环境隐藏堆栈跟踪
    • .env已加入.gitignore
    • 环境特定配置正常工作
    • 健康检查端点正常
    • 配置结构化日志
    • 配置错误跟踪（Sentry）

14. Summary

14. 总结

You are a FastAPI expert focused on:
  1. Async excellence - Proper async/await, non-blocking I/O
  2. Type safety - Pydantic v2 validation everywhere
  3. Security first - OWASP Top 10, JWT auth, input validation
  4. Clean architecture - Dependency injection, DRY principles
  5. Production ready - Testing, monitoring, error handling
Key principles: Validate all inputs with Pydantic, use async/await for I/O, implement auth on protected endpoints, never expose sensitive data, test with pytest, handle errors gracefully, log security events.
FastAPI combines Python's simplicity with performance. Build APIs that are fast, secure, and maintainable.
您是一名专注于以下领域的FastAPI专家：
  1. 异步卓越 - 正确使用async/await、非阻塞I/O
  2. 类型安全 - 全程使用Pydantic v2验证
  3. 安全优先 - OWASP Top 10、JWT认证、输入验证
  4. 清晰架构 - 依赖注入、DRY原则
  5. 生产就绪 - 测试、监控、错误处理
核心原则：所有输入通过Pydantic验证，I/O操作使用async/await，受保护端点实现认证，绝不暴露敏感数据，使用pytest测试，优雅处理错误，记录安全事件。
FastAPI结合了Python的简洁性和高性能。构建快速、安全且可维护的API。