http_log.py
3.22 KB
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
import json
import time
import logging
logger = logging.getLogger(__name__)
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import StreamingResponse
from config import settings
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# 获取请求信息
method = request.method
url = str(request.url)
client_host = request.client.host if request.client else "unknown"
# 判断是否需要记录日志
should_log = request.url.path.startswith(settings.url_prefix)
if should_log:
# 记录请求开始时间
start_time = time.time()
# 读取请求体
request_body = None
if method in ("POST", "PUT", "PATCH"):
try:
body = await request.body()
if body:
request_body = body.decode("utf-8", errors="replace")
# 尝试格式化为JSON
try:
request_body = json.loads(request_body)
except json.JSONDecodeError:
pass # 保持原始字符串
except Exception:
pass
# 处理请求
response = await call_next(request)
# 计算处理时间
process_time = time.time() - start_time
# 读取响应体（仅处理非流式响应）
response_body = None
if not isinstance(response, StreamingResponse):
try:
response_body_bytes = b""
async for chunk in response.body_iterator:
response_body_bytes += chunk
# 重新构建响应
response_body = response_body_bytes.decode("utf-8", errors="replace")
# 尝试格式化为JSON
try:
response_body = json.loads(response_body)
except json.JSONDecodeError:
pass # 保持原始字符串
# 重建响应对象
response = StreamingResponse(
iter([response_body_bytes]),
status_code=response.status_code,
headers=dict(response.headers),
media_type=response.media_type,
)
except Exception:
pass
# 构建单条日志记录
status_code = response.status_code
log_data = {
"method": method,
"url": url,
"client": client_host,
"status": status_code,
"time": f"{process_time:.3f}s"
}
if request_body:
log_data["request_body"] = request_body
if response_body:
log_data["response_body"] = response_body
logger.info(json.dumps(log_data, ensure_ascii=False))
return response
else:
# 不需要记录日志的请求，直接处理
return await call_next(request)