FastAPI 实战教程：高性能 Python Web 框架从入门到生产部署

FastAPI 是当今 Python 领域最受瞩目的 Web 框架之一，以其卓越的性能、自动 API 文档和类型安全著称。本文将从零开始，系统讲解 FastAPI 的核心特性、数据验证、数据库集成、异步编程到生产部署的完整实战流程。

一、为什么选择 FastAPI？

在 Python Web 开发领域，Django 和 Flask 曾长期占据主导地位。然而，随着微服务架构和异步编程的兴起，FastAPI 凭借以下核心优势迅速崛起：

1. 性能卓越：基于 Starlette 和 Pydantic，性能媲美 Node.js 和 Go，成为 Python 中最快的 Web 框架之一。

2. 自动文档：自动生成 Swagger UI 和 ReDoc 文档，无需额外维护 API 文档。

3. 类型安全：基于 Python 类型注解，编译时就能发现错误，减少运行时 bug。

4. 异步优先：原生支持 async/await，轻松处理高并发场景。

5. 开发效率高：智能编辑器支持（自动补全、类型检查），开发体验极佳。

二、快速入门：第一个 FastAPI 应用

2.1 安装依赖

# 安装 FastAPI 和 ASGI 服务器pip install fastapi uvicorn[standard]# 验证安装python -c "import fastapi; print(fastapi.__version__)"

2.2 Hello World 示例

from fastapi import FastAPI# 创建应用实例app = FastAPI(    title="我的第一个 API",    description="FastAPI 入门示例",    version="1.0.0")# 定义路由@app.get("/")async def root():    """根路径返回欢迎信息"""    return {"message": "Hello, FastAPI!"}@app.get("/health")async def health_check():    """健康检查接口"""    return {"status": "healthy", "version": "1.0.0"}# 运行命令: uvicorn main:app --reload

启动服务后，访问 http://127.0.0.1:8000/docs 即可看到自动生成的 Swagger 文档界面。

三、路由与参数处理

3.1 路径参数

from fastapi import FastAPIfrom typing import Optionalapp = FastAPI()# 基础路径参数@app.get("/users/{user_id}")async def get_user(user_id: int):    """根据 ID 获取用户（自动类型转换）"""    return {"user_id": user_id, "name": f"用户{user_id}"}# 枚举路径参数from enum import Enumclass ModelName(str, Enum):    alexnet = "alexnet"    resnet = "resnet"    lenet = "lenet"@app.get("/models/{model_name}")async def get_model(model_name: ModelName):    """枚举限制路径参数"""    return {"model": model_name, "message": f"当前模型: {model_name.value}"}# 路径参数包含路径@app.get("/files/{file_path:path}")async def read_file(file_path: str):    """接收包含斜杠的路径"""    return {"file_path": file_path}

3.2 查询参数

from fastapi import FastAPI, Queryfrom typing import Optional, Listapp = FastAPI()# 可选查询参数@app.get("/items/{item_id}")async def read_item(    item_id: int,    q: Optional[str] = None,  # 可选参数，默认 None    short: bool = False        # 布尔参数):    """查询参数示例"""    result = {"item_id": item_id}    if q:        result.update({"q": q})    if not short:        result.update({"description": "这是详细描述"})    return result# 参数验证@app.get("/users/")async def read_users(    skip: int = Query(0, ge=0, description="跳过的记录数"),    limit: int = Query(10, le=100, description="返回的最大记录数"),    q: Optional[str] = Query(        None,         min_length=3,         max_length=50,        regex="^[a-zA-Z0-9]+$"    )):    """带验证的查询参数"""    return {"skip": skip, "limit": limit, "q": q}# 多值查询参数@app.get("/search/")async def search_items(    tags: List[str] = Query([], alias="tag")):    """支持多个相同参数名: ?tag=python&tag=fastapi"""    return {"tags": tags}

四、请求体与数据验证

4.1 使用 Pydantic 模型

