2026-04-19

FastAPI 全栈指南：从入门到项目实战

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API。它基于 Python 3.8+ 的标准类型提示，支持异步编程，并自动生成交互式 API 文档。本文旨在系统整理 FastAPI 的核心知识点，并结合实际项目经验，提供从零搭建到部署上线的完整参考。

1. 环境搭建与项目初始化

1.1 安装推荐方式

pip install "fastapi[standard]"

这会一次性安装 FastAPI、Uvicorn（ASGI 服务器）、Pydantic、HTTPX 等常用依赖。

1.2 最小应用

from fastapi import FastAPI

app = FastAPI()

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

运行命令：

uvicorn main:app --reload

2. 路由与请求处理

2.1 路径操作装饰器

@app.get()、@app.post()、@app.put()、@app.delete() 等
路径参数：使用 {param} 在路径中声明，函数参数同名接收
查询参数：函数中未在路径中声明的参数自动解释为查询参数

2.2 参数类型与验证

FastAPI 利用类型提示和 Annotated 为参数添加元数据（验证、描述等）。

参数类型	数据来源	声明方式（推荐 Annotated）
路径参数	URL 路径	`item_id: Annotated[int, Path(title="商品ID")]`
查询参数	URL `?` 之后	`q: Annotated[str \| None, Query(max_length=50)] = None`
请求体	HTTP body	`item: Item`（Item 为 Pydantic 模型）
请求头	HTTP headers	`user_agent: Annotated[str \| None, Header()] = None`
Cookie	Cookie 字段	`session_id: Annotated[str \| None, Cookie()] = None`
表单数据	`application/x-www-form-urlencoded`	`username: Annotated[str, Form()]`
文件上传	`multipart/form-data`	`file: Annotated[UploadFile, File()]`

2.3 请求体模型（Pydantic）

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., max_length=50, description="物品名称")
    price: float = Field(..., gt=0, le=10000)
    tax: float | None = Field(None, ge=0)

3. 响应处理

3.1 自动 JSON 转换

可以直接返回 dict、list、Pydantic 模型，FastAPI 会自动转换为 JSON。

3.2 响应模型

使用 response_model 控制输出字段、执行响应验证。

@app.post("/items/", response_model=ItemOut)
async def create_item(item: Item):
    # 返回的 item 会被过滤，只保留 ItemOut 定义的字段
    return item

3.3 状态码与响应头

from fastapi import status
from fastapi.responses import JSONResponse

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return JSONResponse(
        content={"message": "created"},
        headers={"X-Custom-Header": "foo"},
    )

4. 依赖注入（Dependency Injection）

4.1 基本概念

依赖注入允许你将通用的逻辑（如认证、数据库连接）剥离出来，按需注入到路由函数中。

4.2 依赖定义与使用

from fastapi import Depends
from typing import Annotated

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

CommonsDep = Annotated[dict, Depends(common_parameters)]

@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons

4.3 类作为依赖

class Pagination:
    def __init__(self, skip: int = 0, limit: int = 100):
        self.skip = skip
        self.limit = limit

PaginationDep = Annotated[Pagination, Depends(Pagination)]

@app.get("/items/")
async def read_items(pagination: PaginationDep):
    return {"skip": pagination.skip, "limit": pagination.limit}

4.4 带 yield 的依赖（资源管理）

async def get_db_session():
    # 请求开始时获取资源
    session = Session(engine)
    try:
        yield session
    finally:
        # 请求结束后清理资源
        session.close()

5. 数据库集成（SQLModel）

5.1 为什么选择 SQLModel？

SQLModel 结合了 SQLAlchemy（数据库操作）和 Pydantic（数据验证），一个类同时定义数据库表和 API 模型。

5.2 配置引擎与会话依赖

from sqlmodel import create_engine, Session, SQLModel

sqlite_url = "sqlite:///database.db"
engine = create_engine(sqlite_url, connect_args={"check_same_thread": False})

# 依赖项：每个请求一个会话，请求结束后自动关闭
def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

5.3 定义模型与 CRUD 示例

from sqlmodel import Field
from typing import Optional

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(max_length=50)
    email: str = Field(unique=True)

@app.post("/users/")
def create_user(user: User, session: SessionDep):
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

@app.get("/users/")
def read_users(session: SessionDep, skip: int = 0, limit: int = 100):
    users = session.exec(select(User).offset(skip).limit(limit)).all()
    return users

5.4 异步数据库操作

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlmodel.ext.asyncio.session import AsyncSession

