远程办公摸鱼时我偷偷把FastAPI啃完了

凌晨两点半，刚把线上一个OOM的锅甩给隔壁组的老王，我瘫在人体工学椅上，盯着屏幕发呆。远程办公的第三年，我已经习惯了这种"白天开会扯皮，晚上才是真正写代码"的节奏。公司美其名曰"弹性工作制"，实际上就是让你24小时随时待命，毕竟在家办公嘛，连通勤摸鱼的时间都给你"优化"掉了。

说回正题。最近组里要搞一个新的内部工具平台，领导一拍脑袋说要用Python写后端，理由是"Python开发快，AI也好接"。我当时内心OS是：Flask？Django？这俩老古董又要开始折腾了？但作为一个合格的996福报享受者（别问我为什么远程还在996，问就是热爱），我顺手提了一嘴FastAPI，没想到领导居然听过，还让我先搞个demo出来。

于是就有了这篇踩坑记录。如果你也是Python后端新手，或者正在纠结技术选型，希望我的这些经验能帮你少走点弯路。

为啥选FastAPI而不是Flask

在做技术选型的时候，我习惯性地先去看Star数和社区活跃度。毕竟开源项目嘛，Star多不一定好，但Star少一定有问题——这是我在Github上踩了无数坑之后总结出来的血泪教训。

对比维度	Flask	FastAPI	Django
Star数	67k+	77k+	79k+
异步支持	需扩展	原生支持	4.0+支持
自动文档	无	Swagger/ReDoc	需扩展
类型校验	需手动	Pydantic原生	Forms自带
性能基准	中等	接近Go/Node	中等偏下
学习曲线	低	中	高

说实话，Flask我用了好几年了，从v0.10一直用到现在的v3.x，感情是有的。但FastAPI有几个点是真的戳中了我：

第一，原生异步支持。 我们现在的服务动不动就要调一堆外部接口，用Flask的话得自己搞异步方案，要么上aiohttp，要么用gevent打猴子补丁，总有一种"在屎山上雕花"的感觉。FastAPI直接基于Starlette，异步是刻在DNA里的。

第二，自动生成API文档。 这个功能简直是懒人福音。以前写完接口还得手动维护Swagger文档，文档和代码不同步是常态，前端天天追着我说"这个接口文档上写的是返回string，你咋给我返回了个int？"现在FastAPI直接根据代码生成文档，代码改了文档自动更新，终于不用背这个锅了。

第三，Pydantic类型校验。 Python作为一门动态类型语言，在大型项目里真的是又爱又恨。爱它的灵活，恨它的不安全。Pydantic完美解决了这个问题，而且性能还不错（底层是Rust写的，v2版本之后更是起飞）。

环境搭建和第一个接口

废话不多说，直接上代码。我默认你已经装好了Python 3.9+（别问我为什么不是3.8，因为3.8的生命周期快到了，而且一些新特性真的好用）。

# 创建虚拟环境，这个习惯一定要有
python -m venv venv
source venv/bin/activate  # Windows用 venv\Scripts\activate

# 安装FastAPI和uvicorn
pip install fastapi uvicorn[standard]

然后写一个最简单的Hello World：

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="内部工具平台API",
    description="给前端那帮兄弟用的接口",
    version="v0.1.0"
)

@app.get("/")
async def root():
    return {"message": "Hello World，别催了，在写了"}

@app.get("/health")
async def health_check():
    return {"status": "ok", "version": "v0.1.0"}

启动服务：

uvicorn main:app --reload --host 0.0.0.0 --port 8000

打开浏览器访问 http://localhost:8000/docs，你就能看到自动生成的Swagger文档了。第一次看到的时候我真的"卧槽"了一下，这也太方便了吧。

路由和参数校验，别再用request.form了

以前用Flask写接口，参数校验是这样的：

# Flask时代的噩梦
@app.route('/user', methods=['POST'])
def create_user():
    name = request.json.get('name')
    age = request.json.get('age')
    
    if not name:
        return jsonify({"error": "name is required"}), 400
    if not isinstance(age, int) or age < 0 or age > 150:
        return jsonify({"error": "invalid age"}), 400
    
    # 终于开始干正事了...

