作者:阿柴
发布时间:2025年9月18日
标签:Python, FastAPI, Web开发, API, 后端, Pydantic, OpenAPI


一、为什么选择 FastAPI?

在 Python 的 Web 框架生态中,Django 和 Flask 长期占据主导地位。但近年来,一个新兴的框架 FastAPI 正在迅速崛起,成为构建高性能 API 的首选。

FastAPI 的核心优势可以用三个词概括:

:基于 Starlette 和 Pydantic,性能接近 Node.js 和 Go
智能:自动类型提示 + 自动补全,开发体验极佳
标准:自动生成 OpenAPI 文档(Swagger UI 和 ReDoc)

它特别适合:

  • 构建 RESTful API
  • 机器学习模型服务化(ML as a Service)
  • 微服务架构
  • 前后端分离项目

二、FastAPI 是什么?

FastAPI 是一个用于构建 API 的现代、快速(高性能)的 Python Web 框架。

  • 基于 Python 3.8+ 的类型提示(Type Hints)
  • 自动生成交互式 API 文档(Swagger UI 和 ReDoc)
  • 自动请求数据验证、序列化、错误处理
  • 支持异步(async/await
  • 与 Pydantic 深度集成,保证数据安全

官网:https://fastapi.tiangolo.com


三、快速开始:5分钟搭建第一个 API

1. 安装 FastAPI 和 Uvicorn

bash

pip install fastapi uvicorn -i https://pypi.tuna.tsinghua.edu.cn/simple
  • fastapi:核心框架
  • uvicorn:ASGI 服务器,用于运行 FastAPI 应用

2. 创建 main.py

python

from fastapi import FastAPI

# 创建 FastAPI 实例
app = FastAPI()

# 定义一个路由
@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

# 路径参数
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

3. 运行服务

bash

uvicorn main:app --reload
  • main:appmain.py 文件中的 app 实例
  • --reload:开发模式下自动重启

启动后访问:

  • API 文档:http://127.0.0.1:8000/docs(Swagger UI)
  • ReDoc 文档:http://127.0.0.1:8000/redoc

Tips: 无需任何配置,文档自动生成!


四、核心特性详解

1. 类型提示 + 自动验证

FastAPI 利用 Python 的类型提示,自动完成数据验证。

python

from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
    name: str
    price: float
    is_offer: Optional[bool] = None

@app.post("/items/")
def create_item(item: Item):
    return {"item_name": item.name, "price": item.price}
  • 如果请求体中 price 传了字符串 "abc",FastAPI 会自动返回 422 错误。
  • IDE 会自动补全 item.nameitem.price 等字段。

2. 自动生成 API 文档

FastAPI 自动生成符合 OpenAPI 标准的文档:

文档类型 地址 特点
Swagger UI /docs 交互式测试,可直接发送请求
ReDoc /redoc 更美观的文档展示

3. 异步支持(async/await)

python

import asyncio

@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(1)
    return {"message": "异步请求完成"}

适用于高并发、I/O 密集型场景(如数据库查询、文件读写、调用第三方 API)。


4. 依赖注入系统(Dependency Injection)

FastAPI 提供强大的依赖注入机制,用于共享逻辑(如数据库连接、身份验证)。

python

from fastapi import Depends

def common_params(q: str = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
def read_items(common: dict = Depends(common_params)):
    return common

五、项目结构建议(初学者友好)

my_fastapi_app/
├── main.py              # 入口文件
├── api/
│   └── v1/
│       ├── __init__.py
│       ├── routes/
│       │   ├── users.py
│       │   └── items.py
│       └── api_router.py
├── schemas/             # Pydantic 模型
│   ├── user.py
│   └── item.py
├── services/            # 业务逻辑
│   ├── user_service.py
│   └── item_service.py
├── models/              # 数据库模型(SQLAlchemy)
│   └── user.py
└── db/                  # 数据库配置
    └── session.py

这种分层结构有助于实现 路由层(API)与 服务层(业务逻辑)的解耦。


六、实战:构建一个用户管理 API

1. 模拟数据库(db/fake_db.py

我们先用一个简单的列表模拟数据库,便于理解。

python

# db/fake_db.py
from typing import List, Optional
from models.user import UserInDB

# 模拟数据库存储
users_db: List[UserInDB] = [
    UserInDB(id=1, username="alice", email="alice@example.com", hashed_password="fakehash123"),
    UserInDB(id=2, username="bob", email="bob@example.com", hashed_password="fakehash456"),
]

Tips: 在真实项目中,这里会是 SQLAlchemy 的 Session 或数据库连接池。


2. 数据模型(models/user.py

定义数据库中存储的用户结构(包含密码哈希等敏感字段)。

python

# models/user.py
from pydantic import BaseModel

class UserInDB(BaseModel):
    id: int
    username: str
    email: str
    hashed_password: str  # 实际项目中应使用 passlib 等加密

3. API 数据模型(schemas/user.py

定义对外暴露的 API 接口数据结构,遵循 最小暴露原则

python

# schemas/user.py
from pydantic import BaseModel
from typing import Optional

# 创建用户时需要的字段
class UserCreate(BaseModel):
    username: str
    email: str
    password: str

# 返回给客户端的用户信息(不含密码)
class UserOut(BaseModel):
    id: int
    username: str
    email: Optional[str] = None

# 更新用户时的可选字段
class UserUpdate(BaseModel):
    username: Optional[str] = None
    email: Optional[str] = None

4. 服务层:业务逻辑(services/user_service.py

这是核心业务逻辑的所在地。它不关心 HTTP 协议,只关心“怎么处理数据”。

python

# services/user_service.py
from typing import List, Optional
from models.user import UserInDB
from schemas.user import UserCreate, UserOut, UserUpdate
from db.fake_db import users_db
import hashlib

class UserService:
    def __init__(self):
        self.db = users_db

    def _hash_password(self, password: str) -> str:
        """简单模拟密码哈希(生产环境请用 passlib)"""
        return hashlib.sha256(password.encode()).hexdigest()

    def get_all_users(self) -> List[UserOut]:
        """获取所有用户(仅返回公开信息)"""
        return [
            UserOut(id=user.id, username=user.username, email=user.email)
            for user in self.db
        ]

    def get_user_by_id(self, user_id: int) -> Optional[UserOut]:
        """根据 ID 查找用户"""
        for user in self.db:
            if user.id == user_id:
                return UserOut(id=user.id, username=user.username, email=user.email)
        return None

    def get_user_by_username(self, username: str) -> Optional[UserInDB]:
        """根据用户名查找(用于登录验证,返回含密码哈希的完整对象)"""
        for user in self.db:
            if user.username == username:
                return user
        return None

    def create_user(self, user: UserCreate) -> UserOut:
        """创建新用户"""
        # 检查用户名是否已存在
        if self.get_user_by_username(user.username):
            raise ValueError(f"用户名 {user.username} 已存在")

        # 生成新 ID
        new_id = max([u.id for u in self.db]) + 1 if self.db else 1

        # 哈希密码并存入“数据库”
        hashed_pw = self._hash_password(user.password)
        new_user = UserInDB(
            id=new_id,
            username=user.username,
            email=user.email,
            hashed_password=hashed_pw
        )
        self.db.append(new_user)

        return UserOut(id=new_user.id, username=new_user.username, email=new_user.email)

    def update_user(self, user_id: int, update_data: UserUpdate) -> Optional[UserOut]:
        """更新用户信息"""
        for user in self.db:
            if user.id == user_id:
                if update_data.username:
                    user.username = update_data.username
                if update_data.email:
                    user.email = update_data.email
                return UserOut(id=user.id, username=user.username, email=user.email)
        return None

    def delete_user(self, user_id: int) -> bool:
        """删除用户"""
        for i, user in enumerate(self.db):
            if user.id == user_id:
                self.db.pop(i)
                return True
        return False

服务层特点

  • 不依赖 FastAPI
  • 可被多个路由复用
  • 易于单元测试
  • 与数据库/文件系统解耦

5. 路由层:API 接口(api/v1/routes/users.py

负责接收 HTTP 请求,调用服务层,返回响应。

python

# api/v1/routes/users.py
from fastapi import APIRouter, HTTPException, status
from typing import List
from schemas.user import UserCreate, UserOut, UserUpdate
from services.user_service import UserService

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "未找到用户"}}
)

# 创建服务实例(真实项目中可通过依赖注入)
user_service = UserService()

@router.get("/", response_model=List[UserOut])
def list_users():
    """获取用户列表"""
    return user_service.get_all_users()

@router.get("/{user_id}", response_model=UserOut)
def get_user(user_id: int):
    """根据 ID 获取用户"""
    user = user_service.get_user_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户未找到")
    return user

@router.post("/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate):
    """创建新用户"""
    try:
        return user_service.create_user(user)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.put("/{user_id}", response_model=UserOut)
def update_user(user_id: int, update_data: UserUpdate):
    """更新用户信息"""
    user = user_service.update_user(user_id, update_data)
    if not user:
        raise HTTPException(status_code=404, detail="用户未找到")
    return user

@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int):
    """删除用户"""
    success = user_service.delete_user(user_id)
    if not success:
        raise HTTPException(status_code=404, detail="用户未找到")
    return  # 204 No Content

路由层特点

  • 处理 HTTP 状态码、异常
  • 调用服务层
  • 返回标准化响应
  • 使用 response_model 自动校验输出

6. 聚合路由(api/v1/api_router.py

将所有 v1 路由聚合在一起。

python

# api/v1/api_router.py
from fastapi import APIRouter
from api.v1.routes.users import router as users_router

api_router = APIRouter(prefix="/api/v1")
api_router.include_router(users_router)

7. 应用入口(main.py

python

# main.py
from fastapi import FastAPI
from api.v1.api_router import api_router

app = FastAPI(
    title="用户管理 API",
    description="一个基于 FastAPI 的分层架构示例",
    version="1.0.0"
)

app.include_router(api_router)

@app.get("/")
def root():
    return {"message": "欢迎使用用户管理 API!访问 /docs 查看文档"}

七、部署建议

开发环境

bash

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

生产环境(推荐使用 Gunicorn + Uvicorn)

bash

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

Tips: 生产环境不要使用 --reload


八、FastAPI vs Flask vs Django

特性 FastAPI Flask Django
性能 ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐⭐
类型提示 ✅ 原生支持
自动生成文档 ❌(需扩展) ❌(需扩展)
异步支持 ✅(Flask 2.0+)
学习曲线 中等 简单 复杂
适用场景 API 服务、ML 服务 小型项目、原型 全栈项目、CMS

九、总结

FastAPI 凭借其 高性能、智能提示、自动文档、类型安全 等特性,正在成为 Python API 开发的新标准。

无论你是:

  • 刚入门后端开发的新手
  • 想将机器学习模型部署为服务的数据科学家
  • 寻找更高效工具的资深开发者

FastAPI 都值得你花一天时间学习和尝试。


参考资料

  • 官方文档:https://fastapi.tiangolo.com
  • GitHub:https://github.com/tiangolo/fastapi
  • Pydantic:https://pydantic.dev

欢迎关注我的技术博客,获取更多 Python 和后端开发部署实战内容!
💬 欢迎在评论区交流你的 FastAPI 使用体验!

更多推荐