from fastapi import FastAPIfrom pydantic import BaseModel, Field, validator, EmailStrfrom typing import Optional, Listfrom datetime import datetimeapp = FastAPI()# 定义数据模型class User(BaseModel):    """用户数据模型"""    username: str = Field(..., min_length=3, max_length=20, description="用户名")    email: EmailStr = Field(..., description="邮箱地址")    full_name: Optional[str] = Field(None, max_length=50)    age: int = Field(..., ge=0, le=150, description="年龄")    tags: List[str] = Field(default_factory=list)    is_active: bool = True    created_at: datetime = Field(default_factory=datetime.now)        @validator('username')    def username_alphanumeric(cls, v):        """自定义验证器：用户名只能包含字母数字"""        if not v.isalnum():            raise ValueError('用户名只能包含字母和数字')        return v        class Config:        # 配置示例数据        schema_extra = {            "example": {                "username": "johndoe",                "email": "john@example.com",                "full_name": "John Doe",                "age": 28,                "tags": ["python", "fastapi"],                "is_active": True            }        }# 创建用户@app.post("/users/", response_model=User, status_code=201)async def create_user(user: User):    """创建用户（自动验证请求体）"""    # Pydantic 自动验证数据    return user# 响应模型控制class UserOut(BaseModel):    """输出模型（隐藏敏感字段）"""    username: str    email: str    is_active: bool@app.post("/users/public/", response_model=UserOut)async def create_public_user(user: User):    """仅返回指定字段"""    return user

4.2 嵌套模型与复杂验证

from typing import List, Dictfrom pydantic import BaseModel, HttpUrlclass Address(BaseModel):    """地址模型"""    street: str    city: str    country: str = "China"    postal_code: strclass Item(BaseModel):    """商品模型"""    name: str    description: Optional[str] = None    price: float = Field(..., gt=0, description="价格必须大于0")    tax: Optional[float] = None    class Order(BaseModel):    """订单模型（嵌套模型示例）"""    order_id: int    customer: User          # 嵌套用户模型    items: List[Item]       # 商品列表    shipping_address: Address  # 配送地址    metadata: Dict[str, str] = {}  # 元数据字典@app.post("/orders/")async def create_order(order: Order):    """处理复杂的嵌套数据"""    total = sum(item.price * (1 + (item.tax or 0)) for item in order.items)    return {        "order_id": order.order_id,        "customer": order.customer.username,        "total": total,        "items_count": len(order.items)    }

五、数据库集成实战

5.1 SQLAlchemy 异步集成

from fastapi import FastAPI, Depends, HTTPExceptionfrom sqlalchemy import create_engine, Column, Integer, String, Booleanfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import Session, sessionmakerfrom typing import List# 数据库配置SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"# MySQL 示例: "mysql+pymysql://user:password@localhost/dbname"engine = create_engine(    SQLALCHEMY_DATABASE_URL,     connect_args={"check_same_thread": False}  # SQLite 特定配置)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()# 定义 ORM 模型class UserDB(Base):    __tablename__ = "users"        id = Column(Integer, primary_key=True, index=True)    username = Column(String(50), unique=True, index=True)    email = Column(String(100), unique=True)    hashed_password = Column(String(100))    is_active = Column(Boolean, default=True)# 创建表Base.metadata.create_all(bind=engine)# 依赖注入：获取数据库会话def get_db():    db = SessionLocal()    try:        yield db    finally:        db.close()app = FastAPI()# CRUD 操作@app.post("/users/", response_model=User)def create_user_endpoint(user: User, db: Session = Depends(get_db)):    """创建用户"""    # 检查用户是否已存在    db_user = db.query(UserDB).filter(UserDB.email == user.email).first()    if db_user:        raise HTTPException(status_code=400, detail="邮箱已被注册")        # 创建新用户    db_user = UserDB(        username=user.username,        email=user.email,        hashed_password="hashed_" + user.username  # 实际应使用密码哈希    )    db.add(db_user)    db.commit()    db.refresh(db_user)    return db_user@app.get("/users/", response_model=List[User])def list_users(    skip: int = 0,     limit: int = 100,     db: Session = Depends(get_db)):    """获取用户列表"""    users = db.query(UserDB).offset(skip).limit(limit).all()    return users@app.get("/users/{user_id}", response_model=User)def get_user_endpoint(user_id: int, db: Session = Depends(get_db)):    """获取单个用户"""    user = db.query(UserDB).filter(UserDB.id == user_id).first()    if not user:        raise HTTPException(status_code=404, detail="用户不存在")    return user

