Fastapi框架总览与核心架构

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 [e-mail]("/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()