当前位置：首页 > 学习笔记 > 正文内容

FastAPI 教程实战完全指南：从零构建高性能 API 服务

廖万里16小时前学习笔记1

FastAPI 是现代 Python Web 框架的佼佼者，凭借高性能、自动文档、类型提示等特性，成为构建 API 服务的首选。本文将从零开始，系统讲解 FastAPI 的核心概念、路由系统、数据验证、数据库集成、中间件、部署优化等实战技能，助你快速掌握构建高性能 API 服务的完整能力。

一、核心概念与快速入门

FastAPI 是一个现代、高性能的 Python Web 框架，用于构建 API 服务。它的核心优势包括：

高性能：基于 Starlette 和 Pydantic，性能媲美 Node.js 和 Go
自动文档：自动生成 OpenAPI (Swagger) 和 ReDoc 文档
类型提示：充分利用 Python 类型注解，提供强大的编辑器支持
异步支持：原生支持 async/await，轻松处理高并发
数据验证：自动请求数据验证和序列化

1.1 安装与第一个应用

# 安装 FastAPI 和 Uvicorn 服务器# pip install fastapi uvicorn[standard]from fastapi import FastAPI# 创建 FastAPI 应用实例app = FastAPI(    title="我的第一个 API",    description="学习 FastAPI 的示例项目",    version="1.0.0")# 定义根路由@app.get("/")async def root():    """根路径，返回欢迎信息"""    return {"message": "Hello, FastAPI!"}# 定义带路径参数的路由@app.get("/items/{item_id}")async def read_item(item_id: int, q: str = None):    """    获取物品信息    - item_id: 物品ID（路径参数，自动转换为整数）    - q: 可选的查询参数    """    result = {"item_id": item_id}    if q:        result.update({"q": q})    return result# 启动命令: uvicorn main:app --reload# 访问文档: http://127.0.0.1:8000/docs

1.2 项目结构最佳实践

# 推荐的项目目录结构"""my_fastapi_project/├── app/│   ├── __init__.py│   ├── main.py          # 应用入口│   ├── config.py        # 配置管理│   ├── models/          # 数据模型│   │   ├── __init__.py│   │   └── schemas.py   # Pydantic 模型│   ├── routers/         # 路由模块│   │   ├── __init__.py│   │   ├── users.py│   │   └── items.py│   ├── services/        # 业务逻辑│   │   └── auth.py│   └── utils/           # 工具函数│       └── database.py├── tests/               # 测试文件├── requirements.txt└── .env                 # 环境变量"""

二、路由系统与请求处理

FastAPI 的路由系统简洁而强大，支持多种 HTTP 方法和参数类型。

2.1 路由装饰器与 HTTP 方法

from fastapi import FastAPI, HTTPException, statusfrom pydantic import BaseModelfrom typing import Optional, Listapp = FastAPI()# 定义数据模型class Item(BaseModel):    """物品数据模型，用于请求体验证"""    name: str    price: float    description: Optional[str] = None    tags: List[str] = []# 模拟数据库fake_db = {}# POST - 创建资源@app.post("/items/", status_code=status.HTTP_201_CREATED)async def create_item(item: Item):    """创建新物品，返回201状态码"""    item_id = len(fake_db) + 1    fake_db[item_id] = item    return {"id": item_id, "item": item}# GET - 获取资源列表@app.get("/items/")async def list_items(skip: int = 0, limit: int = 10):    """    分页获取物品列表    - skip: 跳过的记录数（查询参数）    - limit: 返回的最大数量    """    items = list(fake_db.values())[skip : skip + limit]    return {"items": items, "count": len(items)}# GET - 获取单个资源@app.get("/items/{item_id}")async def get_item(item_id: int):    """获取指定物品，不存在则返回404"""    if item_id not in fake_db:        raise HTTPException(            status_code=404,            detail=f"物品 {item_id} 不存在"        )    return fake_db[item_id]# PUT - 完整更新资源@app.put("/items/{item_id}")async def update_item(item_id: int, item: Item):    """完整更新物品信息"""    if item_id not in fake_db:        raise HTTPException(status_code=404, detail="物品不存在")    fake_db[item_id] = item    return {"id": item_id, "item": item}# DELETE - 删除资源@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)async def delete_item(item_id: int):    """删除物品，返回204无内容"""    if item_id not in fake_db:        raise HTTPException(status_code=404)    del fake_db[item_id]    return None