5.2 异步数据库（SQLAlchemy 2.0 + async）

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSessionfrom sqlalchemy.orm import sessionmaker# 异步数据库引擎ASYNC_DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"async_engine = create_async_engine(ASYNC_DATABASE_URL)AsyncSessionLocal = sessionmaker(    async_engine,     class_=AsyncSession,     expire_on_commit=False)# 异步依赖async def get_async_db():    async with AsyncSessionLocal() as session:        yield session@app.get("/users/async/{user_id}")async def get_user_async(    user_id: int,     db: AsyncSession = Depends(get_async_db)):    """异步查询用户"""    result = await db.execute(        select(UserDB).where(UserDB.id == user_id)    )    user = result.scalar_one_or_none()    if not user:        raise HTTPException(status_code=404, detail="用户不存在")    return user

六、认证与授权

6.1 JWT 认证实现

from fastapi import FastAPI, Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormfrom jose import JWTError, jwtfrom passlib.context import CryptContextfrom datetime import datetime, timedeltafrom typing import Optional# 安全配置SECRET_KEY = "your-secret-key-keep-it-secret"ALGORITHM = "HS256"ACCESS_TOKEN_EXPIRE_MINUTES = 30pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")app = FastAPI()# 密码处理def verify_password(plain_password: str, hashed_password: str) -> bool:    """验证密码"""    return pwd_context.verify(plain_password, hashed_password)def get_password_hash(password: str) -> str:    """生成密码哈希"""    return pwd_context.hash(password)# JWT 令牌def create_access_token(data: dict, expires_delta: Optional[timedelta] = 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)# 认证路由@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="用户名或密码错误",            headers={"WWW-Authenticate": "Bearer"},        )        access_token = create_access_token(        data={"sub": user.username},        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)    )    return {"access_token": access_token, "token_type": "bearer"}# 受保护的路由@app.get("/users/me/")async def read_users_me(token: str = Depends(oauth2_scheme)):    """获取当前用户信息"""    credentials_exception = HTTPException(        status_code=status.HTTP_401_UNAUTHORIZED,        detail="无法验证凭据",    )    try:        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])        username: str = payload.get("sub")        if username is None:            raise credentials_exception    except JWTError:        raise credentials_exception        # 从数据库获取用户    user = get_user_by_username(username)    return user

七、异步编程与并发处理

7.1 异步外部 API 调用

import httpxfrom fastapi import FastAPIapp = FastAPI()# 异步 HTTP 客户端@app.get("/external-data/")async def fetch_external_data():    """并发调用多个外部 API"""    async with httpx.AsyncClient() as client:        # 并发请求        responses = await asyncio.gather(            client.get("https://api.example.com/users"),            client.get("https://api.example.com/posts"),            client.get("https://api.example.com/comments")        )        return {        "users": responses[0].json(),        "posts": responses[1].json(),        "comments": responses[2].json()    }# 后台任务from fastapi import BackgroundTasksdef send_email(email: str, message: str):    """发送邮件（后台执行）"""    print(f"发送邮件到 {email}: {message}")    # 实际邮件发送逻辑@app.post("/send-notification/")async def send_notification(    email: str,    message: str,    background_tasks: BackgroundTasks):    """异步处理后台任务"""    # 添加后台任务    background_tasks.add_task(send_email, email, message)    return {"message": "通知已发送", "email": email}

八、中间件与生命周期

8.1 自定义中间件