engine = create_async_engine("sqlite+aiosqlite:///database.db")

async def get_async_session():
    async with AsyncSession(engine) as session:
        yield session

AsyncSessionDep = Annotated[AsyncSession, Depends(get_async_session)]

@app.get("/users/")
async def read_users(session: AsyncSessionDep):
    result = await session.exec(select(User))
    return result.scalars().all()

6. 异步编程

6.1 async/await 基础

async def 定义协程函数，内部可使用 await 等待其他协程。
FastAPI 支持混合同步（def）和异步（async def）路由。
同步路由在线程池中运行，不会阻塞事件循环；异步路由在事件循环中运行，适合 IO 密集型操作。

6.2 异步 HTTP 客户端（httpx）

import httpx
from fastapi import HTTPException

@app.get("/external")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        if response.status_code != 200:
            raise HTTPException(status_code=500, detail="External API error")
        return response.json()

7. 错误处理

7.1 HTTPException

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

7.2 自定义异常处理器

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
    )

8. 生命周期事件（Lifespan）

8.1 什么是 Lifespan？

Lifespan 允许你在应用启动和关闭时执行代码，常用于初始化数据库连接、加载模型、启动后台任务等。

8.2 使用示例

from contextlib import asynccontextmanager
from fastapi import FastAPI

async def startup():
    # 连接数据库、创建表、加载 ML 模型等
    print("Starting up...")
    # 例如：SQLModel.metadata.create_all(engine)

async def shutdown():
    # 关闭数据库连接、清理资源等
    print("Shutting down...")

@asynccontextmanager
async def lifespan(app: FastAPI):
    await startup()
    yield
    await shutdown()

app = FastAPI(lifespan=lifespan)

9. 项目结构与最佳实践

9.1 模块化组织

myproject/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI 应用实例
│   ├── config.py        # 配置（环境变量、常量）
│   ├── dependencies.py  # 依赖项定义
│   ├── models.py        # SQLModel/Pydantic 模型
│   ├── crud.py          # 数据库操作函数
│   ├── routers/         # 路由模块
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── users.py
│   └── utils/           # 工具函数
│       └── security.py  # 认证、密码哈希等
├── tests/               # 测试
├── alembic/             # 数据库迁移（可选）
└── requirements.txt

9.2 配置管理

使用 pydantic-settings 管理环境变量：

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My FastAPI App"
    database_url: str = "sqlite:///./database.db"
    secret_key: str

    class Config:
        env_file = ".env"

settings = Settings()

9.3 路由拆分

# app/routers/items.py
from fastapi import APIRouter

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

@router.get("/")
async def read_items():
    return [{"item": "Foo"}]

# app/main.py
from app.routers import items, users

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

10. 部署与性能优化

10.1 生产服务器

使用 uvicorn 搭配 gunicorn（多进程）或 hypercorn（异步）。
推荐配置：uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

10.2 环境变量与安全

敏感信息（密钥、数据库密码）放在 .env 文件中，通过 pydantic-settings 加载。
启用 CORS（如果需要前端跨域访问）：

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应限制为具体域名
    allow_methods=["*"],
    allow_headers=["*"],
)

10.3 监控与日志

使用 loguru 或 Python 标准 logging 记录请求日志。
集成 Prometheus + Grafana 监控指标（可通过 prometheus-fastapi-instrumentator）。

11. 常见问题与经验分享

11.1 数据库会话管理

每个请求使用独立的会话，确保事务隔离。
使用 yield 依赖自动关闭会话，避免连接泄漏。

11.2 异步代码注意事项

不要在异步路由中执行 CPU 密集型操作（会阻塞事件循环）。
同步库（如某些数据库驱动）可能阻塞事件循环，考虑使用 run_in_executor 在线程池中运行。

11.3 性能瓶颈排查

使用 uvicorn --log-level debug 查看请求细节。
利用 py-spy 或 async-profiler 分析性能热点。

12. 后续学习方向

WebSocket：实时双向通信（聊天室、实时通知）。（在我的chatnovel项目已用）
GraphQL：通过 strawberry 或 graphene 集成 GraphQL。
任务队列：使用 celery 或 arq 处理后台任务。
容器化：编写 Dockerfile，使用 Docker Compose 编排服务。

说明：本文档作为个人回顾使用，更多在于shysta个人项目整理（AI辅助）具体以官方文档为准

欢迎在评论区补充你的实战经验或提问！

发现新版本