FastAPI 完整指南：从开发到生产部署

本教程将全面介绍 FastAPI 框架的使用，从基础安装到生产环境部署。内容基于 FastAPI 官方文档整理，适合初学者和有经验的开发者。

Table of contents

FastAPI 简介

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于标准 Python 类型提示构建 API。主要特点：

• 快速：性能与 NodeJS 和 Go 相当，是最快的 Python 框架之一
• 高效开发：开发速度提升约 200% 到 300%
• 更少的 bug：减少约 40% 的人为错误
• 直观：出色的编辑器支持，自动补全无处不在
• 简单：易于使用和学习，减少阅读文档的时间
• 简短：最小化代码重复，每个参数声明都有多个功能
• 健壮：生产可用的代码，自动生成交互式文档
• 基于标准：基于并完全兼容 API 的开放标准 OpenAPI 和 JSON Schema

第一步：安装 FastAPI

前置要求

• Python 3.8 或更高版本
• pip 包管理器

创建虚拟环境

强烈建议使用虚拟环境来隔离项目依赖：

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境（Linux/Mac）
source venv/bin/activate

# 激活虚拟环境（Windows）
venv\Scripts\activate

安装 FastAPI

安装包含所有标准依赖的 FastAPI：

pip install "fastapi[standard]"

这将安装 FastAPI 及其标准依赖，包括：

• uvicorn：ASGI 服务器
• pydantic：数据验证
• email-validator：邮箱验证
• python-multipart：表单和文件上传支持

如果只需要核心功能：

pip install fastapi

单独安装 ASGI 服务器：

pip install "uvicorn[standard]"

第二步：创建第一个应用

基础示例

创建 main.py 文件：

from fastapi import FastAPI

app = FastAPI()

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

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

运行开发服务器

使用 FastAPI CLI（推荐）：

fastapi dev main.py

或使用 Uvicorn：

uvicorn main:app --reload

参数说明：

• main：文件名（main.py）
• app：FastAPI 实例对象
• --reload：代码更改时自动重启（仅开发环境使用）

访问应用

• API 端点：http://127.0.0.1:8000
• 交互式文档（Swagger UI）：http://127.0.0.1:8000/docs
• 替代文档（ReDoc）：http://127.0.0.1:8000/redoc
• OpenAPI 规范：http://127.0.0.1:8000/openapi.json

第三步：核心功能开发

路径参数

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

# 路径参数验证
from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    return {"model_name": model_name, "message": "Deep Learning FTW!"}

查询参数

from typing import Union

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10, q: Union[str, None] = None):
    items = {"skip": skip, "limit": limit}
    if q:
        items.update({"q": q})
    return items

请求体（Request Body）

from pydantic import BaseModel
from typing import Union

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

数据验证

from pydantic import BaseModel, Field, EmailStr
from typing import Union

class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    full_name: Union[str, None] = Field(None, max_length=100)
    age: int = Field(..., gt=0, le=120)

@app.post("/users/")
async def create_user(user: User):
    return user

依赖注入

from fastapi import Depends, HTTPException