from fastapi import FastAPI, Requestfrom fastapi.middleware.cors import CORSMiddlewareimport timeapp = FastAPI()# CORS 中间件app.add_middleware(    CORSMiddleware,    allow_origins=["*"],  # 生产环境应限制具体域名    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],)# 自定义中间件：请求耗时记录@app.middleware("http")async def add_process_time_header(request: Request, call_next):    """记录请求处理时间"""    start_time = time.time()    response = await call_next(request)    process_time = time.time() - start_time    response.headers["X-Process-Time"] = str(process_time)    print(f"请求 {request.url.path} 耗时: {process_time:.3f}s")    return response# 生命周期事件@app.on_event("startup")async def startup_event():    """应用启动时执行"""    print("应用启动中...")    # 初始化数据库连接池、缓存等@app.on_event("shutdown")async def shutdown_event():    """应用关闭时执行"""    print("应用关闭中...")    # 关闭连接池、清理资源

九、测试与文档

9.1 单元测试

from fastapi.testclient import TestClientimport pytestclient = TestClient(app)def test_read_root():    """测试根路由"""    response = client.get("/")    assert response.status_code == 200    assert response.json() == {"message": "Hello, FastAPI!"}def test_create_user():    """测试创建用户"""    response = client.post(        "/users/",        json={            "username": "testuser",            "email": "test@example.com",            "age": 25        }    )    assert response.status_code == 201    data = response.json()    assert data["username"] == "testuser"    assert data["email"] == "test@example.com"# 运行测试: pytest test_main.py -v

9.2 API 文档配置

from fastapi import FastAPIfrom fastapi.openapi.utils import get_openapiapp = FastAPI(    title="My API",    description="API 文档描述",    version="1.0.0",    docs_url="/docs",           # Swagger UI 路径    redoc_url="/redoc",         # ReDoc 路径    openapi_url="/openapi.json" # OpenAPI schema 路径)# 自定义 OpenAPIdef custom_openapi():    if app.openapi_schema:        return app.openapi_schema        openapi_schema = get_openapi(        title="自定义 API 文档",        version="2.0.0",        description="这是自定义的 API 文档",        routes=app.routes,    )    openapi_schema["info"]["x-logo"] = {        "url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png"    }    app.openapi_schema = openapi_schema    return app.openapi_schemaapp.openapi = custom_openapi

十、生产部署最佳实践

10.1 使用 Gunicorn + Uvicorn

# 安装生产依赖pip install gunicorn uvicorn[standard]# 启动命令（多 worker）gunicorn main:app \    --workers 4 \    --worker-class uvicorn.workers.UvicornWorker \    --bind 0.0.0.0:8000 \    --timeout 120 \    --access-logfile - \    --error-logfile -

10.2 Docker 部署

# DockerfileFROM python:3.11-slimWORKDIR /app# 安装依赖COPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txt# 复制代码COPY . .# 启动命令CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "-b", "0.0.0.0:8000"]

10.3 环境变量配置

from pydantic import BaseSettingsclass Settings(BaseSettings):    """应用配置"""    app_name: str = "FastAPI App"    admin_email: str = "admin@example.com"    database_url: str    secret_key: str    debug: bool = False        class Config:        env_file = ".env"  # 从 .env 文件加载settings = Settings()@app.get("/info")async def info():    return {        "app_name": settings.app_name,        "admin_email": settings.admin_email    }

总结

FastAPI 凭借其卓越的性能、现代化的设计理念和出色的开发体验，已成为构建 Python Web API 的首选框架之一。通过本文的学习，你应该掌握了：

• FastAPI 的核心特性与优势
• 路由、参数与请求数据的处理
• Pydantic 数据验证与模型设计
• 数据库集成与异步操作
• JWT 认证与授权实现
• 中间件与生命周期管理
• 测试与文档自动化
• 生产环境部署最佳实践

建议在实际项目中循序渐进，先掌握基础路由和数据验证，再逐步深入异步编程、数据库集成等高级主题。FastAPI 的官方文档非常详尽，遇到问题可随时查阅。

相关链接：

FastAPI 官方文档 | GitHub 仓库 | Pydantic 文档

本文链接：https://www.kkkliao.cn/?id=981 转载需授权！