Python API接口开发实战：从Flask到FastAPI的踩坑全记录

作者： /

开篇：如果你以为写API就是装个Flask打印“Hello World”，那接下来的内容可能会让你头皮发麻。

刚开始我也以为这玩意儿很简单，结果连踩三个坑，一个比一个恶心。先说第一个：用Flask写一个正经的RESTful API，结果返回的数据格式被前端骂了三天。

先看最基础的东西。你写API，至少得知道JSON怎么返回吧？很多人上来就这样：

“python
from flask import Flask

app = Flask(__name__)

@app.route('/user/')
def get_user(user_id):
user = {'id': user_id, 'name': '张三', 'age': 25}
return user # 这特么能跑？能，但问题在后面
`

这段代码在Flask里确实能返回JSON，因为Flask会自动帮你序列化字典。但你要返回一个列表呢？比如：

`python
@app.route('/users')
def get_users():
users = [{'id': 1, 'name': '张三'}, {'id': 2, 'name': '李四'}]
return users # 报错：TypeError: 'list' object is not callable
`

Flask这个设计真的反人类。正确的写法是：

`python
from flask import Flask, jsonify

@app.route('/users')
def get_users():
users = [{'id': 1, 'name': '张三'}, {'id': 2, 'name': '李四'}]
return jsonify(users) # 必须用jsonify包裹
`

另一个坑：返回状态码怎么搞？ 我见过有人直接return '错误'，然后前端拿着200状态码在那儿debug半天。标准做法：

`python
@app.route('/login', methods=['POST'])
def login():
data = request.get_json()
if not data or 'username' not in data:
return jsonify({'error': '缺少用户名'}), 400 # 第二个参数是状态码
# 业务逻辑...
return jsonify({'token': 'xxx'}), 200
`

这里有个细节：request.get_json() 返回的是dict，不是字符串。如果你用 request.form 取数据，那是表单格式，JSON数据取不到。这个官方文档文档不够清晰，我当时愣是看了半小时才明白。

核心：当你从Flask切换到FastAPI，你会发现世界清净了——但前提是你要躲开那些暗坑。

从Flask到FastAPI：不仅仅是换框架

先说为什么换FastAPI。我用Flask写了一个小项目，接口数量从10个涨到50个时，代码已经成了一团意大利面。而且每次修改参数类型，前端就要来找我撕逼——因为Flask没有自动文档生成。

FastAPI天然支持OpenAPI文档，而且自带Swagger UI和ReDoc，部署后直接访问 /docs 就能看到接口文档，前端再也不用追着问“这个字段是string还是int”。

来看一个典型的FastAPI接口：

`python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class UserCreate(BaseModel):
username: str
email: str
age: Optional[int] = None # 可选字段，默认None

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

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
# 模拟数据库操作
user_dict = user.dict()
user_dict['id'] = 1
return user_dict

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
# 模拟查询
if user_id != 1:
raise HTTPException(status_code=404, detail="用户不存在")
return {"id": 1, "username": "张三", "email": "zhangsan@example.com", "age": 25}
`

这里有几个关键点：

  • Pydantic模型：UserCreate 和 UserResponse 是数据模型，FastAPI会自动做类型校验和文档生成。如果你传的age是字符串"abc"，它会直接返回422错误，省得你写一堆if-else。
  • async/await：FastAPI默认支持异步，但注意这不是必须的。如果你的业务逻辑是同步的（比如用SQLite），可以不加async，一样跑。但如果你用异步数据库驱动（如asyncpg、motor），就一定要加。
  • 状态码：直接用 status_code=201 指定，不用像Flask那样返回元组。

    • 但FastAPI也有坑。我第一次写异步接口时，用了 time.sleep() 来模拟耗时操作，结果整个服务卡住了。为什么？因为 time.sleep() 是同步的，它会阻塞事件循环。正确做法：

    `python
    import asyncio

    @app.get("/slow")
    async def slow_endpoint():
    await asyncio.sleep(5) # 用asyncio.sleep，不是time.sleep
    return {"message": "等了5秒"}
    `

    还有个技巧：依赖注入。这是FastAPI最骚的功能之一。比如你每个接口都要验证Token，传统写法是每个函数里复制粘贴验证代码。FastAPI让你这么写：

    `python
    from fastapi import Depends, Header
    from typing import Optional

    async def verify_token(authorization: Optional[str] = Header(None)):
    if not authorization:
    raise HTTPException(status_code=401, detail="未提供Token")
    # 验证逻辑...

    <

    p> return authorization # 返回值可以作为参数传递

    @app.get("/protected")
    async def protected_endpoint(token: str = Depends(verify_token)):
    return {"message": "验证通过", "token": token}
    `

    Depends 会自动调用 verify_token 函数，并把返回值注入到 token 参数里。这样每个接口只需要加一个 token: str = Depends(verify_token) 就搞定了验证，代码量减少40%。

    性能优化：从3.2秒降到0.8秒

    写API不能不提性能。我接手过一个项目，某个列表接口响应时间3.2秒，用户反馈“卡得像PPT”。原因是什么？N+1查询问题。

    比如查询用户列表时，每个用户还要查一次他的订单数量：

    `python

    反例：每个用户单独查询

    users = db.query(User).all()
    for user in users:
    order_count = db.query(Order).filter(Order.user_id == user.id).count()
    user.order_count = order_count
    `

    如果用户有100个，就要查100+1次数据库。改成连表查询或子查询后：

    `python
    from sqlalchemy.orm import aliased
    from sqlalchemy import func

    优化后：一次查询

    users_with_count = db.query(
    User,
    func.count(Ord

    er.id).label('order_count')
    ).outerjoin(
    Order, User.id == Order.user_id
    ).group_by(User.id).all()
    `

    outerjoin 确保没有订单的用户也会显示，count为0。这个改动让接口响应从3.2秒降到0.8秒。

    还有一个容易被忽视的点：数据库连接池。如果用Django ORM或者SQLAlchemy，默认连接池大小通常是5-10。并发一高，请求就会排队。我在生产环境遇到过：20个并发请求，10个直接超时。解决办法是在初始化时调整连接池：

    `python
    from sqlalchemy import create_engine

    engine = create_engine(
    "postgresql://user:pass@localhost/db",
    pool_size=20, # 连接池大小
    max_overflow=10, # 最大溢出连接数
    pool_pre_ping=True # 每次连接前检查是否有效
    )
    `

    pool_pre_ping=True 这个参数让我少加了一周的班——它会在从连接池取出连接时，先发一个SELECT 1检查连接是否还活着，避免“断连”导致的诡异错误。

    总结前：说了这么多，最后给你一些立竿见影的干货。

    总结一下，你可以立刻用的三个点

  • 接口文档自动生成：从Flask换成FastAPI，或者用Flask-RESTx（Flask的扩展），确保每个接口都有文档。前端不再来问“这个字段是啥类型”，省下的时间够你追两季番。
  • 数据模型复用：用Pydantic或Dataclass定义输入输出模型，不要手动写验证。比如 age: int 会自动拒绝字符串，不用你写 if not isinstance(age, int)。
  • 数据库查询优化：如果接口慢了，先查是不是N+1问题。用 explain 分析SQL，或者用工具如 slow_query_log 抓慢查询。别一上来就上Redis缓存——缓存能解决性能问题但引入数据一致性问题，得不偿失。

    • 最后说一个血的教训：永远不要把数据库密码硬编码在代码里。用环境变量或配置中心，比如：

    `python
    import os
    from dotenv import load_dotenv

    load_dotenv() # 从.env文件加载

    DATABASE_URL = os.getenv("DATABASE_URL")
    if not DATABASE_URL:
    raise ValueError("请设置DATABASE_URL环境变量")
    “

    我见过有人把密码写在GitHub公开仓库里，十分钟后被爬虫扫到，数据库被删光，勒索0.5个比特币。你猜最后怎么解决的？没有解决，数据全丢了。

    写API就是一个不断踩坑、填坑的过程。以上这些坑我全踩过，希望你读完能少走些弯路。

    滚动至顶部