1. 架构概览
FastAPI 采用分层架构设计,基于 Starlette ASGI 框架构建,通过继承和组合的方式实现了丰富的功能。其核心设计理念是:高性能、易用性和类型安全。
1.1 整体架构图
graph TB
subgraph "应用层 (Application Layer)"
A[FastAPI Application] --> A1[路径操作装饰器]
A1 --> A2[@app.get/@app.post等]
A --> A3[全局配置]
A3 --> A4[中间件配置]
A3 --> A5[异常处理配置]
A3 --> A6[依赖注入配置]
end
subgraph "路由层 (Routing Layer)"
B[APIRouter] --> B1[APIRoute]
B1 --> B2[路径匹配]
B1 --> B3[参数解析]
B1 --> B4[请求验证]
B --> B5[WebSocket路由]
B5 --> B6[APIWebSocketRoute]
end
subgraph "依赖注入层 (Dependency Injection)"
C[Dependencies System] --> C1[Dependant 分析]
C1 --> C2[依赖解析器]
C2 --> C3[参数注入]
C --> C4[安全依赖]
C4 --> C5[OAuth2/API Key等]
end
subgraph "请求/响应层 (Request/Response)"
D[Request Handler] --> D1[请求体解析]
D1 --> D2[Pydantic验证]
D --> D3[响应序列化]
D3 --> D4[JSON/HTML/其他响应]
end
subgraph "文档生成层 (Documentation)"
E[OpenAPI Generator] --> E1[Schema 生成]
E1 --> E2[Swagger UI]
E1 --> E3[ReDoc]
E --> E4[类型信息提取]
end
subgraph "基础设施层 (Infrastructure)"
F[Starlette Base] --> F1[ASGI 应用]
F1 --> F2[中间件栈]
F2 --> F3[异常处理]
F --> F4[事件系统]
F4 --> F5[启动/关闭事件]
end
A --> B
B --> C
C --> D
A --> E
B --> E
A --> F
B --> F
1.2 核心类继承关系
classDiagram
class Starlette {
+routes: List[BaseRoute]
+middleware: List[Middleware]
+exception_handlers: Dict
+on_startup: List[Callable]
+on_shutdown: List[Callable]
+__call__(scope, receive, send)
}
class FastAPI {
+title: str
+version: str
+openapi_url: str
+docs_url: str
+router: APIRouter
+dependency_overrides: Dict
+get(path) Callable
+post(path) Callable
+openapi() Dict
+include_router(router)
}
class Router {
+routes: List[BaseRoute]
+add_route(path, endpoint)
+route(path, methods)
}
class APIRouter {
+prefix: str
+tags: List[str]
+dependencies: List[Depends]
+add_api_route(path, endpoint)
+get(path) Callable
+post(path) Callable
}
class Route {
+path: str
+endpoint: Callable
+methods: Set[str]
+matches(scope)
+__call__(scope, receive, send)
}
class APIRoute {
+dependant: Dependant
+response_model: Type
+body_field: ModelField
+get_route_handler()
}
Starlette <|-- FastAPI
Router <|-- APIRouter
Route <|-- APIRoute
FastAPI --> APIRouter : contains
APIRouter --> APIRoute : creates
2. FastAPI 类深度分析
2.1 FastAPI 类的核心实现
class FastAPI(Starlette):
"""
FastAPI 主应用类,继承自 Starlette
主要职责:
1. 管理路由系统 (通过 APIRouter)
2. 处理依赖注入
3. 生成 OpenAPI 文档
4. 管理中间件和异常处理
5. 提供便捷的装饰器方法
"""
def __init__(
self,
*,
debug: bool = False, # 调试模式
routes: Optional[List[BaseRoute]] = None, # 路由列表
title: str = "FastAPI", # API 标题
version: str = "0.1.0", # API 版本
openapi_url: Optional[str] = "/openapi.json", # OpenAPI schema 地址
docs_url: Optional[str] = "/docs", # Swagger UI 地址
redoc_url: Optional[str] = "/redoc", # ReDoc 地址
dependencies: Optional[Sequence[Depends]] = None, # 全局依赖
default_response_class: Type[Response] = JSONResponse, # 默认响应类
# ... 更多参数
):
# 保存配置参数
self.title = title
self.version = version
self.openapi_url = openapi_url
# ... 其他配置
# 创建核心路由器
self.router = APIRouter(
routes=routes,
dependencies=dependencies,
default_response_class=default_response_class,
dependency_overrides_provider=self, # 依赖覆盖提供者
)
# 初始化异常处理器
self.exception_handlers = {}
self.exception_handlers[HTTPException] = http_exception_handler
self.exception_handlers[RequestValidationError] = request_validation_exception_handler
# 依赖覆盖机制(用于测试)
self.dependency_overrides = {}
# 调用 Starlette 父类初始化
super().__init__(
debug=debug,
routes=self.router.routes, # 将路由传递给 Starlette
middleware=middleware,
exception_handlers=self.exception_handlers,
)
# 设置内置路由(OpenAPI、Docs等)
self.setup()
2.2 路径操作装饰器的实现机制
FastAPI 的核心特性之一是便捷的装饰器方法(如 @app.get
、@app.post
等)。让我们深入分析其实现:
# FastAPI 类中的装饰器方法实现
def get(
self,
path: str,
*,
response_model: Any = Default(None),
status_code: Optional[int] = None,
tags: Optional[List[Union[str, Enum]]] = None,
dependencies: Optional[Sequence[Depends]] = None,
summary: Optional[str] = None,
description: Optional[str] = None,
# ... 更多参数
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
GET 请求装饰器
实现原理:
1. 接收路径和配置参数
2. 返回一个装饰器函数
3. 装饰器函数将被装饰的函数注册到路由器
Args:
path: URL 路径模式
response_model: 响应数据模型
status_code: 默认状态码
tags: OpenAPI 标签
dependencies: 路径操作依赖
summary: 操作摘要
description: 操作描述
Returns:
装饰器函数
"""
# 委托给路由器的 get 方法
return self.router.get(
path,
response_model=response_model,
status_code=status_code,
tags=tags,
dependencies=dependencies,
summary=summary,
description=description,
# ... 传递所有参数
)
# APIRouter 中的实现
def get(
self,
path: str,
**kwargs
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
APIRouter 的 GET 装饰器实现
"""
return self.api_route(path, methods=["GET"], **kwargs)
def api_route(
self,
path: str,
*,
methods: Optional[List[str]] = None,
**kwargs
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
通用路由装饰器,所有 HTTP 方法装饰器的底层实现
Returns:
装饰器函数,用于装饰路径操作函数
"""
def decorator(func: DecoratedCallable) -> DecoratedCallable:
# 将函数注册为 API 路由
self.add_api_route(
path,
func,
methods=methods,
**kwargs
)
# 返回原始函数(不修改)
return func
return decorator
2.3 路由注册流程时序图
sequenceDiagram
participant Dev as 开发者
participant App as FastAPI应用
participant Router as APIRouter
participant Route as APIRoute
participant Dep as 依赖分析器
Dev->>App: @app.get("/users/{user_id}")
App->>Router: router.get(path, **kwargs)
Router->>Router: api_route(path, methods=["GET"])
Router->>Router: 返回装饰器函数
Dev->>Router: 装饰器装饰函数 def get_user(user_id: int)
Router->>Router: add_api_route(path, func, methods=["GET"])
Router->>Route: 创建 APIRoute 实例
Route->>Dep: 分析函数依赖 get_dependant(func)
Dep->>Dep: 解析参数类型和依赖
Dep->>Route: 返回 Dependant 对象
Route->>Route: 生成路由处理器 get_route_handler()
Router->>Router: routes.append(route)
Router->>App: 路由注册完成
3. APIRouter 深度分析
3.1 APIRouter 的核心职责
APIRouter 是 FastAPI 路由系统的核心,负责:
- 路由组织:将相关的路径操作分组
- 参数传播:将通用配置传播到所有路径操作
- 依赖管理:管理路由级别的依赖
- 前缀处理:为路径添加公共前缀
class APIRouter(routing.Router):
"""
API 路由器,用于组织和管理路径操作
继承自 Starlette 的 Router,增加了 FastAPI 特有功能:
1. 自动 OpenAPI 文档生成
2. 依赖注入支持
3. 响应模型验证
4. 路径参数和查询参数的自动解析
"""
def __init__(
self,
*,
prefix: str = "", # 路径前缀
tags: Optional[List[str]] = None, # OpenAPI 标签
dependencies: Optional[Sequence[Depends]] = None, # 路由依赖
default_response_class: Type[Response] = JSONResponse, # 默认响应类
responses: Optional[Dict] = None, # 额外响应定义
callbacks: Optional[List[BaseRoute]] = None, # OpenAPI 回调
route_class: Type[APIRoute] = APIRoute, # 路由类
# ... 更多参数
):
# 保存配置
self.prefix = prefix
self.tags = tags or []
self.dependencies = list(dependencies or [])
self.default_response_class = default_response_class
self.responses = responses or {}
self.route_class = route_class
# 调用父类初始化
super().__init__(routes=routes or [])
def add_api_route(
self,
path: str,
endpoint: Callable,
**kwargs
) -> None:
"""
添加 API 路由
这是所有路径操作装饰器的最终实现
"""
# 合并路由器级别和路径操作级别的配置
combined_tags = (self.tags or []) + (kwargs.get("tags") or [])
combined_dependencies = list(self.dependencies) + list(kwargs.get("dependencies") or [])
combined_responses = {**self.responses, **(kwargs.get("responses") or {})}
# 处理路径前缀
if self.prefix:
path = self.prefix.rstrip("/") + "/" + path.lstrip("/")
# 创建 APIRoute 实例
route = self.route_class(
path,
endpoint,
tags=combined_tags,
dependencies=combined_dependencies,
responses=combined_responses,
**kwargs
)
# 添加到路由列表
self.routes.append(route)
3.2 路由匹配和分发机制
flowchart TD
A[HTTP 请求到达] --> B[Starlette 应用]
B --> C[遍历路由列表]
C --> D{路径匹配?}
D -->|否| E[下一个路由]
E --> C
D -->|是| F{HTTP 方法匹配?}
F -->|否| G[返回 405 Method Not Allowed]
F -->|是| H[APIRoute.matches()]
H --> I[提取路径参数]
I --> J[调用 APIRoute.__call__()]
J --> K[依赖解析和注入]
K --> L[参数验证]
L --> M[调用路径操作函数]
M --> N[响应序列化]
N --> O[返回 HTTP 响应]
4. APIRoute 详细分析
4.1 APIRoute 的生命周期
class APIRoute(routing.Route):
"""
API 路由,FastAPI 路由系统的核心类
主要功能:
1. 解析函数签名和类型提示
2. 生成依赖图 (Dependant)
3. 创建请求处理器
4. 处理响应模型验证
5. 生成 OpenAPI 文档信息
"""
def __init__(
self,
path: str,
endpoint: Callable,
*,
response_model: Any = Default(None),
status_code: Optional[int] = None,
methods: Optional[Set[str]] = None,
dependencies: Optional[Sequence[params.Depends]] = None,
# ... 更多参数
):
"""
APIRoute 初始化过程:
1. 分析函数签名
2. 解析依赖关系
3. 创建响应模型
4. 生成处理器
"""
self.path = path
self.endpoint = endpoint
# 1. 处理响应模型
if isinstance(response_model, DefaultPlaceholder):
# 从函数返回类型注解推断响应模型
return_annotation = get_typed_return_annotation(endpoint)
if lenient_issubclass(return_annotation, Response):
response_model = None
else:
response_model = return_annotation
self.response_model = response_model
# 2. 分析函数依赖关系
self.dependant = get_dependant(
path=path,
call=endpoint,
dependencies=dependencies,
)
# 3. 处理请求体字段
self.body_field = get_body_field(
dependant=self.dependant,
name=endpoint.__name__
)
# 4. 生成唯一ID(用于 OpenAPI)
if isinstance(generate_unique_id_function, DefaultPlaceholder):
current_generate_unique_id = generate_unique_id
else:
current_generate_unique_id = generate_unique_id_function
self.unique_id = current_generate_unique_id(self)
# 5. 调用父类初始化
super().__init__(
path,
endpoint=endpoint,
methods=methods,
)
# 6. 生成 ASGI 应用(请求处理器)
self.app = request_response(self.get_route_handler())
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
"""
生成路由处理器
这是 FastAPI 的核心:将用户的简单函数转换为完整的 ASGI 应用
"""
return get_request_handler(
dependant=self.dependant, # 依赖图
body_field=self.body_field, # 请求体字段
status_code=self.status_code, # 状态码
response_class=self.response_class, # 响应类
response_field=self.secure_cloned_response_field, # 响应字段
response_model_include=self.response_model_include,
response_model_exclude=self.response_model_exclude,
response_model_by_alias=self.response_model_by_alias,
response_model_exclude_unset=self.response_model_exclude_unset,
response_model_exclude_defaults=self.response_model_exclude_defaults,
response_model_exclude_none=self.response_model_exclude_none,
dependency_overrides_provider=self.dependency_overrides_provider,
embed_body_fields=self._embed_body_fields,
)
4.2 请求处理器的实现
def get_request_handler(
dependant: Dependant,
body_field: Optional[ModelField] = None,
status_code: Optional[int] = None,
response_class: Union[Type[Response], DefaultPlaceholder] = Default(JSONResponse),
response_field: Optional[ModelField] = None,
# ... 更多参数
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
"""
生成请求处理器函数
这个函数返回一个异步函数,该函数:
1. 解析 HTTP 请求
2. 解决依赖注入
3. 调用用户函数
4. 处理响应
Args:
dependant: 依赖图,包含所有需要解析的依赖
body_field: 请求体字段定义
status_code: 默认状态码
response_class: 响应类
response_field: 响应模型字段
Returns:
异步请求处理器函数
"""
# 解析响应类
if isinstance(response_class, DefaultPlaceholder):
actual_response_class = JSONResponse
else:
actual_response_class = response_class
async def app(request: Request) -> Response:
"""
实际的请求处理器
这是每个 HTTP 请求都会执行的函数
"""
try:
# 1. 创建异步退出栈(用于管理依赖的生命周期)
async with AsyncExitStack() as async_exit_stack:
# 2. 解决所有依赖
solved_result = await solve_dependencies(
request=request,
dependant=dependant,
dependency_overrides_provider=dependency_overrides_provider,
async_exit_stack=async_exit_stack,
embed_body_fields=embed_body_fields,
)
# 3. 检查依赖解析错误
if solved_result.errors:
raise RequestValidationError(
_normalize_errors(solved_result.errors)
)
# 4. 调用用户的路径操作函数
assert dependant.call is not None
if iscoroutinefunction(dependant.call):
# 异步函数
raw_response = await dependant.call(**solved_result.values)
else:
# 同步函数,在线程池中运行
raw_response = await run_in_threadpool(
dependant.call, **solved_result.values
)
# 5. 处理响应
if isinstance(raw_response, Response):
# 用户返回了 Response 对象,直接使用
if raw_response.background is None:
raw_response.background = solved_result.background_tasks
return raw_response
else:
# 用户返回了数据,需要序列化
response_data = raw_response
# 应用响应模型(如果有)
if response_field:
response_data = _prepare_response_content(
response_data,
exclude_unset=response_model_exclude_unset,
exclude_defaults=response_model_exclude_defaults,
exclude_none=response_model_exclude_none,
)
# Pydantic 模型序列化
response_data = jsonable_encoder(
response_data,
include=response_model_include,
exclude=response_model_exclude,
by_alias=response_model_by_alias,
exclude_unset=response_model_exclude_unset,
exclude_defaults=response_model_exclude_defaults,
exclude_none=response_model_exclude_none,
)
# 创建响应对象
response = actual_response_class(
content=response_data,
status_code=status_code or 200,
background=solved_result.background_tasks,
)
return response
except Exception as e:
# 异常会被 FastAPI 的异常处理器捕获
raise
return app
5. 架构设计模式分析
5.1 装饰器模式
FastAPI 大量使用装饰器模式来简化 API 开发:
# 用户代码(简洁)
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> User:
return User(id=user_id, name="John")
# FastAPI 内部转换为复杂的路由配置
# 等价于:
app.add_api_route(
"/users/{user_id}",
get_user,
methods=["GET"],
response_model=User,
# ... 大量其他配置
)
5.2 依赖注入模式
FastAPI 实现了强大的依赖注入系统:
# 依赖定义
def get_db() -> Database:
return Database()
def get_current_user(token: str = Depends(get_token)) -> User:
return verify_token(token)
# 自动注入
@app.get("/users/me")
async def get_me(
user: User = Depends(get_current_user), # 自动注入当前用户
db: Database = Depends(get_db) # 自动注入数据库
) -> User:
return user
5.3 中介者模式
FastAPI 应用作为中介者,协调各个组件:
graph TB
A[FastAPI Application] --> B[Router]
A --> C[Dependencies]
A --> D[Middleware]
A --> E[Exception Handlers]
A --> F[OpenAPI Generator]
B --> C
C --> D
D --> E
B --> F
5.4 建造者模式
FastAPI 使用建造者模式构建复杂的路由对象:
# APIRoute 的构建过程
class APIRoute:
def __init__(self, ...):
# 1. 分析函数签名
self.dependant = get_dependant(call=endpoint, ...)
# 2. 处理请求体
self.body_field = get_body_field(dependant=self.dependant, ...)
# 3. 生成处理器
self.app = request_response(self.get_route_handler())
# 4. 设置 OpenAPI 信息
self.openapi_extra = openapi_extra
# ...
6. 性能优化设计
6.1 异步优化
FastAPI 原生支持异步,充分利用 Python 的异步特性:
- 异步路径操作:支持
async def
函数 - 异步依赖:依赖解析过程是异步的
- 异步中间件:中间件栈是异步的
- 线程池集成:同步函数自动在线程池中运行
6.2 类型系统优化
FastAPI 通过类型系统实现:
- 编译时优化:类型检查在应用启动时完成
- 序列化优化:预编译序列化逻辑
- 验证优化:Pydantic 提供高性能验证
6.3 缓存策略
FastAPI 使用多层缓存:
class FastAPI:
def openapi(self) -> Dict[str, Any]:
# OpenAPI schema 缓存
if not self.openapi_schema:
self.openapi_schema = get_openapi(...) # 只生成一次
return self.openapi_schema
7. 可扩展性设计
7.1 插件架构
FastAPI 通过以下方式支持扩展:
- 自定义路由类:可以继承
APIRoute
实现自定义逻辑 - 自定义中间件:标准的 ASGI 中间件接口
- 自定义异常处理器:灵活的异常处理机制
- 依赖覆盖:支持依赖替换(主要用于测试)
7.2 模块化设计
# 主应用
app = FastAPI()
# 用户模块路由器
users_router = APIRouter(prefix="/users", tags=["users"])
# 产品模块路由器
products_router = APIRouter(prefix="/products", tags=["products"])
# 组合到主应用
app.include_router(users_router)
app.include_router(products_router)
这种设计使大型应用可以按功能模块组织,保持代码的清晰和可维护性。
8. 总结
FastAPI 的架构设计体现了现代 Python Web 框架的最佳实践:
- 基于标准:完全兼容 ASGI 标准,与 Python 类型系统深度集成
- 高性能:异步优先,充分利用现代 Python 的性能特性
- 易用性:通过装饰器和依赖注入简化开发体验
- 类型安全:静态类型检查和运行时验证相结合
- 自动化:自动生成 API 文档和客户端代码
- 可扩展:模块化设计支持大型应用开发
下一章我们将深入分析 FastAPI 的路由系统实现细节。