FastAPI 深度剖析：從源碼理解現代 Python Web 框架的設計精髓
引言：FastAPI 的崛起
FastAPI 是由 Sebastián Ramírez 於 2018 年創建的現代 Python Web 框架，它結合了以下技術的優點：
- Starlette：高效能 ASGI 框架
- Pydantic：數據驗證和序列化
- Python Type Hints：自動文檔和編輯器支援
FastAPI 的設計目標是：
- 快速開發：類型提示自動驗證
- 高效能：媲美 Go 和 Node.js
- 標準化：基於 OpenAPI 和 JSON Schema
- 易學易用：直觀的 API 設計
本文將從源碼層面深入剖析 FastAPI 的核心實現。
第一章：ASGI 協議基礎
1.1 WSGI vs ASGI
graph TB
subgraph "WSGI (同步)"
W_REQ[Request] --> W_APP[Application]
W_APP --> W_RES[Response]
W_NOTE[單一請求-響應<br/>阻塞式]
end
subgraph "ASGI (異步)"
A_REQ[Request] --> A_APP[Application]
A_APP --> A_RES[Response]
A_APP --> A_WS[WebSocket]
A_APP --> A_STREAM[Streaming]
A_NOTE[多種協議<br/>非阻塞]
end1.2 ASGI 接口定義
# ASGI Application 介面
async def app(scope: dict, receive: Callable, send: Callable) -> None:
"""
scope: 連接資訊（URL、headers、等）
receive: 接收訊息的協程
send: 發送訊息的協程
"""
pass
# Scope 結構
scope = {
"type": "http", # "http", "websocket", "lifespan"
"asgi": {"version": "3.0"},
"http_version": "1.1",
"method": "GET",
"scheme": "https",
"path": "/users/123",
"query_string": b"name=john",
"headers": [
(b"host", b"example.com"),
(b"content-type", b"application/json"),
],
"server": ("example.com", 443),
"client": ("127.0.0.1", 12345),
}
# HTTP 請求生命週期
async def app(scope, receive, send):
# 接收請求體
body = b""
while True:
message = await receive()
body += message.get("body", b"")
if not message.get("more_body"):
break
# 發送響應
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"text/plain")],
})
await send({
"type": "http.response.body",
"body": b"Hello, World!",
})1.3 FastAPI 與 Starlette 的關係
graph TB
subgraph "FastAPI 架構層次"
FASTAPI[FastAPI]
STARLETTE[Starlette]
UVICORN[Uvicorn/Hypercorn]
ASYNCIO[asyncio]
FASTAPI --> |繼承| STARLETTE
STARLETTE --> |ASGI| UVICORN
UVICORN --> |事件循環| ASYNCIO
end
subgraph "FastAPI 增強功能"
DI[依賴注入]
VALID[Pydantic 驗證]
OPENAPI[OpenAPI 生成]
AUTH[安全認證]
end
FASTAPI --> DI
FASTAPI --> VALID
FASTAPI --> OPENAPI
FASTAPI --> AUTHFastAPI 繼承 Starlette：
# fastapi/applications.py
from starlette.applications import Starlette
class FastAPI(Starlette):
"""
FastAPI 繼承 Starlette，添加了：
- 依賴注入系統
- 請求/響應驗證
- OpenAPI 自動生成
- 安全認證工具
"""
def __init__(
self,
*,
debug: bool = False,
routes: list[BaseRoute] = None,
title: str = "FastAPI",
description: str = "",
version: str = "0.1.0",
openapi_url: str = "/openapi.json",
docs_url: str = "/docs",
redoc_url: str = "/redoc",
# ... 更多參數
):
# 創建專用路由器
self.router: routing.APIRouter = routing.APIRouter(
routes=routes,
dependency_overrides_provider=self,
on_startup=on_startup,
on_shutdown=on_shutdown,
lifespan=lifespan,
)
# 調用父類初始化
super().__init__(
debug=debug,
routes=self.router.routes,
middleware=middleware,
exception_handlers=exception_handlers,
)
# FastAPI 特有屬性
self.title = title
self.description = description
self.version = version
self.openapi_url = openapi_url第二章：路由系統深度解析
2.1 路由註冊流程
sequenceDiagram
participant Dev as Developer
participant App as FastAPI App
participant Router as APIRouter
participant Route as APIRoute
Dev->>App: @app.get("/users/{id}")
App->>Router: add_api_route()
Router->>Route: 創建 APIRoute
Route->>Route: 解析路徑參數
Route->>Route: 分析函數簽名
Route->>Route: 構建依賴圖
Route->>Route: 設置驗證模型2.2 裝飾器實現
# fastapi/routing.py
class APIRouter:
def get(
self,
path: str,
*,
response_model: Any = None,
status_code: int = 200,
tags: list[str] = None,
dependencies: Sequence[Depends] = None,
summary: str = None,
description: str = None,
# ... 更多參數
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""GET 請求裝飾器"""
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags,
dependencies=dependencies,
summary=summary,
description=description,
methods=["GET"],
)
def api_route(
self,
path: str,
*,
methods: list[str] = None,
**kwargs
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""通用路由裝飾器"""
def decorator(func: DecoratedCallable) -> DecoratedCallable:
self.add_api_route(
path,
func,
methods=methods,
**kwargs
)
return func
return decorator
def add_api_route(
self,
path: str,
endpoint: Callable[..., Any],
*,
methods: list[str] = None,
response_model: Any = None,
**kwargs
) -> None:
"""添加 API 路由"""
route = APIRoute(
path=self.prefix + path,
endpoint=endpoint,
methods=methods,
response_model=response_model,
dependency_overrides_provider=self.dependency_overrides_provider,
**kwargs
)
self.routes.append(route)2.3 APIRoute 類詳解
# fastapi/routing.py
class APIRoute(routing.Route):
def __init__(
self,
path: str,
endpoint: Callable[..., Any],
*,
response_model: Any = None,
status_code: int = 200,
dependencies: Sequence[params.Depends] = None,
# ...
) -> None:
self.path = path
self.endpoint = endpoint
self.response_model = response_model
self.status_code = status_code
# 核心：分析端點函數，構建依賴圖
self.dependant = get_dependant(
path=self.path_format,
call=self.endpoint,
)
# 收集所有路徑參數
self.path_params = get_flat_params(self.dependant)
# 設置響應類
if response_model:
self.response_field = create_model_field(
name="Response",
type_=response_model,
)
else:
self.response_field = None
# 生成實際的 ASGI 應用
self.app = request_response(self.get_route_handler())
def get_route_handler(self) -> Callable:
"""生成請求處理函數"""
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.response_field,
response_model_include=self.response_model_include,
response_model_exclude=self.response_model_exclude,
dependency_overrides_provider=self.dependency_overrides_provider,
)2.4 請求處理流程
graph TB
subgraph "請求處理流程"
REQ[HTTP Request] --> MATCH[路由匹配]
MATCH --> PARSE[解析請求數據]
PARSE --> VALIDATE[Pydantic 驗證]
VALIDATE --> DI[依賴注入解析]
DI --> ENDPOINT[執行端點函數]
ENDPOINT --> SERIALIZE[響應序列化]
SERIALIZE --> RES[HTTP Response]
end
VALIDATE --> |驗證失敗| ERR[422 Validation Error]
DI --> |依賴錯誤| ERR# fastapi/routing.py
def get_request_handler(
dependant: Dependant,
body_field: Optional[ModelField] = None,
status_code: int = 200,
response_class: Type[Response] = JSONResponse,
response_field: Optional[ModelField] = None,
# ...
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
"""生成請求處理器"""
async def app(request: Request) -> Response:
try:
# 1. 解析請求體
body = None
if body_field:
if is_body_form:
body = await request.form()
else:
body_bytes = await request.body()
if body_bytes:
body = await request.json()
# 2. 解析並驗證所有依賴
solved_result = await solve_dependencies(
request=request,
dependant=dependant,
body=body,
dependency_overrides_provider=dependency_overrides_provider,
)
values, errors, background_tasks, sub_response = solved_result
# 3. 處理驗證錯誤
if errors:
raise RequestValidationError(errors, body=body)
# 4. 調用端點函數
raw_response = await run_endpoint_function(
dependant=dependant,
values=values,
is_coroutine=is_coroutine,
)
# 5. 處理響應
if isinstance(raw_response, Response):
response = raw_response
else:
# 序列化響應數據
response_data = jsonable_encoder(
raw_response,
include=response_model_include,
exclude=response_model_exclude,
)
response = response_class(
content=response_data,
status_code=status_code,
)
# 6. 設置背景任務
if background_tasks:
response.background = background_tasks
return response
except Exception as e:
raise e
return app第三章：依賴注入系統
3.1 依賴注入概念
graph TB
subgraph "依賴注入示例"
ENDPOINT["/users/{user_id}"]
DEP1[get_db]
DEP2[get_current_user]
DEP3[verify_token]
ENDPOINT --> DEP1
ENDPOINT --> DEP2
DEP2 --> DEP3
end
subgraph "解析順序"
S1[1. verify_token]
S2[2. get_current_user]
S3[3. get_db]
S4[4. endpoint]
S1 --> S2 --> S3 --> S4
end3.2 Depends 類實現
# fastapi/params.py
class Depends:
"""依賴聲明類"""
def __init__(
self,
dependency: Optional[Callable[..., Any]] = None,
*,
use_cache: bool = True,
scope: Optional[str] = None, # "request" or "function"
):
self.dependency = dependency
self.use_cache = use_cache
self.scope = scope
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self.dependency!r})"
def Depends(
dependency: Optional[Callable[..., Any]] = None,
*,
use_cache: bool = True,
) -> Any:
"""
創建依賴注入聲明
用法:
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):
return db.query(Item).all()
"""
return params.Depends(dependency=dependency, use_cache=use_cache)3.3 依賴圖構建
# fastapi/dependencies/utils.py
def get_dependant(
*,
path: str,
call: Callable[..., Any],
name: Optional[str] = None,
security_scopes: Optional[list[str]] = None,
use_cache: bool = True,
scope: Optional[str] = None,
) -> Dependant:
"""分析函數簽名，構建依賴圖"""
# 獲取函數簽名
endpoint_signature = get_typed_signature(call)
signature_params = endpoint_signature.parameters
# 創建 Dependant 對象
dependant = Dependant(
call=call,
name=name,
path=path,
use_cache=use_cache,
scope=scope,
)
# 遍歷所有參數
for param_name, param in signature_params.items():
# 獲取參數類型和默認值
type_annotation = param.annotation
param_default = param.default
# 判斷參數類型
if isinstance(param_default, params.Depends):
# 依賴注入參數
sub_dependant = get_parameterless_sub_dependant(
depends=param_default,
path=path,
)
dependant.dependencies.append(sub_dependant)
elif isinstance(param_default, params.Path):
# 路徑參數
dependant.path_params.append(
create_model_field(
name=param_name,
type_=type_annotation,
default=param_default,
)
)
elif isinstance(param_default, params.Query):
# 查詢參數
dependant.query_params.append(
create_model_field(
name=param_name,
type_=type_annotation,
default=param_default,
)
)
elif isinstance(param_default, params.Header):
# 請求頭參數
dependant.header_params.append(...)
elif isinstance(param_default, params.Cookie):
# Cookie 參數
dependant.cookie_params.append(...)
elif isinstance(param_default, params.Body):
# 請求體參數
dependant.body_params.append(...)
elif type_annotation is Request:
# Request 對象
dependant.request_param_name = param_name
elif type_annotation is Response:
# Response 對象
dependant.response_param_name = param_name
elif type_annotation is BackgroundTasks:
# 背景任務
dependant.background_tasks_param_name = param_name
else:
# 默認作為查詢參數或請求體
...
return dependant3.4 依賴解析執行
# fastapi/dependencies/utils.py
async def solve_dependencies(
*,
request: Union[Request, WebSocket],
dependant: Dependant,
body: Optional[Any] = None,
background_tasks: Optional[BackgroundTasks] = None,
response: Optional[Response] = None,
dependency_overrides_provider: Optional[Any] = None,
dependency_cache: Optional[dict[tuple[Callable[..., Any], tuple[str]], Any]] = None,
) -> tuple[
dict[str, Any],
list[Any],
Optional[BackgroundTasks],
Response,
]:
"""解析所有依賴並執行"""
values: dict[str, Any] = {}
errors: list[Any] = []
if dependency_cache is None:
dependency_cache = {}
# 1. 遞歸解析子依賴
for sub_dependant in dependant.dependencies:
# 檢查緩存
call = sub_dependant.call
cache_key = (call, sub_dependant.scope)
if sub_dependant.use_cache and cache_key in dependency_cache:
solved = dependency_cache[cache_key]
else:
# 遞歸解析
solved = await solve_dependencies(
request=request,
dependant=sub_dependant,
body=body,
background_tasks=background_tasks,
response=response,
dependency_overrides_provider=dependency_overrides_provider,
dependency_cache=dependency_cache,
)
sub_values, sub_errors, bg_tasks, sub_response = solved
if sub_errors:
errors.extend(sub_errors)
continue
# 執行依賴函數
if is_coroutine_callable(call):
solved_result = await call(**sub_values)
else:
solved_result = await run_in_threadpool(call, **sub_values)
# 處理生成器（yield 依賴）
if inspect.isgenerator(solved_result) or inspect.isasyncgen(solved_result):
stack = request.scope.get("fastapi_astack")
solved_result = await stack.enter_async_context(
contextmanager_in_threadpool(solved_result)
)
# 緩存結果
if sub_dependant.use_cache:
dependency_cache[cache_key] = solved_result
values[sub_dependant.name] = solved_result
# 2. 解析路徑參數
path_values, path_errors = await solve_path_params(
dependant=dependant,
request=request,
)
values.update(path_values)
errors.extend(path_errors)
# 3. 解析查詢參數
query_values, query_errors = await solve_query_params(
dependant=dependant,
request=request,
)
values.update(query_values)
errors.extend(query_errors)
# 4. 解析請求頭
header_values, header_errors = await solve_header_params(
dependant=dependant,
request=request,
)
values.update(header_values)
errors.extend(header_errors)
# 5. 解析請求體
body_values, body_errors = await solve_body_params(
dependant=dependant,
body=body,
)
values.update(body_values)
errors.extend(body_errors)
# 6. 注入特殊對象
if dependant.request_param_name:
values[dependant.request_param_name] = request
if dependant.response_param_name:
values[dependant.response_param_name] = response
if dependant.background_tasks_param_name:
values[dependant.background_tasks_param_name] = background_tasks
return values, errors, background_tasks, response3.5 yield 依賴（上下文管理）
# 使用示例
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
app = FastAPI()
# yield 依賴會自動處理清理邏輯
def get_db():
db = SessionLocal()
try:
yield db # 注入 db
finally:
db.close() # 請求結束後執行
@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):
return db.query(Item).all()
# FastAPI 內部處理
# fastapi/concurrency.py
@asynccontextmanager
async def contextmanager_in_threadpool(
cm: ContextManager[T],
) -> AsyncIterator[T]:
"""將同步上下文管理器轉為異步"""
try:
yield await run_in_threadpool(cm.__enter__)
except Exception as e:
ok = await run_in_threadpool(cm.__exit__, type(e), e, None)
if not ok:
raise e
else:
await run_in_threadpool(cm.__exit__, None, None, None)第四章：Pydantic 整合與數據驗證
4.1 Pydantic 模型與 FastAPI
graph TB
subgraph "數據流"
RAW[原始請求數據<br/>JSON/Form]
PYDANTIC[Pydantic Model]
VALID[驗證後數據]
ENDPOINT[端點函數]
RAW --> |解析| PYDANTIC
PYDANTIC --> |驗證| VALID
VALID --> |注入| ENDPOINT
end
subgraph "錯誤處理"
PYDANTIC --> |驗證失敗| ERR[ValidationError]
ERR --> |轉換| HTTP_ERR[422 Response]
end4.2 請求體驗證
from pydantic import BaseModel, Field, validator
from typing import Optional
from fastapi import FastAPI, Body
app = FastAPI()
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: str = Field(..., regex=r"^[\w\.-]+@[\w\.-]+\.\w+$")
password: str = Field(..., min_length=8)
age: Optional[int] = Field(None, ge=0, le=150)
@validator('username')
def username_alphanumeric(cls, v):
assert v.isalnum(), 'must be alphanumeric'
return v
class Config:
json_schema_extra = {
"example": {
"username": "johndoe",
"email": "john@example.com",
"password": "secretpassword",
"age": 25
}
}
@app.post("/users/")
async def create_user(user: UserCreate):
# user 已經驗證完成
return {"username": user.username, "email": user.email}
# FastAPI 內部驗證流程
# fastapi/dependencies/utils.py
async def solve_body_params(
dependant: Dependant,
body: Any,
) -> tuple[dict[str, Any], list[Any]]:
"""解析並驗證請求體"""
values = {}
errors = []
for field in dependant.body_params:
# 從請求體提取數據
value = body.get(field.alias or field.name)
# Pydantic 驗證
v_, errors_ = field.validate(value, values, loc=("body", field.name))
if errors_:
errors.extend(errors_)
else:
values[field.name] = v_
return values, errors4.3 響應模型驗證
from pydantic import BaseModel
from typing import List
from fastapi import FastAPI
app = FastAPI()
class UserOut(BaseModel):
id: int
username: str
email: str
# 注意：不包含 password
class Config:
from_attributes = True # 支援 ORM 模式
class UserDB(BaseModel):
id: int
username: str
email: str
hashed_password: str
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
# 返回包含 hashed_password 的完整對象
user = UserDB(
id=user_id,
username="john",
email="john@example.com",
hashed_password="$2b$12$..."
)
# FastAPI 自動過濾，只返回 UserOut 定義的字段
return user
# 內部實現
# fastapi/routing.py
def _prepare_response_content(
res: Any,
*,
exclude_unset: bool,
exclude_defaults: bool = False,
exclude_none: bool = False,
) -> Any:
"""準備響應內容"""
if isinstance(res, BaseModel):
return res.model_dump(
by_alias=True,
exclude_unset=exclude_unset,
exclude_defaults=exclude_defaults,
exclude_none=exclude_none,
)
return res4.4 複雜類型驗證
from pydantic import BaseModel, validator, root_validator
from typing import List, Optional, Union
from datetime import datetime
from enum import Enum
class OrderStatus(str, Enum):
pending = "pending"
processing = "processing"
shipped = "shipped"
delivered = "delivered"
class Address(BaseModel):
street: str
city: str
country: str
postal_code: str
class OrderItem(BaseModel):
product_id: int
quantity: int = Field(..., gt=0)
unit_price: float = Field(..., gt=0)
@property
def total(self) -> float:
return self.quantity * self.unit_price
class Order(BaseModel):
id: Optional[int] = None
customer_id: int
items: List[OrderItem] = Field(..., min_items=1)
shipping_address: Address
billing_address: Optional[Address] = None
status: OrderStatus = OrderStatus.pending
created_at: datetime = Field(default_factory=datetime.utcnow)
@root_validator
def set_billing_address(cls, values):
if values.get('billing_address') is None:
values['billing_address'] = values.get('shipping_address')
return values
@property
def total_amount(self) -> float:
return sum(item.total for item in self.items)
@app.post("/orders/", response_model=Order)
async def create_order(order: Order):
# order 已經完全驗證
return order第五章：OpenAPI 自動生成
5.1 OpenAPI Schema 生成流程
graph TB
subgraph "OpenAPI 生成流程"
ROUTES[收集所有路由]
PARAMS[提取參數信息]
MODELS[收集 Pydantic 模型]
SCHEMA[生成 JSON Schema]
OPENAPI[組裝 OpenAPI 文檔]
JSON[/openapi.json]
SWAGGER[Swagger UI]
REDOC[ReDoc]
end
ROUTES --> PARAMS
PARAMS --> MODELS
MODELS --> SCHEMA
SCHEMA --> OPENAPI
OPENAPI --> JSON
JSON --> SWAGGER
JSON --> REDOC5.2 OpenAPI 生成源碼
# fastapi/openapi/utils.py
def get_openapi(
*,
title: str,
version: str,
openapi_version: str = "3.1.0",
summary: Optional[str] = None,
description: Optional[str] = None,
routes: Sequence[BaseRoute],
tags: Optional[list[dict[str, Any]]] = None,
servers: Optional[list[dict[str, Union[str, Any]]]] = None,
# ...
) -> dict[str, Any]:
"""生成 OpenAPI Schema"""
info: dict[str, Any] = {"title": title, "version": version}
if summary:
info["summary"] = summary
if description:
info["description"] = description
output: dict[str, Any] = {
"openapi": openapi_version,
"info": info,
}
if servers:
output["servers"] = servers
# 收集所有路徑操作
all_fields: dict[str, Any] = {}
paths: dict[str, dict[str, Any]] = {}
for route in routes:
if isinstance(route, routing.APIRoute):
# 獲取路徑操作
result = get_openapi_path(
route=route,
model_name_map=model_name_map,
field_mapping=field_mapping,
)
if result:
path, path_dict = result
if path in paths:
paths[path].update(path_dict)
else:
paths[path] = path_dict
if paths:
output["paths"] = paths
# 生成組件 Schema
if all_fields:
output["components"] = {"schemas": all_fields}
if tags:
output["tags"] = tags
return output
def get_openapi_path(
*,
route: routing.APIRoute,
model_name_map: dict[type[Any], str],
field_mapping: dict[tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue],
) -> Optional[tuple[str, dict[str, Any]]]:
"""為單個路由生成 OpenAPI 路徑項"""
path = route.path_format
operation: dict[str, Any] = {}
# 操作 ID
operation["operationId"] = route.unique_id
# 摘要和描述
if route.summary:
operation["summary"] = route.summary
if route.description:
operation["description"] = route.description
# 標籤
if route.tags:
operation["tags"] = route.tags
# 參數
parameters = get_openapi_params(route=route)
if parameters:
operation["parameters"] = parameters
# 請求體
if route.body_field:
request_body = get_openapi_request_body(route=route)
if request_body:
operation["requestBody"] = request_body
# 響應
operation["responses"] = get_openapi_responses(route=route)
# 安全
if route.security:
operation["security"] = route.security
return path, {route.methods[0].lower(): operation}5.3 Swagger UI 集成
# fastapi/openapi/docs.py
def get_swagger_ui_html(
*,
openapi_url: str,
title: str,
swagger_js_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js",
swagger_css_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css",
swagger_favicon_url: str = "https://fastapi.tiangolo.com/img/favicon.png",
oauth2_redirect_url: Optional[str] = None,
init_oauth: Optional[dict[str, Any]] = None,
swagger_ui_parameters: Optional[dict[str, Any]] = None,
) -> HTMLResponse:
"""生成 Swagger UI HTML"""
current_swagger_ui_parameters = swagger_ui_default_parameters.copy()
if swagger_ui_parameters:
current_swagger_ui_parameters.update(swagger_ui_parameters)
html = f"""
<!DOCTYPE html>
<html>
<head>
<link type="text/css" rel="stylesheet" href="{swagger_css_url}">
<link rel="shortcut icon" href="{swagger_favicon_url}">
<title>{title}</title>
</head>
<body>
<div id="swagger-ui"></div>
<script src="{swagger_js_url}"></script>
<script>
const ui = SwaggerUIBundle({{
url: "{openapi_url}",
dom_id: '#swagger-ui',
presets: [
SwaggerUIBundle.presets.apis,
SwaggerUIBundle.SwaggerUIStandalonePreset
],
layout: "StandaloneLayout",
{json.dumps(current_swagger_ui_parameters)[1:-1]}
}})
</script>
</body>
</html>
"""
return HTMLResponse(html)
# FastAPI 應用中自動註冊
# fastapi/applications.py
class FastAPI(Starlette):
def setup(self) -> None:
if self.openapi_url:
# 註冊 OpenAPI JSON 端點
@self.get(self.openapi_url, include_in_schema=False)
async def openapi():
return JSONResponse(self.openapi())
if self.docs_url:
# 註冊 Swagger UI
@self.get(self.docs_url, include_in_schema=False)
async def swagger_ui_html():
return get_swagger_ui_html(
openapi_url=self.openapi_url,
title=self.title + " - Swagger UI",
)第六章：中間件與異常處理
6.1 中間件架構
graph TB
subgraph "中間件棧"
REQ[Request]
MW1[ServerErrorMiddleware]
MW2[ExceptionMiddleware]
MW3[AuthMiddleware]
MW4[CORSMiddleware]
MW5[GZipMiddleware]
APP[Application]
RES[Response]
REQ --> MW1
MW1 --> MW2
MW2 --> MW3
MW3 --> MW4
MW4 --> MW5
MW5 --> APP
APP --> MW5
MW5 --> MW4
MW4 --> MW3
MW3 --> MW2
MW2 --> MW1
MW1 --> RES
end6.2 自定義中間件
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
import time
app = FastAPI()
# 方法 1: 使用裝飾器
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
# 方法 2: 繼承 BaseHTTPMiddleware
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# 請求前邏輯
print(f"Request: {request.method} {request.url}")
response = await call_next(request)
# 請求後邏輯
print(f"Response: {response.status_code}")
return response
app.add_middleware(LoggingMiddleware)
# 方法 3: 純 ASGI 中間件（更高效）
class PureASGIMiddleware:
def __init__(self, app: ASGIApp):
self.app = app
async def __call__(self, scope: Scope, receive: Receive, send: Send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
# 自定義邏輯
async def custom_send(message):
if message["type"] == "http.response.start":
# 添加自定義頭
headers = list(message.get("headers", []))
headers.append((b"x-custom-header", b"value"))
message["headers"] = headers
await send(message)
await self.app(scope, receive, custom_send)
app.add_middleware(PureASGIMiddleware)6.3 異常處理
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel
from typing import Any
app = FastAPI()
# 自定義異常
class CustomException(Exception):
def __init__(self, name: str, detail: str):
self.name = name
self.detail = detail
# 異常處理器
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
return JSONResponse(
status_code=418,
content={
"error": exc.name,
"detail": exc.detail,
"path": str(request.url)
},
)
# 覆蓋驗證錯誤處理
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
request: Request,
exc: RequestValidationError
):
return JSONResponse(
status_code=422,
content={
"detail": exc.errors(),
"body": exc.body,
},
)
# 全局異常處理
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"detail": "Internal server error"},
)
# FastAPI 內建異常處理
# fastapi/exception_handlers.py
async def request_validation_exception_handler(
request: Request,
exc: RequestValidationError,
) -> JSONResponse:
return JSONResponse(
status_code=HTTP_422_UNPROCESSABLE_ENTITY,
content={"detail": jsonable_encoder(exc.errors())},
)
async def http_exception_handler(
request: Request,
exc: HTTPException,
) -> Union[JSONResponse, Response]:
if exc.status_code in {204, 304}:
return Response(status_code=exc.status_code, headers=exc.headers)
return JSONResponse(
{"detail": exc.detail},
status_code=exc.status_code,
headers=exc.headers,
)第七章：效能優化
7.1 異步 vs 同步
from fastapi import FastAPI
import asyncio
import httpx
app = FastAPI()
# 異步端點 - 推薦用於 I/O 密集型操作
@app.get("/async")
async def async_endpoint():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data")
return response.json()
# 同步端點 - FastAPI 自動在線程池中運行
@app.get("/sync")
def sync_endpoint():
# CPU 密集型操作
import time
time.sleep(1) # 模擬阻塞操作
return {"message": "done"}
# 混合使用
@app.get("/mixed")
async def mixed_endpoint():
# 異步 I/O
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com")
# CPU 密集型任務放到線程池
from starlette.concurrency import run_in_threadpool
result = await run_in_threadpool(cpu_intensive_task, response.json())
return result7.2 響應優化
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse, UJSONResponse, StreamingResponse
import orjson
app = FastAPI()
# 使用更快的 JSON 序列化器
@app.get("/fast-json", response_class=ORJSONResponse)
async def fast_json():
return {"large": "data" * 1000}
# 全局設置默認響應類
app = FastAPI(default_response_class=ORJSONResponse)
# 流式響應
@app.get("/stream")
async def stream_data():
async def generate():
for i in range(100):
yield f"data: {i}\n\n"
await asyncio.sleep(0.1)
return StreamingResponse(
generate(),
media_type="text/event-stream"
)
# 背景任務
from fastapi import BackgroundTasks
@app.post("/send-notification")
async def send_notification(
email: str,
background_tasks: BackgroundTasks
):
background_tasks.add_task(send_email, email)
return {"message": "Notification sent in background"}
async def send_email(email: str):
# 耗時操作
await asyncio.sleep(5)
print(f"Email sent to {email}")7.3 緩存策略
from fastapi import FastAPI, Depends
from functools import lru_cache
import redis.asyncio as redis
app = FastAPI()
# 配置緩存
class Settings:
redis_url: str = "redis://localhost"
@lru_cache()
def get_settings():
return Settings()
# Redis 緩存
async def get_redis():
return await redis.from_url(get_settings().redis_url)
# 使用緩存裝飾器
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
@app.on_event("startup")
async def startup():
redis_client = await get_redis()
FastAPICache.init(RedisBackend(redis_client), prefix="fastapi-cache")
@app.get("/cached-data")
@cache(expire=60) # 緩存 60 秒
async def get_cached_data():
# 昂貴的操作
return {"data": "expensive to compute"}
# 手動緩存控制
@app.get("/manual-cache/{item_id}")
async def get_item(
item_id: int,
redis: redis.Redis = Depends(get_redis)
):
cache_key = f"item:{item_id}"
# 嘗試從緩存獲取
cached = await redis.get(cache_key)
if cached:
return {"data": cached, "source": "cache"}
# 從數據庫獲取
data = await fetch_from_db(item_id)
# 存入緩存
await redis.setex(cache_key, 300, data)
return {"data": data, "source": "database"}第八章：最佳實踐
8.1 專案結構
project/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 應用入口
│ ├── config.py # 配置管理
│ ├── dependencies.py # 共享依賴
│ │
│ ├── api/
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── endpoints/
│ │ │ │ ├── users.py
│ │ │ │ ├── items.py
│ │ │ │ └── auth.py
│ │ │ └── router.py
│ │ └── deps.py # API 依賴
│ │
│ ├── core/
│ │ ├── __init__.py
│ │ ├── security.py # 安全相關
│ │ └── exceptions.py # 自定義異常
│ │
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py # SQLAlchemy 模型
│ │ └── item.py
│ │
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── user.py # Pydantic 模型
│ │ └── item.py
│ │
│ ├── crud/
│ │ ├── __init__.py
│ │ ├── base.py # CRUD 基類
│ │ └── user.py
│ │
│ └── db/
│ ├── __init__.py
│ ├── session.py # 數據庫會話
│ └── base.py # 模型基類
│
├── tests/
├── alembic/ # 數據庫遷移
├── pyproject.toml
└── Dockerfile8.2 依賴注入最佳實踐
# app/dependencies.py
from typing import Generator, Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from jose import JWTError, jwt
from app.db.session import SessionLocal
from app.models.user import User
from app.core.config import settings
# 數據庫依賴
def get_db() -> Generator:
db = SessionLocal()
try:
yield db
finally:
db.close()
# OAuth2 依賴
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# 當前用戶依賴
async def get_current_user(
db: Annotated[Session, Depends(get_db)],
token: Annotated[str, Depends(oauth2_scheme)]
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
)
user_id: str = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = db.query(User).filter(User.id == user_id).first()
if user is None:
raise credentials_exception
return user
# 活躍用戶依賴
async def get_current_active_user(
current_user: Annotated[User, Depends(get_current_user)]
) -> User:
if not current_user.is_active:
raise HTTPException(
status_code=400,
detail="Inactive user"
)
return current_user
# 管理員依賴
async def get_current_admin(
current_user: Annotated[User, Depends(get_current_active_user)]
) -> User:
if not current_user.is_admin:
raise HTTPException(
status_code=403,
detail="Not enough permissions"
)
return current_user
# 使用類型別名簡化
CurrentUser = Annotated[User, Depends(get_current_active_user)]
AdminUser = Annotated[User, Depends(get_current_admin)]
DBSession = Annotated[Session, Depends(get_db)]
# 在端點中使用
@router.get("/users/me")
async def read_users_me(current_user: CurrentUser):
return current_user
@router.delete("/users/{user_id}")
async def delete_user(
user_id: int,
admin: AdminUser,
db: DBSession
):
# 只有管理員可以刪除用戶
...總結
FastAPI 的核心優勢在於其精妙的設計：
graph TB
subgraph "FastAPI 核心特性"
STARLETTE[Starlette<br/>高效能 ASGI]
PYDANTIC[Pydantic<br/>數據驗證]
TYPEHINTS[Type Hints<br/>自動文檔]
DI[依賴注入<br/>可測試性]
OPENAPI[OpenAPI<br/>標準化]
end
STARLETTE --> FAST[高效能]
PYDANTIC --> SAFE[類型安全]
TYPEHINTS --> DOCS[自動文檔]
DI --> TEST[易於測試]
OPENAPI --> STANDARD[標準化]關鍵要點：
- ASGI 優勢：支援異步、WebSocket、流式響應
- 類型系統：編輯器支援、自動驗證、自動文檔
- 依賴注入：解耦、可測試、可重用
- Pydantic 整合：強大的數據驗證和序列化
- OpenAPI 自動生成：免維護的 API 文檔
FastAPI 證明了 Python 也可以構建高效能的現代 Web 應用。