FastAPI 教程：2026年用 Python 构建 REST API

FastAPI 已成为构建 API 最流行的 Python Web 框架。它结合了 Flask 的简洁性和 Node.js/Go 的性能，自动交互式文档和基于 Python 类型提示的内置数据验证。本教程涵盖从项目设置到认证、数据库集成和部署的所有内容。

为什么选择 FastAPI？

FastAPI 提供了开发体验和性能的独特组合。

• Automatic interactive API documentation (Swagger UI + ReDoc)
• Data validation and serialization with Pydantic
• Native async/await support for high concurrency
• Type-safe development with Python type hints
• 40% fewer bugs through automatic validation (per FastAPI benchmarks)
• Performance on par with Node.js and Go frameworks

FastAPI vs Flask vs Django REST Framework

项目设置

使用正确的项目结构、虚拟环境和依赖设置新的 FastAPI 项目。

# Create project directory
mkdir fastapi-project && cd fastapi-project

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Linux/macOS
# venv\Scripts\activate   # Windows

# Install dependencies
pip install fastapi uvicorn[standard] pydantic-settings sqlalchemy[asyncio] asyncpg python-jose[cryptography] passlib[bcrypt] python-multipart httpx pytest

# Project structure
fastapi-project/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI app instance
│   ├── config.py          # Settings with pydantic-settings
│   ├── database.py        # Database connection
│   ├── models/            # SQLAlchemy models
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/           # Pydantic schemas
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/           # API route handlers
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── auth.py
│   ├── services/          # Business logic
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── dependencies.py    # Shared dependencies
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_users.py
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

你的第一个 API

创建基本的 FastAPI 应用来了解基础知识。

# app/main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize resources
    print("Starting up...")
    yield
    # Shutdown: cleanup resources
    print("Shutting down...")

app = FastAPI(
    title="My API",
    description="A production-ready REST API built with FastAPI",
    version="1.0.0",
    lifespan=lifespan,
)

@app.get("/")
async def root():
    return {"message": "Hello, FastAPI!"}

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# Run with:
# uvicorn app.main:app --reload --port 8000
#
# API docs available at:
# http://localhost:8000/docs      (Swagger UI)
# http://localhost:8000/redoc     (ReDoc)

路由和路径参数

FastAPI 支持路径参数、查询参数和请求体的自动验证。

路径参数

from fastapi import FastAPI, Path, HTTPException

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int = Path(..., title="User ID", ge=1, description="The ID of the user")
):
    # user_id is automatically validated as a positive integer
    if user_id > 1000:
        raise HTTPException(status_code=404, detail="User not found")
    return {"user_id": user_id, "name": f"User {user_id}"}

# Enum path parameters
from enum import Enum

class UserRole(str, Enum):
    admin = "admin"
    editor = "editor"
    viewer = "viewer"

@app.get("/users/role/{role}")
async def get_users_by_role(role: UserRole):
    return {"role": role, "message": f"Listing all {role.value} users"}

查询参数

from fastapi import Query
from typing import Optional

@app.get("/users")
async def list_users(
    skip: int = Query(default=0, ge=0, description="Number of records to skip"),
    limit: int = Query(default=20, ge=1, le=100, description="Max records to return"),
    search: Optional[str] = Query(default=None, min_length=1, max_length=100),
    sort_by: str = Query(default="created_at", pattern="^(name|email|created_at)$"),
    active: bool = Query(default=True),
):
    return {
        "skip": skip,
        "limit": limit,
        "search": search,
        "sort_by": sort_by,
        "active": active,
    }

使用 Pydantic 模型的请求体

Pydantic 模型提供自动请求验证、序列化和文档。

