学习笔记说明
这是一篇Python的学习笔记，关于Python 中间件，仅做记录
核心概念
- 横切关注点（Cross-Cutting Concerns）：描述了跨越多个功能、可被横切的共同逻辑的现象
- AOP（Aspect Oriented Programming）：应该将那些跨域多个功能、可被横切的共同逻辑分离出去，不侵入功能本身
- DI：AOP的一种实现手段
理解中间件
FastAPI的中间件是AOP的一种实现手段
async def my_middleware(request: Request, call_next):
# 操作...
response = await call_next(request) # 传递给下一个中间件
# 操作...
return xxx
app.middleware("http")(my_middleware)
1
2
3
4
5
6
7
2
3
4
5
6
7
统一响应格式中间件
apps/web-service/app/core/middleware/response.py
import json
from fastapi import Request
from fastapi.responses import JSONResponse
async def unified_response(request: Request, call_next):
response = await call_next(request)
if not request.url.path.startswith("/api/"):
return response
body = b""
async for chunk in response.body_iterator:
body += chunk
headers = dict(response.headers)
headers.pop("content-length", None)
if response.status_code >= 400:
err = json.loads(body) if body else {}
return JSONResponse(
content={
"code": err.get("code", str(response.status_code)),
"data": None,
"message": err.get("message", ""),
},
status_code=response.status_code,
headers=headers,
)
data = json.loads(body) if body else None
return JSONResponse(
content={"code": "0", "data": data, "message": "success"},
status_code=response.status_code,
headers=headers,
)
MIDDLEWARE = (unified_response, {})
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
apps/web-service/app/core/middleware/__init__.py
import inspect
from fastapi import FastAPI
from app.core.middleware import response
MIDDLEWARES = [
response.MIDDLEWARE
]
def register_middleware(app: FastAPI) -> None:
for callable_obj, kwargs in MIDDLEWARES:
if inspect.isclass(callable_obj):
app.add_middleware(callable_obj, **kwargs)
else:
app.middleware("http")(callable_obj)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
apps/web-service/app/main.py
from app.core.middleware import register_middleware
register_middleware(app)
1
2
3
2
3
apps/web-service/app/core/openapi.py
from fastapi import FastAPI
def _build_envelope_properties(data_schema: dict) -> dict:
return {
"code": {
"type": "string",
"description": "状态码，0 表示成功",
"example": "0",
},
"data": data_schema,
"message": {
"type": "string",
"description": "提示信息",
"example": "success",
},
}
def _make_envelope(data_schema: dict) -> dict:
return {
"type": "object",
"properties": _build_envelope_properties(data_schema),
"required": ["code", "data", "message"],
}
def _wrap_ref_schema(openapi_schema: dict, schema: dict) -> dict:
ref_path: str = schema["$ref"]
ref_name = ref_path.split("/")[-1]
title = ref_name.replace("_", " ")
wrapper_name = f"ApiResponse_{ref_name}"
schemas = openapi_schema.setdefault("components", {}).setdefault("schemas", {})
if wrapper_name not in schemas:
schemas[wrapper_name] = {
"title": f"ApiResponse[{title}]",
"type": "object",
"properties": _build_envelope_properties(schema),
"required": ["code", "data", "message"],
}
return {"$ref": f"#/components/schemas/{wrapper_name}"}
def _wrap_array_schema(openapi_schema: dict, schema: dict) -> dict:
items = schema.get("items")
if isinstance(items, dict) and "$ref" in items:
item_name = items["$ref"].split("/")[-1]
title = f"List[{item_name.replace('_', ' ')}]"
wrapper_name = f"ApiResponse_List_{item_name}"
schemas = openapi_schema.setdefault("components", {}).setdefault("schemas", {})
if wrapper_name not in schemas:
schemas[wrapper_name] = {
"title": f"ApiResponse[{title}]",
"type": "object",
"properties": _build_envelope_properties(schema),
"required": ["code", "data", "message"],
}
return {"$ref": f"#/components/schemas/{wrapper_name}"}
return _make_envelope(schema)
def setup_openapi(app: FastAPI) -> None:
_original = app.openapi
def _custom():
if app.openapi_schema:
return app.openapi_schema
schema = _original()
for path, path_item in schema.get("paths", {}).items():
if not path.startswith("/api/"):
continue
for method in ("get", "post", "put", "delete", "patch"):
operation = path_item.get(method)
if operation is None:
continue
responses = operation.get("responses", {})
for status_code_str, response in list(responses.items()):
status_code = int(status_code_str)
if status_code not in (200, 201):
del responses[status_code_str]
continue
content = response.get("content", {})
json_content = content.get("application/json")
if json_content is None:
continue
original_schema = json_content.get("schema")
if original_schema is None:
continue
if "$ref" in original_schema:
wrapped = _wrap_ref_schema(schema, original_schema)
elif original_schema.get("type") == "array":
wrapped = _wrap_array_schema(schema, original_schema)
else:
wrapped = _make_envelope(original_schema)
json_content["schema"] = wrapped
app.openapi_schema = schema
return schema
app.openapi = _custom
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
apps/web-service/app/main.py
from app.core.openapi import setup_openapi
setup_openapi(app)
1
2
3
2
3
用于性能调试的中间件
apps/web-service/app/core/middleware/process_time.py
import time
from fastapi import Request
async def process_time(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
response.headers["X-Process-Time"] = str(round(time.perf_counter() - start, 4))
return response
MIDDLEWARE = (process_time, {})
1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
用于解决跨域的中间件
.env.example / .env
# 其他配置
WEB_CORS_ORIGINS=* # 跨域白名单，多个来源用逗号分隔，* 表示允许所有
WEB_CORS_EXPOSE_HEADERS=X-Process-Time # 允许前端读取的响应头，多个用逗号分隔
# 其他配置
1
2
3
4
5
2
3
4
5
apps/web-service/app/core/config.py
class _WebSettings(_BaseSettingsWithEnv):
app_name: str = "Web Service API" # 实际读取 WEB_APP_NAME
cors_origins: str = "" # 实际读取 WEB_CORS_ORIGINS，多个来源用逗号分隔
cors_expose_headers: str = "" # 实际读取 WEB_CORS_EXPOSE_HEADERS
# 配置读取方式
model_config = {"env_prefix": "WEB_"}
1
2
3
4
5
6
7
2
3
4
5
6
7
apps/web-service/app/core/middleware/cors.py
from starlette.middleware.cors import CORSMiddleware
from app.core.config import web_settings
origins = [o.strip() for o in web_settings.cors_origins.split(",") if o.strip()]
expose_headers = [
h.strip() for h in web_settings.cors_expose_headers.split(",") if h.strip()
]
MIDDLEWARE = (
CORSMiddleware,
{
"allow_origins": origins,
"allow_credentials": True,
"allow_methods": ["*"],
"allow_headers": ["*"],
"expose_headers": expose_headers,
},
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