每个接口都要写一堆校验逻辑，重复代码能写几百行。在FastAPI里，这一切都优雅了：

from fastapi import FastAPI, Query, Path, Body
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# 定义数据模型，这就是Pydantic的威力
class UserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=50, description="用户名")
    age: int = Field(..., ge=0, le=150, description="年龄")
    email: Optional[str] = Field(None, description="邮箱，可选")
    role: str = Field("user", description="角色，默认user")

class UserResponse(BaseModel):
    id: int
    name: str
    age: int
    email: Optional[str]
    role: str

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # 到这里的时候，参数已经校验完毕了
    # 如果校验失败，FastAPI会自动返回422和详细的错误信息
    # 你只需要关心业务逻辑
    
    # 模拟数据库操作
    return UserResponse(
        id=1,
        name=user.name,
        age=user.age,
        email=user.email,
        role=user.role
    )

# 路径参数校验
@app.get("/users/{user_id}")
async def get_user(
    user_id: int = Path(..., ge=1, description="用户ID，必须大于0")
):
    return {"user_id": user_id}

# 查询参数校验
@app.get("/users")
async def list_users(
    page: int = Query(1, ge=1, description="页码"),
    size: int = Query(20, ge=1, le=100, description="每页数量"),
    keyword: Optional[str] = Query(None, description="搜索关键词")
):
    return {
        "page": page,
        "size": size,
        "keyword": keyword,
        "data": []
    }

这段代码看着简单，但背后FastAPI帮你做了太多事情：

1. 自动解析请求体JSON
2. 自动校验字段类型和约束
3. 校验失败自动返回422和详细的错误信息（哪个字段错了、期望什么类型、实际什么值）
4. 自动生成Swagger文档，包括参数说明和示例

前端同事第一次用我们接口的时候，直接看着文档就把对接搞定了，以前那种"接口联调一整天"的场景再也没出现过。

依赖注入，这才是FastAPI的精髓

依赖注入（Dependency Injection）这个词听起来很高大上，但其实在FastAPI里用起来非常直观。简单来说，就是把一些公共的逻辑抽出来，需要的时候注入到接口里。

最常见的场景就是鉴权和数据库连接：

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

# 模拟一个用户数据库
fake_users_db = {
    "token123": {"username": "admin", "role": "admin"},
    "token456": {"username": "zhangsan", "role": "user"},
}

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    """鉴权依赖：验证token并返回当前用户"""
    token = credentials.credentials
    user = fake_users_db.get(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

def require_admin(user: dict = Depends(get_current_user)):
    """权限依赖：要求当前用户是admin"""
    if user["role"] != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin privileges required"
        )
    return user

# 需要登录的接口
@app.get("/me")
async def get_me(user: dict = Depends(get_current_user)):
    return {"username": user["username"], "role": user["role"]}

# 需要管理员权限的接口
@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: dict = Depends(require_admin)  # 依赖可以嵌套！
):
    return {"message": f"User {user_id} deleted by {admin['username']}"}

依赖注入的好处不仅仅是代码复用，更重要的是可测试性。写单元测试的时候，你可以轻松地把依赖mock掉，不用真的去连数据库或者调外部服务。这一点在写测试的时候体会特别深。

还有一个很实用的场景是数据库Session管理：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 数据库Session依赖
async def get_db():
    db = SessionLocal()
    try:
        yield db  # 注意这里用yield，FastAPI会在请求结束后执行finally
    finally:
        db.close()

@app.get("/items")
async def read_items(db: Session = Depends(get_db)):
    # db会自动注入，请求结束后自动关闭
    items = db.query(Item).all()
    return items