# app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=100, examples=["John Doe"])
    email: EmailStr = Field(..., examples=["john@example.com"])
    password: str = Field(..., min_length=8, max_length=128)
    age: Optional[int] = Field(default=None, ge=13, le=150)
    bio: Optional[str] = Field(default=None, max_length=500)

    model_config = {
        "json_schema_extra": {
            "examples": [{
                "name": "John Doe",
                "email": "john@example.com",
                "password": "securePass123",
                "age": 28,
                "bio": "Software developer"
            }]
        }
    }

class UserResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    age: Optional[int] = None
    bio: Optional[str] = None
    created_at: datetime
    # Note: password is NOT included in the response

    model_config = {"from_attributes": True}

class UserUpdate(BaseModel):
    name: Optional[str] = Field(default=None, min_length=2, max_length=100)
    bio: Optional[str] = Field(default=None, max_length=500)
    age: Optional[int] = Field(default=None, ge=13, le=150)

# Usage in route
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # Pydantic automatically validates the request body
    # Invalid data returns a 422 with detailed error messages
    return {"id": 1, **user.model_dump(), "created_at": datetime.now()}

响应模型

使用响应模型控制返回给客户端的数据。

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

class UserListResponse(BaseModel):
    users: List[UserResponse]
    total: int
    page: int
    per_page: int

@app.get("/users", response_model=UserListResponse)
async def list_users(page: int = 1, per_page: int = 20):
    # response_model ensures only declared fields are returned
    # Even if the database returns extra fields (like password_hash),
    # they will be filtered out automatically
    return {
        "users": [],
        "total": 0,
        "page": page,
        "per_page": per_page,
    }

高级验证

Pydantic 和 FastAPI 提供丰富的验证功能。

from pydantic import BaseModel, field_validator, model_validator
from typing import Optional
import re

class UserCreate(BaseModel):
    username: str
    email: str
    password: str
    confirm_password: str

    @field_validator('username')
    @classmethod
    def username_must_be_alphanumeric(cls, v: str) -> str:
        if not re.match(r'^[a-zA-Z0-9_]+$', v):
            raise ValueError('Username must be alphanumeric')
        if len(v) < 3:
            raise ValueError('Username must be at least 3 characters')
        return v.lower()

    @field_validator('password')
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain an uppercase letter')
        if not re.search(r'[0-9]', v):
            raise ValueError('Password must contain a number')
        return v

    @model_validator(mode='after')
    def passwords_match(self):
        if self.password != self.confirm_password:
            raise ValueError('Passwords do not match')
        return self

错误处理

FastAPI 提供内置的异常处理器。

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# Custom exception class
class AppException(Exception):
    def __init__(self, status_code: int, detail: str, error_code: str):
        self.status_code = status_code
        self.detail = detail
        self.error_code = error_code

# Register custom exception handler
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.error_code,
            "detail": exc.detail,
            "path": str(request.url),
        },
    )

# Usage
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await find_user(user_id)
    if not user:
        raise AppException(
            status_code=404,
            detail=f"User with ID {user_id} not found",
            error_code="USER_NOT_FOUND",
        )
    return user

数据库集成

使用 SQLAlchemy 连接 FastAPI 到数据库。

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

engine = create_async_engine(DATABASE_URL, echo=True, pool_size=20, max_overflow=10)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# Dependency for database sessions
async def get_db():
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# app/models/user.py
from sqlalchemy import Column, Integer, String, DateTime, func
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False, index=True)
    password_hash = Column(String(255), nullable=False)
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

完整 CRUD API

构建完整的增删改查 API。

# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserResponse, UserUpdate
from typing import List

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

@router.get("/", response_model=List[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 20,
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User).offset(skip).limit(limit))
    users = result.scalars().all()
    return users

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(data: UserCreate, db: AsyncSession = Depends(get_db)):
    user = User(name=data.name, email=data.email, password_hash=hash_password(data.password))
    db.add(user)
    await db.flush()
    await db.refresh(user)
    return user

@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, data: UserUpdate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(user, field, value)
    await db.flush()
    return user

@router.delete("/{user_id}", status_code=204)
async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    await db.delete(user)

