0
0
0

app/routers/orders.py

♀胡浩宇
2025-06-17 09:06
阅读 782

从“Hello World”到高并发：我在项目中如何用 FastAPI 搞定 Python 后端开发

从“Hello World”到高并发：我在项目中如何用 FastAPI 搞定 Python 后端开发

大家好，我是小陈，一个在创业公司干了快五年的后端工程师。最近一年我一直在使用 FastAPI 来搭建公司的新业务线接口系统，也踩了不少坑，今天就来和大家分享一下我的真实经历。如果你是刚入门 Python 后端的新手，或者想从 Flask、Django 等框架转过来试试 FastAPI，这篇文章应该能给你一些实际的帮助。

背景：为什么选 FastAPI？

事情要从我们团队的一次技术重构说起。那时我们正在做一个 SaaS 类的订单管理系统，需要快速响应业务需求的同时支持高并发访问。原来的系统是基于 Flask 搭建的，虽然功能都实现了，但随着用户量增长，性能瓶颈逐渐显现出来——特别是 API 接口响应慢、缺乏异步支持、文档不统一等问题越来越严重。

我们当时面临几个关键诉求：

  • 快速迭代
  • 高性能（尤其是 I/O 密集型操作）
  • 自动化接口文档
  • 强类型支持（Python 的 type hints）

于是我们开始考察主流的 Python Web 框架，像 Django REST framework、Tornado、Quart、Starlette……最后选择了 FastAPI。说白了，它正好是我们想要的一切的一个最佳折中：

  • 性能接近 Node.js 和 Go（得益于 Starlette 和 Pydantic）
  • 自带交互式 Swagger 和 ReDoc 文档
  • 支持同步 + 异步编程
  • 极强的类型提示支持，配合 IDE 提升代码质量
  • 上手成本低，适合初创团队快速搭建原型

接下来我就结合我们项目的实际开发过程，讲讲我是怎么一步步把 FastAPI 推进生产环境的。

第一阶段：新手入门 —— Hello World 到基础 CRUD

我们的第一个任务就是重构已有的订单管理模块。之前这部分是用 Flask 做的，代码结构杂乱、没有单元测试、而且接口文档靠口口相传。

于是我决定从头来过，先搭个最小可行 FastAPI 应用看看感觉。

初始化项目结构

project/
├── app/
│   ├── main.py
│   ├── routers/
│   │   └── orders.py
│   ├── models/
│   │   └── order_model.py
│   ├── schemas/
│   │   └── order_schema.py
│   ├── database.py
└── requirements.txt

这种结构借鉴了 Django 和 DRF 的方式，便于后期维护，同时避免所有代码都写在一个文件里。routers 存放各个业务模块的路由，models 是数据库模型，schemas 处理请求参数校验。

实现一个最简单的 Order 接口


from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter(prefix="/orders")

class OrderCreate(BaseModel):
    product_id: int
    quantity: int

@router.post("/")
def create_order(order: OrderCreate):
    return {"message": "Order created", "data": order}

这个例子很简单，但你会发现，即使是最基础的创建接口，FastAPI 已经提供了自动的数据校验和 OpenAPI 文档生成。你只需要运行 uvicorn app.main:app --reload，然后打开 /docs 页面，就可以看到自动生成的交互式文档。

数据库连接与 ORM 设计

我们选择的是 SQLAlchemy + asyncpg（PostgreSQL）作为 ORM 方案。不过 FastAPI 默认推荐的是 Tortoise ORM 或者 SQLAlchemy 的异步支持，我们在实践中发现如果只是做一般的 CRUD，并不需要太复杂的 ORM，所以最终采用了传统的 SQLAlchemy + asyncpg 结合的中间层封装方式。

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

然后在 FastAPI 中通过 Dependency Injection 注入数据库连接：

from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

这样每个接口都可以通过 db: Session = Depends(get_db) 获取数据库上下文。

小插曲：一开始为了节省时间，我们直接用了 flask-sqlalchemy 那套逻辑，结果 FastAPI 不吃这一套，导致大量阻塞调用拖垮异步优势。后来及时重构为真正的 asyncpg 驱动，性能明显提升。

第二阶段：实战进阶 —— 面对真实业务场景的挑战

随着订单模块稳定上线，我们逐步将其他模块迁移进来，也开始面对更复杂的问题：权限控制、日志记录、异常处理、缓存优化……

挑战1：权限验证怎么做才优雅？

刚开始我们是在每个接口函数中加判断语句，比如：

if user.role != "admin":
    raise HTTPException(status_code=403, detail="权限不足")

这显然不可持续。后来我们引入了 FastAPI 的依赖注入机制，将认证与鉴权封装成独立组件。

# dependencies.py

from fastapi import Depends, HTTPException

def verify_token(token: str = Header(...)):
    # 模拟token校验
    if token != "valid_token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return True

def require_admin(verify: bool = Depends(verify_token)):
    # 更进一步的权限检查
    if not is_admin:
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return True

然后在接口中使用：

@router.get("/all", dependencies=[Depends(require_admin)])
def get_all_orders(db: Session = Depends(get_db)):
    ...

这样做之后，权限体系可以灵活扩展，还能复用于不同模块，大大提高了系统的可维护性。

挑战2：异步任务处理怎么做？

我们有些接口需要处理外部服务调用，例如调用微信支付接口或第三方物流查询，这些操作耗时长，不能阻塞主线程。

FastAPI 内置了对异步的支持，所以我们直接用了 async def 定义接口：

import httpx