2.2 参数类型详解

from fastapi import FastAPI, Path, Query, Header, Cookie, Form, File, UploadFileapp = FastAPI()# 路径参数 - 使用 Path 进行验证@app.get("/users/{user_id}/items/{item_id}")async def get_user_item(    user_id: int = Path(..., title="用户ID", ge=1, description="必须大于0"),    item_id: int = Path(..., title="物品ID", gt=0),):    """    路径参数验证：    - ... 表示必填    - ge=1: 大于等于1    - gt=0: 大于0    """    return {"user_id": user_id, "item_id": item_id}# 查询参数 - 使用 Query 进行验证@app.get("/search/")async def search_items(    q: str = Query(..., min_length=3, max_length=50, description="搜索关键词"),    page: int = Query(1, ge=1, description="页码"),    size: int = Query(10, ge=1, le=100, description="每页数量"),):    """    查询参数验证：    - min_length/max_length: 字符串长度限制    - ge/le: 数值范围限制    """    return {"query": q, "page": page, "size": size}# 文件上传@app.post("/upload/")async def upload_file(file: UploadFile = File(...)):    """    文件上传处理    - UploadFile 是异步的文件包装器    """    content = await file.read()    return {        "filename": file.filename,        "content_type": file.content_type,        "size": len(content)    }

三、数据验证与模型设计

FastAPI 使用 Pydantic 进行数据验证，确保数据的正确性和一致性。

from pydantic import BaseModel, Field, EmailStr, validatorfrom datetime import datetimefrom typing import Optional, Listfrom enum import Enum# 枚举类型class ItemStatus(str, Enum):    """物品状态枚举"""    ACTIVE = "active"    INACTIVE = "inactive"# 基础模型class ItemBase(BaseModel):    """物品基础模型"""    name: str = Field(..., min_length=1, max_length=100, description="物品名称")    price: float = Field(..., gt=0, description="价格，必须大于0")    description: Optional[str] = Field(None, max_length=500)    status: ItemStatus = ItemStatus.ACTIVE# 创建模型class ItemCreate(ItemBase):    """创建物品时的请求模型"""    tags: List[str] = Field(default_factory=list, description="标签列表")        @validator("name")    def name_must_not_contain_special_chars(cls, v):        """验证名称不能包含特殊字符"""        if any(char in v for char in ["@", "#", "$", "%"]):            raise ValueError("名称不能包含特殊字符")        return v# 响应模型class ItemResponse(ItemBase):    """物品响应模型，包含额外字段"""    id: int    created_at: datetime    updated_at: datetime        class Config:        from_attributes = True

四、数据库集成

FastAPI 可以与多种数据库集成，包括 PostgreSQL、MySQL、SQLite、MongoDB 等。

# database.py - 数据库配置from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker# 数据库连接 URLSQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"# 创建引擎engine = create_engine(    SQLALCHEMY_DATABASE_URL,    connect_args={"check_same_thread": False})# 创建会话工厂SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 创建基类Base = declarative_base()# 依赖注入：获取数据库会话def get_db():    """获取数据库会话的依赖函数"""    db = SessionLocal()    try:        yield db    finally:        db.close()

五、认证与授权

实现安全的用户认证是 API 开发的重要环节，FastAPI 提供了完善的 OAuth2 支持。

from fastapi import APIRouter, Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearerfrom jose import JWTError, jwtfrom passlib.context import CryptContextfrom 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 方案oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")def create_access_token(data: dict):    """创建 JWT Token"""    to_encode = data.copy()    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)    to_encode.update({"exp": expire})    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

六、部署与优化

# 使用 Gunicorn + Uvicorn 部署# gunicorn.conf.pyimport multiprocessingbind = "0.0.0.0:8000"workers = multiprocessing.cpu_count() * 2 + 1worker_class = "uvicorn.workers.UvicornWorker"keepalive = 120# 启动命令# gunicorn main:app -c gunicorn.conf.py