async def get_token_header(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

@app.get("/items/", dependencies=[Depends(get_token_header)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

数据库集成（SQLAlchemy）

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import Depends

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 模型定义
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    name = Column(String)

Base.metadata.create_all(bind=engine)

# 依赖
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 路由
@app.post("/users/")
def create_user(email: str, name: str, db: Session = Depends(get_db)):
    db_user = User(email=email, name=name)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

认证和授权

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

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

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 验证用户逻辑
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
        )
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

CORS 配置

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应指定具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

第四步：项目结构最佳实践

推荐的项目结构

my_fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用入口
│   ├── config.py            # 配置文件
│   ├── dependencies.py      # 全局依赖
│   ├── models/              # 数据库模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/             # Pydantic 模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/             # 路由模块
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   ├── services/            # 业务逻辑
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── utils/               # 工具函数
│       ├── __init__.py
│       └── security.py
├── tests/                   # 测试文件
│   ├── __init__.py
│   └── test_main.py
├── alembic/                 # 数据库迁移
├── .env                     # 环境变量
├── .gitignore
├── requirements.txt
└── README.md

模块化路由

app/routers/users.py:

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from ..dependencies import get_db
from ..schemas import user as schemas
from ..services import user_service

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

@router.get("/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = user_service.get_users(db, skip=skip, limit=limit)
    return users

@router.post("/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    return user_service.create_user(db=db, user=user)

app/main.py:

from fastapi import FastAPI
from .routers import users, items

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

app.include_router(users.router)
app.include_router(items.router)

@app.get("/")
async def root():
    return {"message": "Welcome to My API"}

配置管理

app/config.py:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My FastAPI App"
    database_url: str
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    
    class Config:
        env_file = ".env"

settings = Settings()

.env:

DATABASE_URL=postgresql://user:password@localhost/dbname
SECRET_KEY=your-secret-key-here

第五步：生产环境部署

使用 Uvicorn + Gunicorn

生产环境推荐使用 Gunicorn 作为进程管理器，Uvicorn 作为 worker。

安装依赖：

pip install gunicorn uvicorn[standard]

启动命令：

gunicorn app.main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 120 \
    --access-logfile - \
    --error-logfile -

参数说明：

• --workers：工作进程数，通常设置为 CPU 核心数的 2-4 倍
• --worker-class：使用 Uvicorn worker 类
• --bind：绑定地址和端口
• --timeout：请求超时时间（秒）

使用 Docker 部署

Dockerfile:

FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY ./app ./app

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["gunicorn", "app.main:app", \
     "--workers", "4", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000"]

docker-compose.yml:

version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/dbname
      - SECRET_KEY=your-secret-key
    depends_on:
      - db
    restart: unless-stopped

  db:
    image: postgres:15
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=dbname
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped

volumes:
  postgres_data:

构建和运行：

docker-compose up -d

使用 Nginx 反向代理

/etc/nginx/sites-available/fastapi:

server {
    listen 80;
    server_name your-domain.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # WebSocket 支持
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        
        # 超时设置
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
    }
}

启用配置：

sudo ln -s /etc/nginx/sites-available/fastapi /etc/nginx/sites-enabled/
sudo nginx -t
sudo systemctl reload nginx

HTTPS 配置（Let’s Encrypt）

# 安装 Certbot
sudo apt install certbot python3-certbot-nginx

# 获取证书
sudo certbot --nginx -d your-domain.com

# 自动续期
sudo certbot renew --dry-run

Systemd 服务配置

/etc/systemd/system/fastapi.service:

[Unit]
Description=FastAPI Application
After=network.target

[Service]
Type=notify
User=www-data
Group=www-data
WorkingDirectory=/path/to/your/app
Environment="PATH=/path/to/your/venv/bin"
ExecStart=/path/to/your/venv/bin/gunicorn app.main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000
ExecReload=/bin/kill -s HUP $MAINPID
KillMode=mixed
TimeoutStopSec=5
PrivateTmp=true
Restart=on-failure

[Install]
WantedBy=multi-user.target

启用服务：

sudo systemctl daemon-reload
sudo systemctl enable fastapi
sudo systemctl start fastapi
sudo systemctl status fastapi

第六步：性能优化

异步数据库操作

使用异步数据库驱动提升性能：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

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

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).filter(User.id == user_id))
    user = result.scalar_one_or_none()
    return user

缓存策略

使用 Redis 缓存：

import redis.asyncio as redis
from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache

app = FastAPI()

@app.on_event("startup")
async def startup():
    redis_client = redis.from_url("redis://localhost")
    FastAPICache.init(RedisBackend(redis_client), prefix="fastapi-cache")

@app.get("/items/{item_id}")
@cache(expire=60)
async def read_item(item_id: int):
    # 这个结果会被缓存 60 秒
    return {"item_id": item_id, "name": "Item Name"}

连接池配置

from sqlalchemy import create_engine

engine = create_engine(
    DATABASE_URL,
    pool_size=20,          # 连接池大小
    max_overflow=0,        # 超出 pool_size 后可创建的连接数
    pool_pre_ping=True,    # 使用前检查连接是否有效
    pool_recycle=3600,     # 连接回收时间（秒）
)

响应压缩

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)

第七步：监控和日志

日志配置

import logging
from logging.handlers import RotatingFileHandler

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        RotatingFileHandler('app.log', maxBytes=10485760, backupCount=5),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    logger.info(f"Reading item {item_id}")
    return {"item_id": item_id}

健康检查端点

from fastapi import status

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

@app.get("/readiness")
async def readiness_check(db: Session = Depends(get_db)):
    try:
        # 检查数据库连接
        db.execute("SELECT 1")
        return {"status": "ready"}
    except Exception as e:
        return {"status": "not ready", "error": str(e)}

性能监控（Prometheus）

from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

Instrumentator().instrument(app).expose(app)

访问 /metrics 端点查看指标。

第八步：测试

单元测试

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Test Item", "price": 10.5}
    )
    assert response.status_code == 200
    assert response.json()["name"] == "Test Item"

异步测试

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_read_main():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200

运行测试：

pip install pytest pytest-asyncio httpx
pytest

常见问题

1. CORS 错误

确保正确配置 CORS 中间件，并在生产环境中指定具体的允许域名。

2. 数据库连接池耗尽

增加连接池大小或优化查询，确保及时关闭数据库会话。

3. 请求超时

调整 Gunicorn 的 --timeout 参数，或优化慢查询。

4. 内存泄漏

使用内存分析工具（如 memory_profiler）定位问题，确保正确关闭资源。

5. 静态文件服务

from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="static"), name="static")

日常维护

依赖更新

定期更新依赖包：

pip list --outdated
pip install --upgrade fastapi uvicorn

数据库迁移

使用 Alembic 管理数据库迁移：

pip install alembic
alembic init alembic
alembic revision --autogenerate -m "Add users table"
alembic upgrade head

日志轮转

配置日志轮转避免日志文件过大：

from logging.handlers import TimedRotatingFileHandler

handler = TimedRotatingFileHandler(
    'app.log',
    when='midnight',
    interval=1,
    backupCount=30
)

备份策略

定期备份数据库和配置文件：

# PostgreSQL 备份
pg_dump -U user dbname > backup_$(date +%Y%m%d).sql

# 恢复
psql -U user dbname < backup_20260301.sql

总结

本教程涵盖了 FastAPI 从开发到生产部署的完整流程。FastAPI 的高性能和易用性使其成为构建现代 API 的理想选择。遵循最佳实践，合理配置生产环境，可以构建出稳定、高效的 API 服务。

更多信息请参考：

• FastAPI 官方文档
• FastAPI GitHub 仓库
• Uvicorn 文档
• Pydantic 文档

内容基于 FastAPI 官方文档改编和社区最佳实践整理。