Steven
2b493bd9f8
CI / backend (push) Canceled after 0s
跑通迁移与测试后修掉的真问题： - alembic 0001：asyncpg 走预备语句，一次 execute 只能一条 SQL。 CREATE FUNCTION / CREATE POLICY / GRANT 全部拆成单语句逐条执行， 否则报 "cannot insert multiple commands into a prepared statement"。 - 创建工作区：ORM flush 只按 relationship 排序，裸外键不算数， member 可能先于 workspace 落地撞 FK。改为先 flush workspace 再插 member。 - datetime.UTC 是 3.11+ 才有的，本机 3.10 直接 ImportError。改用 timezone.utc。 - 幂等中间件：原来会缓存 4xx。校验失败被缓存后，用户改对参数仍拿到同一个 422。 改为只缓存 <400。 - 发码节流窗口改为可配置 SMS_THROTTLE_SECONDS（测试置 0）， 节流用例自己显式打开，避免"整库关掉节流"导致节流从未被验证。 - 租户隔离用例：pytest.raises(Exception) 改为精确的 ProgrammingError， 否则任何异常都会让"应用角色不能建表"这条断言假通过。 验收：pytest 33 passed；ruff check / format 通过；真机跑 uvicorn， /health、/version、登录、建工作区、列表、/auth/me 全通。
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""结构化日志：JSON 行 + request_id 贯穿 + 敏感字段脱敏。
|
|
|
|
脱敏是**无条件**的：任何日志调用里的 dict 都会过一遍 _mask。
|
|
不要为了方便调试关掉它——AGENTS.md 第 7 条。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import sys
|
|
from contextvars import ContextVar
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")
|
|
user_id_ctx: ContextVar[str] = ContextVar("user_id", default="-")
|
|
workspace_id_ctx: ContextVar[str] = ContextVar("workspace_id", default="-")
|
|
|
|
_SENSITIVE_HINTS = (
|
|
"password",
|
|
"passwd",
|
|
"token",
|
|
"secret",
|
|
"api_key",
|
|
"apikey",
|
|
"authorization",
|
|
"cookie",
|
|
"credential",
|
|
"private_key",
|
|
)
|
|
MASK = "***"
|
|
|
|
|
|
def _mask(value: Any, depth: int = 0) -> Any:
|
|
if depth > 6:
|
|
return "<deep>"
|
|
if isinstance(value, dict):
|
|
return {
|
|
key: (
|
|
MASK
|
|
if any(hint in str(key).lower() for hint in _SENSITIVE_HINTS)
|
|
else _mask(val, depth + 1)
|
|
)
|
|
for key, val in value.items()
|
|
}
|
|
if isinstance(value, list | tuple):
|
|
return [_mask(item, depth + 1) for item in value]
|
|
return value
|
|
|
|
|
|
class JsonFormatter(logging.Formatter):
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
payload: dict[str, Any] = {
|
|
"ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
|
|
"level": record.levelname,
|
|
"logger": record.name,
|
|
"msg": record.getMessage(),
|
|
"request_id": request_id_ctx.get(),
|
|
"user_id": user_id_ctx.get(),
|
|
"ws_id": workspace_id_ctx.get(),
|
|
}
|
|
extra = getattr(record, "extra_fields", None)
|
|
if isinstance(extra, dict):
|
|
payload.update(_mask(extra))
|
|
if record.exc_info:
|
|
payload["exc"] = self.formatException(record.exc_info)
|
|
return json.dumps(payload, ensure_ascii=False, default=str)
|
|
|
|
|
|
def setup_logging(level: str = "INFO") -> None:
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
handler.setFormatter(JsonFormatter())
|
|
root = logging.getLogger()
|
|
root.handlers = [handler]
|
|
root.setLevel(level)
|
|
# uvicorn 的 access log 交给自己的中间件，避免重复
|
|
for noisy in ("uvicorn.access", "sqlalchemy.engine.Engine"):
|
|
logging.getLogger(noisy).handlers = []
|
|
logging.getLogger(noisy).propagate = False
|
|
|
|
|
|
def get_logger(name: str) -> logging.Logger:
|
|
return logging.getLogger(name)
|