@router.post("/pay")
async def make_payment(order_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.post("https://payment.example.com/pay", json={"order_id": order_id})
    return response.json()

但是很快我们就发现问题：当并发请求特别多的时候，异步请求之间互相影响，某些请求会因为等待太久而失败。后来我们在项目中接入了 Redis 队列 + Celery worker 进行任务解耦。

挑战3：接口性能下降，怎么办？

有个接口负责拉取近7天的所有订单数据进行汇总计算。这个接口在高峰期会出现超时问题。

排查后发现是两个原因：

  1. 查询语句没加索引
  2. 查询返回的数据量太大，没有分页

解决方案也很简单：

  • 给订单表加上合适的索引（例如按 created_at 和 status 分组）
  • 对大结果集启用分页（Page/Size），并限制最大条数
  • 使用缓存策略，如 Redis 缓存高频查询结果

改造后的代码如下：

@router.get("/recent")
def get_recent_orders(
    db: Session = Depends(get_db),
    page: int = 1,
    size: int = 50
):
    start = (page - 1) * size
    end = page * size
    return db.query(Order).filter(Order.created_at >= yesterday).offset(start).limit(size).all()

加上 Redis 缓存：

cache = redis.Redis(host='redis', port=6379)

def get_cached_result(key):
    cached = cache.get(key)
    if cached:
        return json.loads(cached)
    return None

def set_cache_result(key, data):
    cache.setex(key, 60 * 10, json.dumps(data))

有了这些优化后，接口响应时间平均降低了 40%，成功率也大幅提升。

第三阶段：部署上线与运维实战

FastAPI 本身非常轻量，部署也不难，但上线过程中我们还是遇到了一些典型问题。

生产部署方案

我们采用的是以下架构：

  • 应用层：Gunicorn + Uvicorn Worker（生产环境）
  • 反向代理：Nginx
  • 数据库：PostgreSQL
  • 缓存：Redis
  • 日志收集：ELK（Elasticsearch + Logstash + Kibana）
  • 监控：Prometheus + Grafana

部署命令参考：

gunicorn -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000 --workers 4

其中 -k 参数指定使用 Uvicorn 的 worker，这样才能发挥 FastAPI 的异步优势；workers 数量一般设为 CPU 核心数的倍数（通常是核心数乘以2）。

日志处理建议

一定要做好日志分级输出（info/warn/error），并且集中到 ELK 里方便分析。建议使用 logging.config.dictConfig 配置：

import logging
import logging.config

LOGGING_CONFIG = {
    'version': 1,
    'handlers': {
        'default': {
            'level': 'INFO',
            'formatter': 'default',
            'class': 'logging.StreamHandler',
        },
    },
    'formatters': {
        'default': {
            'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
        }
    },
}

logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger(__name__)

还可以加入 request id 打印，方便追踪问题：

from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = str(uuid.uuid4())
        logger.info(f"Starting request {request_id}")
        response = await call_next(request)
        logger.info(f"Finished request {request_id}")
        return response

middleware = [
    Middleware(RequestIDMiddleware),
]
app = FastAPI(middleware=middleware)

性能监控怎么做？

接入 Prometheus 指标暴露比较简单：

  • 安装依赖：pip install prometheus-fastapi-instrumentator
  • 添加初始化代码：
from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app)

然后在 /metrics 接口就能拿到默认的指标了。再配合 Grafana 展示 QPS、错误率、响应时间等，整个系统的可观测性就起来了。

最后：给新手的几点建议

FastAPI 真的挺适合刚入门的同学上手的，尤其是如果你熟悉 Python 并且有前端交互经验的话。但我也想分享几个亲身体会的小建议：

  1. 不要一开始就追求完美设计，先跑起来再说。FastAPI 的优势就在于足够轻量，快速迭代，慢慢打磨。
  2. 重视类型提示（Type Hints），哪怕一开始你还不熟悉 Pydantic，也建议坚持使用 BaseModel 来定义数据结构，这对后期维护帮助巨大。
  3. 合理划分路由层级，别让所有接口都挤在一个文件里。你可以按模块拆，也可以按版本拆（v1/v2），方便后续升级。
  4. 异步不是万能药，很多地方同步就够了，除非你是处理 IO 密集型任务，否则没必要强制用 async。
  5. 早做性能压测，上线前务必模拟高并发场景，可以用 Locust 测试接口表现。
  6. 多看官方文档和社区案例，FastAPI 社区活跃度很高，很多常见问题都有现成的解决方案。
  7. 监控和日志必须搞起来，不然出了问题就像瞎子摸象，根本无从下手。

数据流转过程-1

总结

从最初的“Hello World”到现在的百万级请求系统，FastAPI 在我们这里证明了自己的实力。它不仅让我们提升了开发效率，也让后端系统变得更加健壮、易维护。

对于初学者来说，FastAPI 的学习曲线不算陡峭，只要肯动手敲代码、多试错，很快就能上手。更重要的是，它帮你养成良好的编码习惯：注重类型安全、文档规范、性能优化。

如果你打算进入后端开发领域，或者想找一个能让你快速实现想法的工具，FastAPI 绝对值得尝试。希望这篇文章能成为你踏上这条路的第一块垫脚石。

作者简介：
小陈，五年全栈工程师，专注后端与系统架构。热爱开源技术，喜欢研究各种高性能服务架构。欢迎关注我的 GitHub 和 Medium，一起交流成长 😊

文中提及的技术栈可根据具体项目需求灵活调整，不构成硬性要求。

标签:Python综合

评论 0

最热最新
暂无评论

为你推荐

暂无相关推荐
匿名用户Lv.1
0
影响力
0
文章
0
粉丝

相关推荐

暂无相关推荐

精选内容

暂无精选内容