Fastapi框架总览与核心架构
FastAPI 是一个基于 Python 的现代 Web 框架,专注于 高性能、高并发 和 开发效率,特别适合构建 异步 API 服务、微服务接口,同时在大模型接口封装中也广泛应用。它基于 Starlette(异步 Web 框架) 和 Pydantic(数据验证模型)构建,拥有非常清晰的架构层次和组件机制。
FastAPI 与其它框架的对比
框架 | 类型校验 | 异步支持 | 文档生成 | 性能 |
---|---|---|---|---|
Flask | ❌ | ⚠️ 部分支持 | ❌ | 中等 |
Django | ⚠️ 基本支持 | ❌ | ❌ | 中等偏低 |
FastAPI | ✅ 完整支持 | ✅ 完整支持 | ✅ 自动化 | 非常优秀 |
Sanic | ❌ | ✅ | ❌ | 优秀 |
Tornado | ❌ | ✅ | ❌ | 高 |
✅ 什么是“路由”?
🚏 路由定义:
“路由”就是 URL 到处理函数的映射。
在 Web 框架中,用户访问 /users/123,系统就需要知道该调用哪个函数来处理这个请求,这个“地址 → 函数”的映射关系就是路由。
🚧 路由定义包括三个部分:
项 | 含义 | 举例 |
---|---|---|
URL 路由 | 请求的路径(path) | /users/{user_id} 、/login |
请求方法 | HTTP 动作类型 | GET , POST , PUT , DELETE 等 |
请求参数规范 | 参数类型与位置 | 路径参数、查询参数、请求体、表单、文件 |
📌 什么是 HTTP 动作类型?
HTTP 是浏览器和服务器通信的协议,它有几种“动作”方式(也叫“方法”)来表示客户端希望干什么。
方法 | 含义 | 举例 |
---|---|---|
GET | 获取资源 | 浏览页面、获取数据 /items/1 |
POST | 创建资源 | 提交表单、新建数据 /login |
PUT | 修改整个资源 | 替换资源 /users/1 |
PATCH | 修改部分资源 | 改昵称 /users/1/name |
DELETE | 删除资源 | 删除用户 /users/1 |
✅ 什么是参数类型与位置?路径参数、查询参数、请求体、表单、文件 是什么意思?
📌 在 FastAPI 中,参数可以来自不同位置:
参数类型 | 来源 | 示例路径/形式 | FastAPI 接收方式 |
---|---|---|---|
路径参数 | URL 本身 | /items/123 → item_id=123 | 函数参数 |
查询参数 | URL 的 ?key=val | /search?q=apple | 函数参数 |
请求体(JSON) | POST/PUT 里的 body | {"name": "apple", "price": 1.2} | Pydantic 模型 |
表单参数 | HTML form 提交 | form 表单中的字段 | Form(...) |
文件上传 | 上传图片/文件 | 文件作为 body 的部分 | File(...) |
✅ Pydantic 是什么?
Pydantic 是一个数据验证框架,用于定义和校验模型结构。
from pydantic import BaseModelclass User(BaseModel):username: strage: intu = User(username="hao", age="20") # 会自动转成 int
在 FastAPI 中用于:
校验请求体
响应结构
自动生成文档
一、FastAPI 框架核心特性总览
特性 | 说明 |
---|---|
类型注解驱动开发 | 使用 Python 3.6+ 的类型提示,自动生成文档、校验入参 |
自动生成 OpenAPI 文档 | 自动提供 Swagger UI 和 Redoc 文档接口 |
基于 Starlette 异步内核 | 支持高性能的异步调用(WebSocket、Streaming 等) |
依赖注入机制 | 具备轻量级的依赖注入系统,适合模块化扩展 |
请求数据自动校验 | 使用 Pydantic 模型校验入参,异常处理机制优雅 |
支持 OAuth2/JWT 等认证 | 方便集成现代安全机制(OAuth2、JWT、Bearer Token 等) |
极高性能 | 性能可媲美 NodeJS / Go,比 Flask 和 Django 快数倍 |
1️⃣ 类型注解驱动开发(Type Hints = Contract + Logic)
✅ 描述
FastAPI 利用了 Python 3.6+ 的类型注解(Type Hints),不仅仅用于代码提示,更直接决定了入参校验、自动文档、序列化格式等。
✅ 示例
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):return {"item_id": item_id, "query": q}
✅ 效果
item_id: int → FastAPI 自动将 URL 参数转为 int,非 int 会报 422 Unprocessable Entity
q: str = None → 这是一个可选的 query 参数,自动解析 query string
自动生成 OpenAPI 文档,支持前端自动调试
2️⃣ 自动生成交互式 API 文档(OpenAPI + Swagger)
✅ 描述
FastAPI 内置支持生成:
/docs → 使用 Swagger UI
/redoc → 使用 ReDoc
/openapi.json → 原始 JSON 格式的 OpenAPI 规范
✅ 优点
前后端联调更高效
自动生成参数示例、响应格式、错误码定义
可以直接在线调用和测试接口
3️⃣ Pydantic 数据模型(Request 校验 & Response 构造)
✅ 描述
FastAPI 使用 Pydantic 作为请求体和响应体的数据校验与构造工具,拥有极强的类型支持、嵌套能力和默认值机制。
✅ 示例
from pydantic import BaseModelclass Item(BaseModel):name: strprice: floatis_offer: bool = False
@app.post("/items/")
def create_item(item: Item):return item
✅ 功能
自动将 JSON 请求体解析为 Python 对象
参数不合法时自动返回 422 错误 + 错误详情
同时可作为响应结构文档
4️⃣ 异步支持(基于 Starlette 的 async/await)
✅ 描述
FastAPI 完全基于 Starlette 的异步内核,原生支持异步视图函数和并发调用。
✅ 示例
import httpx@app.get("/async_request")
async def async_request():async with httpx.AsyncClient() as client:r = await client.get("https://example.com")return {"status": r.status_code}
✅ 优点
支持大规模并发(如多模型同时调用)
不阻塞主线程,适合 IO 密集型任务(数据库、网络请求等)
5️⃣ 依赖注入机制(Dependency Injection)
✅ 描述
通过 Depends(…) 提供函数或对象的依赖注入,FastAPI 自动解析依赖调用关系。
✅ 示例
from fastapi import Dependsdef get_db():db = connect_to_db()try:yield dbfinally:db.close()@app.get("/users/")
def get_users(db = Depends(get_db)):return db.query("SELECT * FROM users")
✅ 优点
把常用逻辑(数据库连接、鉴权、安全校验)封装成可复用模块
可缓存、可作用域控制、支持异步
二、FastAPI 核心架构分层
┌────────────────────────────┐
│ 1. 接口层(API Router) │ ← 请求入口,定义路由
├────────────────────────────┤
│ 2. 控制器层(View/Controller)│ ← 调用服务逻辑 + 响应格式
├────────────────────────────┤
│ 3. 服务层(Service Layer) │ ← 编排业务逻辑/异步任务/模型调用
├────────────────────────────┤
│ 4. 数据访问层(Repository) │ ← 访问数据库/外部接口
├────────────────────────────┤
│ 5. 模型层(Schema/Model) │ ← Pydantic 模型 / ORM 模型
├────────────────────────────┤
│ 6. 框架层(Middleware/Startup)│ ← 中间件 / 生命周期 / DI 注入
└────────────────────────────┘
1️⃣ 接口层(Router Layer)
作用:定义 URL 路由、请求方法、请求参数规范
实现方式:使用 APIRouter 或直接在 FastAPI() 实例中注册路由
from fastapi import APIRouter
router = APIRouter()@router.get("/users/{user_id}")
def get_user(user_id: int):...
特点:
类型注解决定参数类型、是否必填
自动生成文档
路由模块可拆分,适合模块化开发(如 /users/、/items/)
2️⃣ 控制器层(Controller/View Layer)
作用:处理请求流程控制,组织调用服务层,并定义返回模型
实现方式:通常和路由函数合并,或封装为类/函数依赖项
@router.post("/login", response_model=Token)
def login(user: LoginForm, service: AuthService = Depends()):return service.authenticate(user)
特点:
包含参数校验、服务调用、响应封装
可做基本异常捕捉、权限判断
可封装为类结构,实现面向对象风格控制器
3️⃣ 服务层(Service Layer / UseCase)
作用:编排业务逻辑,组织多个数据源调用,支持异步任务
实现方式:写为类或独立函数,通过依赖注入使用
class AuthService:def __init__(self, user_repo: UserRepo = Depends()):self.user_repo = user_repodef authenticate(self, login_data: LoginForm) -> Token:user = self.user_repo.find_by_username(login_data.username)...
4️⃣ 数据访问层(Repository / DAO)
作用:封装对数据库、缓存、第三方 API 的访问逻辑
实现方式:使用 ORM(如 SQLAlchemy)、ODM 或直接 SQL 封装
class UserRepo:def __init__(self, db: Session = Depends(get_db)):self.db = dbdef find_by_username(self, username: str):return self.db.query(User).filter(User.username == username).first()
好处:
逻辑复用
屏蔽数据源实现差异
支持单元测试 Mock
5️⃣ 模型层(Pydantic Schema + ORM Model)
5.1 ✅ 请求/响应模型(Pydantic)
class UserIn(BaseModel):username: strpassword: strclass UserOut(BaseModel):id: intusername: str
途:
入参校验
返回结果标准化
文档自动生成
5.2 ✅ 数据库模型(SQLAlchemy/ORM)
class User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True)username = Column(String)password_hash = Column(String)
6️⃣ 框架层(Middleware / Startup / Dependency)
作用:
实现跨请求逻辑(如日志、权限、限流)
注册应用级资源(连接池、缓存、队列)
提供依赖注入机制
✅ 中间件示例:
@app.middleware("http")
async def add_log(request: Request, call_next):response = await call_next(request)response.headers["X-Service"] = "fastapi"return response
✅ 启动事件:
@app.on_event("startup")
def init():global redis_poolredis_pool = init_redis_pool()
✅ 依赖注入定义:
def get_settings():return Settings()@app.get("/config")
def read_config(settings: Settings = Depends(get_settings)):return settings
🔁 各层之间的调用关系
前端请求 → Router → Controller → Service → Repository → DB/API
↓
Pydantic Schema
↓
响应模型 → 自动文档
🔔 客户端发出请求(URL + Headers + Body)
↓
🔍 FastAPI 解析 HTTP 方法 + URL 模式
↓
📦 匹配对应函数 + 分析函数签名
↓
📥 解析参数来源(路径 / 查询 / 请求体 / 表单 / 文件)
↓
✅ 自动类型转换(str → int 等)+ Pydantic 校验
↓
🚀 执行函数逻辑,返回 JSON 响应
三、核心模块拆解与职责
模块/层 | 主要职责 | FastAPI 实现方式 |
---|---|---|
1️⃣ 路由注册 | 定义请求路径、HTTP 方法、绑定处理函数 | @app.get/post(...) ,APIRouter |
2️⃣ 视图函数 | 处理参数解析、调用业务逻辑、构造响应 | 普通 def/async def 函数 |
3️⃣ 数据模型 | 请求体/响应体结构定义、类型校验、文档生成 | Pydantic 模型 |
4️⃣ 依赖注入 | 注入共享资源(数据库连接、配置、认证信息等) | Depends(...) |
5️⃣ 异步执行 | 非阻塞调用 IO(如数据库、HTTP 请求、模型推理) | async def + await + Uvicorn |
6️⃣ 中间件与事件 | 拦截请求/响应,全局处理逻辑,应用启动/关闭初始化清理 | @app.middleware(...) , @app.on_event(...) |
1️⃣ 路由注册模块(Routing System)
✅ 作用:
将 HTTP 方法 + URL 路径映射到某个 Python 函数上(即路由处理函数)
✅ 典型写法:
from fastapi import FastAPIapp = FastAPI()@app.get("/users/{user_id}")
def get_user(user_id: int):return {"user_id": user_id}
📌 特点:
自动绑定参数(路径参数、查询参数、Body 参数)
自动纳入 OpenAPI 文档
可模块化拆分,适合大型项目管理
2️⃣ 视图函数模块(View / Controller)
✅ 作用:
接收客户端请求参数(由 FastAPI 自动注入)
调用服务逻辑
构造标准化响应
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):items = db.query(Item).offset(skip).limit(limit).all()return items
📌 特点:
通常放在 controller 或 api/ 中
可组合依赖项(如认证信息、数据库连接)
返回的数据会被自动 JSON 序列化
3️⃣ 数据模型模块(Pydantic Model)
✅ 作用:
定义请求/响应结构
类型强校验 + 默认值管理
自动生成 API 文档说明
from pydantic import BaseModelclass UserIn(BaseModel):username: strpassword: strclass UserOut(BaseModel):id: intusername: str
📌 用法:
函数参数用作请求体接收(POST / PUT)
返回值用作响应模型
@app.post("/users/", response_model=UserOut)
def create_user(user: UserIn):...return UserOut(id=1, username=user.username)
4️⃣ 依赖注入模块(Dependency Injection)
✅ 作用:
将一些“重复调用或需要共享状态”的组件,如数据库、配置、认证信息,通过 Depends 注入进来
from fastapi import Dependsdef get_db():db = SessionLocal()try:yield dbfinally:db.close()@app.get("/users/")
def list_users(db=Depends(get_db)):return db.query(User).all()
📌 应用场景:
数据库连接
鉴权与用户信息注入
公共服务类(如缓存服务、Token 解析器)
5️⃣ 异步执行模块(Async Runtime)
✅ 作用:
高性能并发处理,如:
并发处理多个 HTTP 请求
非阻塞 IO(数据库/模型调用)
后台任务、事件流处理
@app.get("/news/")
async def fetch_news():async with httpx.AsyncClient() as client:resp = await client.get("https://newsapi.org/...")return resp.json()