这里有个坑要提醒大家：get_db用的是yield而不是return。FastAPI会在请求处理完成后自动执行finally块里的代码，确保数据库连接被正确释放。如果你用return，连接就不会被关闭，高并发下连接池直接打满，线上就等着OOM吧。别问我怎么知道的，去年双11的时候我们组就出过这个事故，当时排查了一晚上，最后发现是一个同事把yield写成了return。

中间件和异常处理，生产环境必备

写demo和上生产是两码事。生产环境你得考虑日志、CORS、异常处理、请求耗时统计等等。下面是我总结的一套生产级中间件配置：

import time
import logging
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 生命周期管理（FastAPI v0.100+推荐写法）
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    logger.info("🚀 Application starting up...")
    # 初始化数据库连接池、Redis连接等
    yield
    # 关闭时执行
    logger.info("👋 Application shutting down...")
    # 清理资源

app = FastAPI(lifespan=lifespan)

# CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend.com"],  # 别用*，安全审计过不了
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    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(round(process_time * 1000, 2))
    
    # 记录慢请求
    if process_time > 1.0:  # 超过1秒的请求记录warning
        logger.warning(
            f"Slow request: {request.method} {request.url.path} "
            f"took {process_time:.2f}s"
        )
    
    return response

# 全局异常处理
class BusinessError(Exception):
    """自定义业务异常"""
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message

@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=200,  # 业务异常返回200，通过code区分
        content={
            "code": exc.code,
            "message": exc.message,
            "data": None
        }
    )

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # 未预期的异常，记录详细日志但返回通用错误信息
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={
            "code": 500,
            "message": "Internal server error",
            "data": None
        }
    )

这里有个设计上的考虑：业务异常我返回的是HTTP 200，通过响应体里的code字段来区分。这是因为我们前端统一封装了请求库，所有非200的HTTP状态码都会走错误拦截。如果后端返回400、404这些，前端会当成网络错误处理，用户体验不好。当然这个看团队约定，没有绝对的对错。

异步编程的正确姿势

FastAPI原生支持异步，但不是所有场景都适合用async。这里有个很多新手容易踩的坑：

# ❌ 错误示范：在async函数里调用同步阻塞代码
@app.get("/bad")
async def bad_example():
    # 这个requests.get是同步阻塞的！
    # 它会阻塞整个事件循环，导致其他请求也被卡住
    import requests
    response = requests.get("https://api.example.com/data")
    return response.json()

# ✅ 正确方式1：使用异步HTTP客户端
@app.get("/good1")
async def good_example1():
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

# ✅ 正确方式2：如果只有同步库，用run_in_executor
@app.get("/good2")
async def good_example2():
    import asyncio
    import requests
    
    loop = asyncio.get_event_loop()
    # 把同步操作丢到线程池里执行
    response = await loop.run_in_executor(
        None, 
        lambda: requests.get("https://api.example.com/data")
    )
    return response.json()

# ✅ 正确方式3：直接定义为普通函数，FastAPI会自动放到线程池
@app.get("/good3")
def good_example3():
    # 注意这里没有async，FastAPI会把它放到线程池执行
    import requests
    response = requests.get("https://api.example.com/data")
    return response.json()

第三种方式很多人不知道。FastAPI的设计很聪明：如果你的路由函数是async def，它会在事件循环里直接执行；如果是普通的def，它会自动放到线程池里执行，不会阻塞事件循环。所以如果你用的库没有异步版本（比如一些老的数据库驱动），直接写成普通函数就行。

项目结构，别把所有代码塞一个文件

写demo的时候一个main.py就够了，但实际项目一定要做好目录划分。下面是我用的项目结构，经过几个项目验证，还算合理：

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用入口
│   ├── config.py            # 配置管理
│   ├── dependencies.py      # 公共依赖
│   ├── models/              # 数据库模型（SQLAlchemy）
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/             # Pydantic模型（请求/响应）
│   │   ├── __init__.py
│   │   └── user.py
│   ├── api/                 # 路由模块
│   │   ├── __init__.py
│   │   ├── router.py        # 路由汇总
│   │   └── v1/
│   │       ├── __init__.py
│   │       └── user.py
│   ├── services/            # 业务逻辑层
│   │   ├── __init__.py
│   │   └── user.py
│   ├── repositories/        # 数据访问层
│   │   ├── __init__.py
│   │   └── user.py
│   └── core/                # 核心模块
│       ├── __init__.py
│       ├── security.py      # 安全相关
│       └── exceptions.py    # 自定义异常
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_user.py
├── alembic/                 # 数据库迁移
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