# Register router in main.py
# app.include_router(router)

认证和授权

实现基于 JWT 的认证。

# app/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel

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

SECRET_KEY = "your-secret-key"  # Use environment variable in production
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

class Token(BaseModel):
    access_token: str
    token_type: str

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    # Fetch user from database
    user = await find_user_by_id(int(user_id))
    if user is None:
        raise credentials_exception
    return user

@router.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = await authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
        )
    access_token = create_access_token(
        data={"sub": str(user.id)},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return {"access_token": access_token, "token_type": "bearer"}

# Protected endpoint
@router.get("/me", response_model=UserResponse)
async def get_current_user_profile(current_user = Depends(get_current_user)):
    return current_user

中间件

添加日志、CORS、请求计时等中间件。

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time, logging

app = FastAPI()

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# GZip compression
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Custom request timing middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.4f}"
    logging.info(f"{request.method} {request.url.path} - {process_time:.4f}s")
    return response

后台任务

FastAPI 支持响应发送后运行的后台任务。

from fastapi import BackgroundTasks

async def send_welcome_email(email: str, name: str):
    # Simulate sending email (replace with actual email service)
    await asyncio.sleep(2)
    print(f"Welcome email sent to {name} at {email}")

async def log_user_creation(user_id: int):
    print(f"User {user_id} created at {datetime.now()}")

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(
    data: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
):
    user = await user_service.create(db, data)

    # These run AFTER the response is sent
    background_tasks.add_task(send_welcome_email, user.email, user.name)
    background_tasks.add_task(log_user_creation, user.id)

    return user  # Response is sent immediately

测试

使用 TestClient 测试 API 端点。

# tests/test_users.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

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

@pytest.mark.anyio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users", json={
        "name": "John Doe",
        "email": "john@example.com",
        "password": "SecurePass123",
    })
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "John Doe"
    assert data["email"] == "john@example.com"
    assert "password" not in data  # Password should be excluded

@pytest.mark.anyio
async def test_create_user_validation_error(client: AsyncClient):
    response = await client.post("/users", json={
        "name": "J",  # Too short
        "email": "not-an-email",
        "password": "weak",
    })
    assert response.status_code == 422
    errors = response.json()["detail"]
    assert len(errors) > 0

@pytest.mark.anyio
async def test_get_user_not_found(client: AsyncClient):
    response = await client.get("/users/99999")
    assert response.status_code == 404

# Run tests:
# pytest tests/ -v --tb=short

部署

使用 Docker 部署 FastAPI 应用到生产环境。

Docker 部署

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY ./app ./app

# Create non-root user
RUN adduser --disabled-password --no-create-home appuser
USER appuser

# Run with multiple workers for production
CMD ["uvicorn", "app.main:app", \
     "--host", "0.0.0.0", \
     "--port", "8000", \
     "--workers", "4"]

# docker-compose.yml
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/mydb
      - JWT_SECRET=your-production-secret
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - pg-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  pg-data:

性能技巧

1. I/O 密集操作使用 async 端点。
2. CPU 密集操作使用同步端点。
3. 启用 GZip 响应压缩。
4. 使用数据库连接池。
5. 使用 Redis 缓存常用数据。
6. 生产环境使用多个 Uvicorn workers。
7. 使用 py-spy 分析性能瓶颈。

最佳实践

1. 所有请求/响应使用 Pydantic 模型。
2. 将应用分为路由、模型、模式和服务。
3. 使用依赖注入处理数据库会话和认证。
4. 使用环境变量存储配置。
5. 编写全面的测试。
6. 使用异步数据库驱动。
7. 为 API 添加文档描述。
8. 一致地处理错误。

总结

FastAPI 是一个结合了 Python 开发体验和高性能的生产级框架。通过 Pydantic 自动数据验证、交互式 API 文档和原生异步支持，它是构建 Python REST API 的最佳选择。

常见问题