这个结构的核心思想是分层：

• schemas层：定义请求和响应的数据结构，纯Pydantic模型，不依赖数据库
• models层：数据库表结构映射，SQLAlchemy模型
• repositories层：数据访问，封装数据库操作
• services层：业务逻辑，调用repositories
• api层：路由定义，调用services，做参数校验和响应格式化

分层的好处是每一层的职责清晰，改动影响范围可控。比如换数据库的时候，只需要改repositories层和models层，业务逻辑和接口定义完全不用动。

配置管理，别把密码硬编码

这个应该是老生常谈了，但还是忍不住要提。用pydantic-settings来管理配置，支持环境变量和.env文件：

# app/config.py
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # 应用配置
    APP_NAME: str = "内部工具平台"
    DEBUG: bool = False
    API_V1_PREFIX: str = "/api/v1"
    
    # 数据库配置
    DATABASE_URL: str = "postgresql://localhost/mydb"
    DB_POOL_SIZE: int = 20
    DB_MAX_OVERFLOW: int = 10
    
    # Redis配置
    REDIS_URL: str = "redis://localhost:6379/0"
    
    # JWT配置
    SECRET_KEY: str = "your-secret-key"  # 生产环境用环境变量覆盖
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
    
    # CORS配置
    CORS_ORIGINS: list[str] = ["http://localhost:3000"]
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

然后创建一个.env文件（记得加到.gitignore里）：

DEBUG=false
DATABASE_URL=postgresql://user:pass@prod-db:5432/mydb
SECRET_KEY=super-secret-key-that-nobody-knows
REDIS_URL=redis://prod-redis:6379/0

部署到生产环境

最后说说部署。我们用的是Docker + K8s，下面是Dockerfile：

FROM python:3.11-slim

WORKDIR /app

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

# 复制代码
COPY . .

# 非root用户运行
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

# 启动命令，注意workers数量
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

关于workers数量，有个经验公式：workers = CPU核数 * 2 + 1。但如果是IO密集型的服务（大部分Web服务都是），可以适当多一些。我们线上8核的机器一般开12-16个worker。

还有一个小tips：如果你的服务有大量的数据库操作，建议把--workers设小一点，把连接池设大一点。因为每个worker都有自己独立的连接池，worker太多会导致数据库连接数爆炸。我们之前出过一次事故，开了32个worker，每个worker连接池20，直接把数据库的max_connections打满了。

一些心得体会

写了这么多，最后总结几点个人感受：

1. FastAPI确实好用，但也不是万能的。如果你的项目很简单，就几个CRUD接口，Flask可能更合适，毕竟FastAPI的学习成本还是比Flask高一些的。

2. 类型提示很重要。刚开始用FastAPI的时候觉得写类型提示很烦，但写多了之后发现这真的是好习惯。代码可读性提高了，IDE的自动补全也更好用了，重构的时候也更有信心。

3. 异步不是银弹。不要为了异步而异步，如果你的业务逻辑主要是CPU密集型（比如图片处理、数据分析），异步反而会更慢，因为Python的GIL在那里。

4. 文档即代码。FastAPI的自动文档功能真的改变了我的工作方式。以前写完接口还得专门抽时间写文档，现在文档就是代码的一部分，代码改了文档自动更新，维护成本几乎为零。

好了，已经凌晨四点了，明天还得早起开周会。希望这篇文章能帮到正在学FastAPI的你。如果有什么问题，欢迎在评论区交流。

不说了，我得去把那个OOM的监控告警处理一下，老王那边肯定又在群里@我了。

---

最后更新：2024年，于某个远程办公的深夜