feat: 新增全链路追踪、优雅关停支持与统一异常处理，优化渠道查询接口
1. 新增TraceId中间件实现全链路请求追踪 2. 新增在途请求计数器与中间件支持优雅关停 3. 重构全局异常处理器，统一三模块异常响应格式 4. 优化渠道查询接口，新增多维度过滤与模糊匹配 5. 重构channels模块数据库会话管理，避免事务冲突 6. 新增目录搜索建议与目录导出功能
This commit is contained in:
parent
e6e24a91de
commit
12ec5af0de
@ -16,17 +16,26 @@ from collections import defaultdict, deque
|
||||
|
||||
import uvicorn
|
||||
from fastapi import FastAPI, Request, status
|
||||
from fastapi.exceptions import RequestValidationError
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
|
||||
from yuxi.external_systems.exceptions import ExternalSystemError
|
||||
from yuxi.scheduler.exceptions import SchedulerError
|
||||
from yuxi.channels.contract.errors import Error as ChannelError
|
||||
from server.routers import router
|
||||
from server.utils.error_handlers import (
|
||||
unified_error_handler,
|
||||
unhandled_exception_handler,
|
||||
request_validation_error_handler,
|
||||
)
|
||||
from server.utils.lifespan import lifespan
|
||||
from server.utils.auth_middleware import is_public_path
|
||||
from server.utils.common_utils import setup_logging
|
||||
from server.utils.access_log_middleware import AccessLogMiddleware
|
||||
from server.utils.trace_middleware import TraceIdMiddleware
|
||||
from server.utils.inflight_tracker import InflightTrackingMiddleware
|
||||
|
||||
# 设置日志配置
|
||||
setup_logging()
|
||||
@ -44,47 +53,12 @@ app = FastAPI(lifespan=lifespan)
|
||||
app.include_router(router, prefix="/api")
|
||||
|
||||
|
||||
@app.exception_handler(ExternalSystemError)
|
||||
async def external_system_error_handler(_request: Request, exc: ExternalSystemError) -> JSONResponse:
|
||||
"""外部系统限界上下文异常统一映射为 HTTP 响应。
|
||||
|
||||
- ``status_code`` 来自异常类。
|
||||
- ``details`` 透传到响应体，供前端结构化处理。
|
||||
- 429/503 且 ``details`` 含 ``retry_after`` 时，自动设置 ``Retry-After`` 响应头。
|
||||
"""
|
||||
headers: dict[str, str] = {}
|
||||
if exc.status_code in (429, 503):
|
||||
retry_after = exc.details.get("retry_after")
|
||||
if retry_after is not None:
|
||||
headers["Retry-After"] = str(retry_after)
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content={"detail": exc.message, "details": exc.details},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(SchedulerError)
|
||||
async def scheduler_error_handler(_request: Request, exc: SchedulerError) -> JSONResponse:
|
||||
"""定时任务调度限界上下文异常统一映射为 HTTP 响应。
|
||||
|
||||
- ``status_code`` 来自异常类（400 / 404 / 409 / 413 / 500）。
|
||||
- ``details`` 透传到响应体，供前端结构化处理。
|
||||
"""
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content={"detail": exc.message, "details": exc.details},
|
||||
)
|
||||
|
||||
|
||||
# CORS 设置
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
# 三模块异常统一注册到同一个 handler
|
||||
app.add_exception_handler(ChannelError, unified_error_handler)
|
||||
app.add_exception_handler(ExternalSystemError, unified_error_handler)
|
||||
app.add_exception_handler(SchedulerError, unified_error_handler)
|
||||
app.add_exception_handler(RequestValidationError, request_validation_error_handler)
|
||||
app.add_exception_handler(Exception, unhandled_exception_handler)
|
||||
|
||||
|
||||
def _extract_client_ip(request: Request) -> str:
|
||||
@ -115,7 +89,15 @@ class LoginRateLimitMiddleware(BaseHTTPMiddleware):
|
||||
retry_after = int(max(1, RATE_LIMIT_WINDOW_SECONDS - (now - attempt_history[0])))
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
||||
content={"detail": "登录尝试过于频繁，请稍后再试"},
|
||||
content={
|
||||
"success": False,
|
||||
"error": {
|
||||
"code": "RATE_LIMIT",
|
||||
"message": "登录尝试过于频繁，请稍后再试",
|
||||
"trace_id": None,
|
||||
"details": {"retry_after": retry_after},
|
||||
},
|
||||
},
|
||||
headers={"Retry-After": str(retry_after)},
|
||||
)
|
||||
|
||||
@ -165,12 +147,20 @@ class AuthMiddleware(BaseHTTPMiddleware):
|
||||
return await call_next(request)
|
||||
|
||||
|
||||
# 添加访问日志中间件（记录请求处理时间）
|
||||
app.add_middleware(AccessLogMiddleware)
|
||||
|
||||
# 添加鉴权中间件
|
||||
# 中间件注册顺序为 LIFO（后注册先执行）：
|
||||
# CORSMiddleware 最内层，InflightTrackingMiddleware 最外层最先执行
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
app.add_middleware(LoginRateLimitMiddleware)
|
||||
app.add_middleware(AuthMiddleware)
|
||||
app.add_middleware(AccessLogMiddleware)
|
||||
app.add_middleware(TraceIdMiddleware)
|
||||
app.add_middleware(InflightTrackingMiddleware)
|
||||
|
||||
if __name__ == "__main__":
|
||||
# uvicorn.run(app, host="0.0.0.0", port=5050, threads=10, workers=10, reload=True)
|
||||
|
||||
@ -16,7 +16,6 @@ from enum import Enum
|
||||
from typing import Any, Literal
|
||||
|
||||
from fastapi import APIRouter, Depends, Request
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from yuxi.channels.contract.dtos.common import Operator, OperatorRole
|
||||
from yuxi.channels.contract.dtos.control import ControlResult
|
||||
from yuxi.channels.contract.errors import (
|
||||
@ -51,13 +50,12 @@ from yuxi.channels.infrastructure.channel_use_cases import (
|
||||
ChannelUseCases,
|
||||
create_channel_use_cases,
|
||||
)
|
||||
from yuxi.storage.postgres.manager import pg_manager
|
||||
from yuxi.storage.postgres.models_business import User
|
||||
from yuxi.utils.crypto import SENSITIVE_HTTP_HEADERS
|
||||
from yuxi.utils.datetime_utils import coerce_any_to_utc_datetime
|
||||
from yuxi.utils.trace_context import get_trace_id
|
||||
|
||||
from server.utils.auth_middleware import get_db
|
||||
|
||||
# 时间粒度查询参数类型（analytics / dashboard / pairing / content_review / outbox 共享）
|
||||
Granularity = Literal["hour", "day", "week"]
|
||||
|
||||
@ -86,23 +84,29 @@ async def get_channel_di_container(request: Request) -> DependencyInjectionConta
|
||||
|
||||
|
||||
async def get_channel_use_cases(
|
||||
db: AsyncSession = Depends(get_db),
|
||||
di_container: DependencyInjectionContainer = Depends(get_channel_di_container),
|
||||
) -> ChannelUseCases:
|
||||
"""装配 channels 控制面 UseCase 集合。
|
||||
"""装配 channels 控制面 UseCase 集合（请求级独立 session）。
|
||||
|
||||
将请求级 DB 会话与 DI 容器传入 ``create_channel_use_cases``，构造本次
|
||||
请求可用的 ``ChannelUseCases``。UseCase 内部端口方法在控制面管道事务中
|
||||
执行。
|
||||
channels 模块 **不复用** 鉴权依赖（``get_db``）持有的 ``AsyncSession``，
|
||||
而是通过 ``pg_manager.get_async_session_context()`` 开启独立的请求级
|
||||
会话。原因：鉴权阶段执行的 ``SELECT`` 会触发 SQLAlchemy 2.0 autobegin
|
||||
语义开启隐式事务，若 channels 复用同一 session，控制面
|
||||
``TransactionPort.begin()`` 会在已存在的事务上再次 ``begin()``，
|
||||
抛出 ``InvalidRequestError: A transaction is already begun``。
|
||||
|
||||
独立 session 使事务边界由 channels 应用层独占控制，与鉴权会话隔离，
|
||||
避免事务状态互相污染。会话生命周期限制在本次请求内，请求结束自动
|
||||
commit（无异常）或 rollback（有异常）后关闭。
|
||||
|
||||
Args:
|
||||
db: 请求级异步 DB 会话。
|
||||
di_container: channels 模块 DI 容器。
|
||||
|
||||
Returns:
|
||||
Yields:
|
||||
已装配的 ``ChannelUseCases``，含 ``session_management`` 等端口字段。
|
||||
"""
|
||||
return create_channel_use_cases(db, di_container)
|
||||
async with pg_manager.get_async_session_context() as db:
|
||||
yield create_channel_use_cases(db, di_container)
|
||||
|
||||
|
||||
def build_operator(user: User, request: Request) -> Operator:
|
||||
|
||||
@ -1,38 +1,51 @@
|
||||
"""目录查询域 Router（FR-14，DIR-QUERY-01~04 + DIR-01 + DIR-CACHE-CLEAR）。
|
||||
"""目录查询域 Router（FR-14，DIR-QUERY-01~04 + DIR-01 + DIR-CACHE-CLEAR + Task 5）。
|
||||
|
||||
本 router 实现渠道通讯录目录查询的全部 HTTP 端点，覆盖用户搜索 / 群组搜索 /
|
||||
用户资料 / 群组详情 / 群组成员 / 缓存清理共 6 个操作。所有端点统一采用模板 A
|
||||
（控制面端口路由），通过 ``get_channel_use_cases`` 装配 ``ChannelUseCases``，
|
||||
经 ``directory`` 端口调用类型化方法，由 ``ChannelControlService._executeControl``
|
||||
内部走控制面管道（auth → permission → rate_limit → dispatch → audit）。
|
||||
用户资料 / 群组详情 / 群组成员 / 缓存清理 / 目录导出共 7 个操作。除导出端点
|
||||
返回 ``StreamingResponse`` 文件流外，其余端点统一采用模板 A（控制面端口路由），
|
||||
通过 ``get_channel_use_cases`` 装配 ``ChannelUseCases``，经 ``directory`` 端口
|
||||
调用类型化方法，由 ``ChannelControlService._executeControl`` 内部走控制面管道
|
||||
（auth → permission → rate_limit → dispatch → audit）。导出端点同样经控制面
|
||||
管道执行，handler 返回内容字符串，router 包装为文件流以保持审计链路。
|
||||
|
||||
鉴权策略：全部端点使用 ``get_admin_user`` 依赖，要求管理员或超级管理员角色。
|
||||
角色校验由依赖函数完成，router 内不做角色判断（规范 §4）。
|
||||
|
||||
路径设计：静态后缀路径（``/search``）先于动态单段路径（``/{peer_id}`` /
|
||||
``/{group_id}``）声明，动态单段先于多段子资源路径（``/{group_id}/members``）
|
||||
声明（规范 §6.5），避免静态路径被动态参数捕获。子 router 不自行设置 prefix，
|
||||
根前缀 ``/channels`` 由 ``channels_router`` 聚合 router 统一追加。
|
||||
路径设计：静态后缀路径（``/search`` / ``/export``）先于动态单段路径
|
||||
（``/{peer_id}`` / ``/{group_id}``）声明，动态单段先于多段子资源路径
|
||||
（``/{group_id}/members``）声明（规范 §6.5），避免静态路径被动态参数捕获。
|
||||
子 router 不自行设置 prefix，根前缀 ``/channels`` 由 ``channels_router`` 聚合
|
||||
router 统一追加。
|
||||
|
||||
端点清单（对应《05-目录查询域设计方案 v1.1》§2.3）：
|
||||
- GET /{channel_type}/{account_id}/directory/users/search DIR-QUERY-01 search_directory_users
|
||||
- GET /{channel_type}/{account_id}/directory/groups/search DIR-QUERY-02 search_directory_groups
|
||||
- GET /{channel_type}/{account_id}/directory/users/{peer_id} DIR-QUERY-03 get_directory_user_profile
|
||||
- GET /{channel_type}/{account_id}/directory/groups/{group_id} DIR-01 get_directory_group_detail
|
||||
- GET /{channel_type}/{account_id}/directory/groups/{group_id}/members DIR-QUERY-04 get_directory_group_members
|
||||
- DELETE /{channel_type}/{account_id}/directory/cache DIR-CACHE-CLEAR clear_directory_cache
|
||||
端点清单（对应《05-目录查询域设计方案 v1.1》§2.3 + Task 5）：
|
||||
- GET /{channel_type}/{account_id}/directory/users/search
|
||||
DIR-QUERY-01 search_directory_users
|
||||
- GET /{channel_type}/{account_id}/directory/groups/search
|
||||
DIR-QUERY-02 search_directory_groups
|
||||
- GET /{channel_type}/{account_id}/directory/users/{peer_id}
|
||||
DIR-QUERY-03 get_directory_user_profile
|
||||
- GET /{channel_type}/{account_id}/directory/groups/{group_id}
|
||||
DIR-01 get_directory_group_detail
|
||||
- GET /{channel_type}/{account_id}/directory/groups/{group_id}/members
|
||||
DIR-QUERY-04 get_directory_group_members
|
||||
- DELETE /{channel_type}/{account_id}/directory/cache
|
||||
DIR-CACHE-CLEAR clear_directory_cache
|
||||
- POST /{channel_type}/{account_id}/directory/export
|
||||
TASK-5 export_directory
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from typing import Any, Literal
|
||||
|
||||
from fastapi import APIRouter, Body, Depends, Query, Request
|
||||
from fastapi.responses import StreamingResponse
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
from yuxi.channels.contract.dtos.channel import ChannelType
|
||||
from yuxi.channels.contract.dtos.directory import (
|
||||
BatchDirectoryProfilesQuery,
|
||||
ClearDirectoryCacheCmd,
|
||||
DirectoryEntryType,
|
||||
DirectoryQuery,
|
||||
)
|
||||
from yuxi.storage.postgres.models_business import User
|
||||
@ -83,6 +96,38 @@ async def search_directory_users(
|
||||
return {"success": True, "data": serialize_control_data(result.data)}
|
||||
|
||||
|
||||
@directory_router.get(
|
||||
"/{channel_type}/{account_id}/directory/suggestions",
|
||||
response_model=dict,
|
||||
)
|
||||
async def search_directory_suggestions(
|
||||
channel_type: ChannelType,
|
||||
account_id: str,
|
||||
request: Request,
|
||||
keyword: str = Query(..., min_length=1, description="搜索关键词"),
|
||||
limit: int = Query(default=10, ge=1, le=20, description="建议数量"),
|
||||
use_cases=Depends(get_channel_use_cases),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
) -> dict[str, Any]:
|
||||
"""搜索渠道目录建议（FR-14）。
|
||||
|
||||
根据关键词搜索渠道侧目录建议，用于输入联想。5min TTL 缓存。对应控制
|
||||
面操作 ``directory/search_suggestions``（由
|
||||
``ChannelControlService.searchDirectorySuggestions`` 内部构造
|
||||
``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。
|
||||
"""
|
||||
operator = build_operator(current_user, request)
|
||||
query = DirectoryQuery(keyword=keyword, limit=limit)
|
||||
result = await use_cases.directory.searchDirectorySuggestions(
|
||||
channel_type=channel_type,
|
||||
account_id=account_id,
|
||||
query=query,
|
||||
operator=operator,
|
||||
)
|
||||
raiseOnControlFailure(result)
|
||||
return {"success": True, "data": serialize_control_data(result.data)}
|
||||
|
||||
|
||||
@directory_router.get(
|
||||
"/{channel_type}/{account_id}/directory/groups/search",
|
||||
response_model=dict,
|
||||
@ -157,6 +202,35 @@ class BatchProfilesRequest(BaseModel):
|
||||
min_length=1,
|
||||
description="待查询的对端 ID 列表（可能是用户 ID 或群组 ID）",
|
||||
)
|
||||
types: list[Literal["user", "group"]] | None = Field(
|
||||
default=None,
|
||||
description="按条目类型过滤（user / group），不传入则查询全部类型",
|
||||
)
|
||||
|
||||
|
||||
class ExportDirectoryRequest(BaseModel):
|
||||
"""目录导出请求（Task 5）。"""
|
||||
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
scope: Literal["users", "groups"] = Field(
|
||||
default="users",
|
||||
description="导出范围（users 或 groups）",
|
||||
)
|
||||
keyword: str | None = Field(
|
||||
default=None,
|
||||
description="搜索关键词",
|
||||
)
|
||||
format: Literal["csv", "json"] = Field(
|
||||
default="csv",
|
||||
description="导出格式（csv 或 json）",
|
||||
)
|
||||
limit: int = Field(
|
||||
default=1000,
|
||||
ge=1,
|
||||
le=1000,
|
||||
description="导出数量限制（1~1000）",
|
||||
)
|
||||
|
||||
|
||||
@directory_router.post(
|
||||
@ -182,7 +256,11 @@ async def batch_directory_profiles(
|
||||
``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。
|
||||
"""
|
||||
operator = build_operator(current_user, request)
|
||||
query = BatchDirectoryProfilesQuery(peer_ids=tuple(payload.peer_ids))
|
||||
types = tuple(DirectoryEntryType(t) for t in payload.types) if payload.types is not None else None
|
||||
query = BatchDirectoryProfilesQuery(
|
||||
peer_ids=tuple(payload.peer_ids),
|
||||
types=types,
|
||||
)
|
||||
result = await use_cases.directory.batchGetDirectoryProfiles(
|
||||
channel_type=channel_type,
|
||||
account_id=account_id,
|
||||
@ -193,6 +271,53 @@ async def batch_directory_profiles(
|
||||
return {"success": True, "data": serialize_control_data(result.data)}
|
||||
|
||||
|
||||
@directory_router.post(
|
||||
"/{channel_type}/{account_id}/directory/export",
|
||||
)
|
||||
async def export_directory(
|
||||
channel_type: ChannelType,
|
||||
account_id: str,
|
||||
request: Request,
|
||||
payload: ExportDirectoryRequest = Body(...),
|
||||
use_cases=Depends(get_channel_use_cases),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
) -> StreamingResponse:
|
||||
"""导出目录数据（FR-14，Task 5）。
|
||||
|
||||
搜索渠道侧用户或群组目录，将结果格式化为 CSV 或 JSON 文件流返回。
|
||||
经控制面管道执行 ``directory/export`` 操作，保持审计链路；handler 返回
|
||||
``{"content": str, "format": str, "filename": str}``，router 包装为
|
||||
``StreamingResponse`` 触发浏览器下载。
|
||||
"""
|
||||
operator = build_operator(current_user, request)
|
||||
result = await use_cases.directory.exportDirectory(
|
||||
channel_type=channel_type,
|
||||
account_id=account_id,
|
||||
scope=payload.scope,
|
||||
keyword=payload.keyword,
|
||||
format=payload.format,
|
||||
limit=payload.limit,
|
||||
operator=operator,
|
||||
)
|
||||
raiseOnControlFailure(result)
|
||||
data = serialize_control_data(result.data)
|
||||
content: str = data["content"]
|
||||
fmt: str = data["format"]
|
||||
filename: str = data["filename"]
|
||||
media_type = "text/csv; charset=utf-8" if fmt == "csv" else "application/json; charset=utf-8"
|
||||
|
||||
async def _stream():
|
||||
yield content.encode("utf-8")
|
||||
|
||||
return StreamingResponse(
|
||||
_stream(),
|
||||
media_type=media_type,
|
||||
headers={
|
||||
"Content-Disposition": f"attachment; filename={filename}",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@directory_router.get(
|
||||
"/{channel_type}/{account_id}/directory/groups/{group_id}",
|
||||
response_model=dict,
|
||||
|
||||
@ -145,10 +145,19 @@ async def list_admin_messages(
|
||||
channel_type: ChannelType | None = Query(default=None, description="按渠道过滤"),
|
||||
start_time: str | None = Query(default=None, description="起始时间 ISO 8601"),
|
||||
end_time: str | None = Query(default=None, description="截止时间 ISO 8601"),
|
||||
channel_status: Literal["sent", "delivered", "read", "recalled", "edited"] | None = Query(
|
||||
channel_status: str | None = Query(
|
||||
default=None,
|
||||
description="按渠道侧状态过滤（sent/delivered/read/recalled/edited）",
|
||||
description="按渠道侧状态过滤，支持逗号分隔多值（如 sent,delivered）",
|
||||
),
|
||||
role: str | None = Query(
|
||||
default=None,
|
||||
description="按角色过滤，支持逗号分隔多值（如 admin,assistant）",
|
||||
),
|
||||
account_id: str | None = Query(default=None, description="按渠道账户业务 ID 模糊匹配"),
|
||||
peer_id: str | None = Query(default=None, description="按对端 ID 模糊匹配"),
|
||||
message_id: str | None = Query(default=None, description="按消息业务 ID 模糊匹配"),
|
||||
channel_msg_id: str | None = Query(default=None, description="按渠道消息 ID 模糊匹配"),
|
||||
channel_session_id: str | None = Query(default=None, description="按渠道会话 ID 模糊匹配"),
|
||||
limit: int = Query(default=50, ge=1, le=200, description="分页大小"),
|
||||
offset: int = Query(default=0, ge=0, description="分页偏移"),
|
||||
use_cases=Depends(get_channel_use_cases),
|
||||
@ -169,6 +178,12 @@ async def list_admin_messages(
|
||||
start_time=start_time_dt,
|
||||
end_time=end_time_dt,
|
||||
channel_status=channel_status,
|
||||
role=role,
|
||||
account_id=account_id,
|
||||
peer_id=peer_id,
|
||||
message_id=message_id,
|
||||
channel_msg_id=channel_msg_id,
|
||||
channel_session_id=channel_session_id,
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
operator=operator,
|
||||
@ -182,8 +197,18 @@ async def search_messages(
|
||||
request: Request,
|
||||
keyword: str = Query(..., min_length=2, description="搜索关键词"),
|
||||
channel_type: ChannelType | None = Query(default=None, description="按渠道过滤"),
|
||||
channel_status: str | None = Query(
|
||||
default=None,
|
||||
description="按渠道侧状态过滤，支持逗号分隔多值（如 sent,delivered）",
|
||||
),
|
||||
role: str | None = Query(
|
||||
default=None,
|
||||
description="按角色过滤，支持逗号分隔多值（如 admin,assistant）",
|
||||
),
|
||||
channel_session_id: str | None = Query(default=None, description="按会话 ID 过滤"),
|
||||
peer_id: str | None = Query(default=None, description="按对端 ID 过滤"),
|
||||
message_id: str | None = Query(default=None, description="按消息业务 ID 模糊匹配"),
|
||||
channel_msg_id: str | None = Query(default=None, description="按渠道消息 ID 模糊匹配"),
|
||||
start_time: str | None = Query(default=None, description="起始时间（ISO 8601）"),
|
||||
end_time: str | None = Query(default=None, description="截止时间（ISO 8601）"),
|
||||
limit: int = Query(default=20, ge=1, le=100, description="分页大小"),
|
||||
@ -193,8 +218,8 @@ async def search_messages(
|
||||
) -> dict[str, Any]:
|
||||
"""全文搜索消息（MSG-SEARCH-01）。
|
||||
|
||||
按关键词搜索消息内容，支持渠道、会话、对端 ID、时间范围过滤。
|
||||
对应控制面操作 ``message/search``。
|
||||
按关键词搜索消息内容，支持渠道、状态、角色、会话、对端 ID、消息 ID、
|
||||
渠道消息 ID、时间范围过滤。对应控制面操作 ``message/search``。
|
||||
"""
|
||||
operator = build_operator(current_user, request)
|
||||
start_time_dt = parse_datetime("start_time", start_time)
|
||||
@ -205,8 +230,12 @@ async def search_messages(
|
||||
offset=offset,
|
||||
operator=operator,
|
||||
channel_type=channel_type,
|
||||
channel_status=channel_status,
|
||||
role=role,
|
||||
channel_session_id=channel_session_id,
|
||||
peer_id=peer_id,
|
||||
message_id=message_id,
|
||||
channel_msg_id=channel_msg_id,
|
||||
start_time=start_time_dt,
|
||||
end_time=end_time_dt,
|
||||
)
|
||||
|
||||
@ -82,11 +82,12 @@ class UpdateRetryPolicyRequest(BaseModel):
|
||||
async def list_outbox_messages(
|
||||
request: Request,
|
||||
channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"),
|
||||
channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤"),
|
||||
channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤，支持模糊搜索"),
|
||||
status: OutboxStatus | None = Query(default=None, description="按状态过滤"),
|
||||
message_id: str | None = Query(default=None, description="按消息 ID 过滤"),
|
||||
channel_msg_id: str | None = Query(default=None, description="按渠道侧消息 ID 过滤"),
|
||||
message_id: str | None = Query(default=None, description="按消息 ID 模糊搜索"),
|
||||
channel_msg_id: str | None = Query(default=None, description="按渠道侧消息 ID 模糊搜索"),
|
||||
channel_session_id: str | None = Query(default=None, description="按渠道会话 ID 过滤"),
|
||||
last_error: str | None = Query(default=None, description="按错误关键词筛选"),
|
||||
created_after: str | None = Query(default=None, description="创建时间下界（ISO 8601）"),
|
||||
created_before: str | None = Query(default=None, description="创建时间上界（ISO 8601）"),
|
||||
retry_count_min: int | None = Query(
|
||||
@ -103,14 +104,15 @@ async def list_outbox_messages(
|
||||
created_before_dt = parse_datetime("created_before", created_before)
|
||||
query_filter = OutboxQueryFilter(
|
||||
channel_type=channel_type,
|
||||
channel_account_id=channel_account_id,
|
||||
status=status,
|
||||
message_id=message_id,
|
||||
channel_msg_id=channel_msg_id,
|
||||
channel_session_id=channel_session_id,
|
||||
created_after=created_after_dt,
|
||||
created_before=created_before_dt,
|
||||
retry_count_min=retry_count_min,
|
||||
channel_account_id_like=channel_account_id,
|
||||
message_id_like=message_id,
|
||||
channel_msg_id_like=channel_msg_id,
|
||||
last_error_like=last_error,
|
||||
)
|
||||
result = await use_cases.outbox_management.listOutboxEntries(
|
||||
query_filter,
|
||||
@ -147,6 +149,7 @@ async def get_outbox_trend(
|
||||
start_time: str = Query(..., description="起始时间（ISO 8601，含）"),
|
||||
end_time: str = Query(..., description="结束时间（ISO 8601，含）"),
|
||||
channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"),
|
||||
channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤"),
|
||||
granularity: Literal["minute", "hour", "day"] = Query(default="hour", description="时间粒度"),
|
||||
metric: Literal["queue_depth", "retry_count", "dead_count"] = Query(default="queue_depth", description="度量指标"),
|
||||
use_cases=Depends(get_channel_use_cases),
|
||||
@ -164,6 +167,7 @@ async def get_outbox_trend(
|
||||
start_time=start_time_dt,
|
||||
end_time=end_time_dt,
|
||||
channel_type=channel_type,
|
||||
channel_account_id=channel_account_id,
|
||||
granularity=granularity,
|
||||
metric=metric,
|
||||
)
|
||||
@ -175,6 +179,17 @@ async def get_outbox_trend(
|
||||
@outbox_router.get("/dead-letter", response_model=dict)
|
||||
async def list_dead_letter_outbox(
|
||||
request: Request,
|
||||
channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"),
|
||||
channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤，支持模糊搜索"),
|
||||
message_id: str | None = Query(default=None, description="按消息 ID 模糊搜索"),
|
||||
channel_msg_id: str | None = Query(default=None, description="按渠道侧消息 ID 模糊搜索"),
|
||||
channel_session_id: str | None = Query(default=None, description="按渠道会话 ID 过滤"),
|
||||
last_error: str | None = Query(default=None, description="按错误关键词筛选"),
|
||||
created_after: str | None = Query(default=None, description="创建时间下界（ISO 8601）"),
|
||||
created_before: str | None = Query(default=None, description="创建时间上界（ISO 8601）"),
|
||||
retry_count_min: int | None = Query(
|
||||
default=None, ge=0, description="最小重试次数下界（含），用于筛选已发生重试的条目"
|
||||
),
|
||||
limit: int = Query(default=100, ge=1, le=200, description="分页大小"),
|
||||
offset: int = Query(default=0, ge=0, description="分页偏移"),
|
||||
use_cases=Depends(get_channel_use_cases),
|
||||
@ -182,7 +197,21 @@ async def list_dead_letter_outbox(
|
||||
) -> dict[str, Any]:
|
||||
"""查询死信队列（OBX-04）。对应控制面操作 outbox/dead_letter_list。"""
|
||||
operator = build_operator(current_user, request)
|
||||
created_after_dt = parse_datetime("created_after", created_after)
|
||||
created_before_dt = parse_datetime("created_before", created_before)
|
||||
query_filter = OutboxQueryFilter(
|
||||
channel_type=channel_type,
|
||||
channel_session_id=channel_session_id,
|
||||
created_after=created_after_dt,
|
||||
created_before=created_before_dt,
|
||||
retry_count_min=retry_count_min,
|
||||
channel_account_id_like=channel_account_id,
|
||||
message_id_like=message_id,
|
||||
channel_msg_id_like=channel_msg_id,
|
||||
last_error_like=last_error,
|
||||
)
|
||||
result = await use_cases.outbox_management.listDeadLetterOutboxEntries(
|
||||
query_filter,
|
||||
operator=operator,
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
|
||||
92
backend/server/utils/error_handlers.py
Normal file
92
backend/server/utils/error_handlers.py
Normal file
@ -0,0 +1,92 @@
|
||||
"""全局异常处理器集中实现。
|
||||
|
||||
3 个 handler：
|
||||
- unified_error_handler：处理三模块异常（满足 UnifiedError 协议）
|
||||
- request_validation_error_handler：处理 Pydantic 请求体校验失败
|
||||
- unhandled_exception_handler：兜底原生 Exception
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi.exceptions import RequestValidationError
|
||||
from fastapi.responses import JSONResponse
|
||||
from yuxi.channels.contract.errors import Error as ChannelError
|
||||
from yuxi.external_systems.exceptions import ExternalSystemError
|
||||
from yuxi.scheduler.exceptions import SchedulerError
|
||||
|
||||
from server.utils.error_response import build_error_response, build_error_response_from_exc
|
||||
|
||||
|
||||
async def unified_error_handler(
|
||||
_request: Request,
|
||||
exc: ChannelError | ExternalSystemError | SchedulerError,
|
||||
) -> JSONResponse:
|
||||
"""三模块异常统一处理器。
|
||||
|
||||
三模块异常基类都满足 UnifiedError 协议（status_code/error_code/message/
|
||||
details/trace_id），通过 build_error_response_from_exc() 统一构建响应。
|
||||
不再有 isinstance 特判，不再有查表/类属性分支，逻辑完全相同。
|
||||
|
||||
Args:
|
||||
_request: FastAPI 请求对象（未使用）。
|
||||
exc: 三模块异常之一。
|
||||
|
||||
Returns:
|
||||
统一格式的错误响应。
|
||||
"""
|
||||
return build_error_response_from_exc(exc)
|
||||
|
||||
|
||||
async def unhandled_exception_handler(_request: Request, exc: Exception) -> JSONResponse:
|
||||
"""原生异常兜底处理器。
|
||||
|
||||
所有未被特定异常处理器捕获的 Exception 在此统一兜底，
|
||||
返回结构化 500 响应。trace_id 自动从 ContextVar 获取。
|
||||
|
||||
Args:
|
||||
_request: FastAPI 请求对象（未使用）。
|
||||
exc: 未被捕获的原生异常。
|
||||
|
||||
Returns:
|
||||
500 内部服务器错误响应。
|
||||
"""
|
||||
logging.getLogger("uvicorn.error").exception("unhandled exception", exc_info=exc)
|
||||
return build_error_response(
|
||||
status_code=500,
|
||||
code="INTERNAL",
|
||||
message="Internal server error",
|
||||
)
|
||||
|
||||
|
||||
async def request_validation_error_handler(
|
||||
_request: Request,
|
||||
exc: RequestValidationError,
|
||||
) -> JSONResponse:
|
||||
"""Pydantic 请求体校验失败统一处理器。
|
||||
|
||||
将 FastAPI 默认的 422 + ``{"detail": [...]}`` 转换为项目统一格式
|
||||
422 + ``{"success": false, "error": {...}}``，确保响应格式一致。
|
||||
|
||||
Pydantic v2 的 ``model_validator`` / ``field_validator`` 中抛出的
|
||||
``ValueError`` 子类（含项目的 ``ValidationError``）会被包装为
|
||||
``pydantic.ValidationError``，再由 FastAPI 转为 ``RequestValidationError``，
|
||||
本处理器统一捕获并转换为项目格式。状态码保持 422（FastAPI 标准），
|
||||
仅统一响应结构。
|
||||
"""
|
||||
errors = exc.errors()
|
||||
first = errors[0] if errors else {}
|
||||
loc = first.get("loc", [])
|
||||
field = ".".join(str(part) for part in loc) if loc else ""
|
||||
message = first.get("msg", "request validation failed")
|
||||
# Pydantic 对 ValueError 子类的包装会加 "Value error, " 前缀，去除以还原原始 message
|
||||
if message.startswith("Value error, "):
|
||||
message = message[len("Value error, ") :]
|
||||
return build_error_response(
|
||||
status_code=422,
|
||||
code="VALIDATION_ERROR",
|
||||
message=message,
|
||||
details={"field": field, "errors": errors},
|
||||
)
|
||||
122
backend/server/utils/error_response.py
Normal file
122
backend/server/utils/error_response.py
Normal file
@ -0,0 +1,122 @@
|
||||
"""统一错误响应构建器。
|
||||
|
||||
定义 ``UnifiedError`` 协议（三模块异常基类都实现此协议），
|
||||
提供 ``build_error_response()`` 和 ``build_error_response_from_exc()``
|
||||
统一构建错误响应，确保所有全局异常处理器输出格式一致。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Protocol
|
||||
|
||||
from fastapi.responses import JSONResponse
|
||||
from yuxi.utils.trace_context import get_trace_id
|
||||
|
||||
|
||||
class UnifiedError(Protocol):
|
||||
"""统一异常协议。
|
||||
|
||||
三模块异常基类（ChannelError / ExternalSystemError / SchedulerError）
|
||||
都实现此协议，允许统一 handler 处理。
|
||||
|
||||
字段：
|
||||
status_code: HTTP 状态码。
|
||||
error_code: 稳定错误码（字符串，如 "VALIDATION_ERROR"）。
|
||||
message: 人类可读错误信息。
|
||||
details: 业务字段字典（如 references, retry_after 等）。
|
||||
trace_id: 链路追踪 ID。
|
||||
"""
|
||||
|
||||
status_code: int
|
||||
error_code: str
|
||||
message: str
|
||||
details: dict[str, Any]
|
||||
trace_id: str | None
|
||||
|
||||
|
||||
def _extract_retry_after(exc: UnifiedError) -> dict[str, str]:
|
||||
"""统一提取 Retry-After 响应头。
|
||||
|
||||
429/503 且 details 含 retry_after 时设置。
|
||||
503 且无 retry_after 时设置默认 60 秒（ChannelDegradedError 等降级场景）。
|
||||
channels 的 RateLimitError 已将 retry_after 写入 details（通过 to_dict()）。
|
||||
external_systems 的 RateLimitExceededError 已将 retry_after 写入 details。
|
||||
|
||||
Args:
|
||||
exc: 满足 UnifiedError 协议的异常对象。
|
||||
|
||||
Returns:
|
||||
含 Retry-After 头的字典（无需要时为空字典）。
|
||||
"""
|
||||
headers: dict[str, str] = {}
|
||||
if exc.status_code in (429, 503):
|
||||
retry_after = exc.details.get("retry_after")
|
||||
if retry_after is not None:
|
||||
headers["Retry-After"] = str(retry_after)
|
||||
elif exc.status_code == 503:
|
||||
# 503 降级场景无明确 retry_after 时设置默认 60 秒
|
||||
headers["Retry-After"] = "60"
|
||||
return headers
|
||||
|
||||
|
||||
def build_error_response(
|
||||
status_code: int,
|
||||
code: str,
|
||||
message: str,
|
||||
details: dict[str, Any] | None = None,
|
||||
trace_id: str | None = None,
|
||||
headers: dict[str, str] | None = None,
|
||||
) -> JSONResponse:
|
||||
"""构建统一错误响应。
|
||||
|
||||
所有全局异常处理器调用此函数，确保响应格式一致。
|
||||
trace_id 优先使用传入值，其次从 ContextVar 获取。
|
||||
|
||||
Args:
|
||||
status_code: HTTP 状态码。
|
||||
code: 稳定错误码。
|
||||
message: 人类可读错误信息。
|
||||
details: 业务字段字典。
|
||||
trace_id: 链路追踪 ID（为 None 时从 ContextVar 获取）。
|
||||
headers: 额外响应头（如 Retry-After）。
|
||||
|
||||
Returns:
|
||||
统一格式的 JSONResponse。
|
||||
"""
|
||||
resolved_trace_id = trace_id or get_trace_id()
|
||||
return JSONResponse(
|
||||
status_code=status_code,
|
||||
content={
|
||||
"success": False,
|
||||
"error": {
|
||||
"code": code,
|
||||
"message": message,
|
||||
"trace_id": resolved_trace_id,
|
||||
"details": details or {},
|
||||
},
|
||||
},
|
||||
headers=headers or {},
|
||||
)
|
||||
|
||||
|
||||
def build_error_response_from_exc(exc: UnifiedError) -> JSONResponse:
|
||||
"""从统一异常对象构建错误响应（统一入口）。
|
||||
|
||||
所有满足 UnifiedError 协议的异常都通过此函数构建响应，
|
||||
确保 handler 逻辑完全一致，不再有 isinstance 特判。
|
||||
|
||||
Args:
|
||||
exc: 满足 UnifiedError 协议的异常对象。
|
||||
|
||||
Returns:
|
||||
统一格式的 JSONResponse。
|
||||
"""
|
||||
headers = _extract_retry_after(exc)
|
||||
return build_error_response(
|
||||
status_code=exc.status_code,
|
||||
code=exc.error_code,
|
||||
message=exc.message,
|
||||
details=exc.details,
|
||||
trace_id=exc.trace_id,
|
||||
headers=headers,
|
||||
)
|
||||
78
backend/server/utils/inflight_tracker.py
Normal file
78
backend/server/utils/inflight_tracker.py
Normal file
@ -0,0 +1,78 @@
|
||||
"""在途请求计数器与中间件，用于优雅关停时排空在途请求。
|
||||
|
||||
``InflightRequestTracker`` 通过 ``increment`` / ``decrement`` 维护在途请求
|
||||
计数，计数归零时 ``wait_drained`` 立即返回；超时未归零则抛 ``TimeoutError``，
|
||||
由 ``HostShutdown._waitForInflightRequests`` 捕获并记录告警后强制进入下一步
|
||||
（§11.9）。
|
||||
|
||||
``InflightTrackingMiddleware`` 在每个 HTTP 请求进入时 ``increment``、结束时
|
||||
``decrement``（``finally`` 保证异常路径也执行），确保计数准确。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
from starlette.requests import Request
|
||||
|
||||
__all__ = ["InflightRequestTracker", "InflightTrackingMiddleware"]
|
||||
|
||||
|
||||
class InflightRequestTracker:
|
||||
"""在途请求计数器，用于优雅关停时排空在途请求。
|
||||
|
||||
关键不变量：
|
||||
- ``_count`` 归零时 ``_zero_event`` 被 set，``wait_drained`` 立即返回。
|
||||
- ``decrement`` 防御性归零，避免计数变为负数。
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._count = 0
|
||||
self._zero_event = asyncio.Event()
|
||||
self._zero_event.set() # 初始为 0，已触发
|
||||
|
||||
@property
|
||||
def count(self) -> int:
|
||||
"""当前在途请求数。"""
|
||||
return self._count
|
||||
|
||||
def increment(self) -> None:
|
||||
"""在途请求计数 +1，清除归零事件。"""
|
||||
self._count += 1
|
||||
self._zero_event.clear()
|
||||
|
||||
def decrement(self) -> None:
|
||||
"""在途请求计数 -1，归零时触发归零事件。"""
|
||||
self._count -= 1
|
||||
if self._count <= 0:
|
||||
self._count = 0
|
||||
self._zero_event.set()
|
||||
|
||||
async def wait_drained(self, timeout: float) -> None:
|
||||
"""等待在途请求排空。
|
||||
|
||||
参数：
|
||||
timeout: 等待超时（秒），超时后抛 ``TimeoutError``。
|
||||
|
||||
Raises:
|
||||
TimeoutError: 超时未排空。
|
||||
"""
|
||||
await asyncio.wait_for(self._zero_event.wait(), timeout=timeout)
|
||||
|
||||
|
||||
class InflightTrackingMiddleware(BaseHTTPMiddleware):
|
||||
"""在途请求计数中间件。
|
||||
|
||||
每个请求进入时 ``increment``、结束时 ``decrement``（``finally`` 保证
|
||||
异常路径也执行）。注册时应最后注册（LIFO 顺序中最先执行），确保覆盖
|
||||
所有后续中间件与路由处理。
|
||||
"""
|
||||
|
||||
async def dispatch(self, request: Request, call_next):
|
||||
tracker: InflightRequestTracker = request.app.state.inflight_tracker
|
||||
tracker.increment()
|
||||
try:
|
||||
return await call_next(request)
|
||||
finally:
|
||||
tracker.decrement()
|
||||
@ -15,11 +15,15 @@ from yuxi.knowledge import knowledge_base
|
||||
from yuxi.utils import logger
|
||||
from yuxi.agents.backends.sandbox import init_sandbox_provider, shutdown_sandbox_provider
|
||||
from yuxi import get_version
|
||||
from server.utils.inflight_tracker import InflightRequestTracker
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
|
||||
# 初始化在途请求计数器（InflightTrackingMiddleware 依赖）
|
||||
app.state.inflight_tracker = InflightRequestTracker()
|
||||
|
||||
# 注册外部系统适配器（避免模块导入副作用）
|
||||
try:
|
||||
register_adapters()
|
||||
|
||||
37
backend/server/utils/trace_middleware.py
Normal file
37
backend/server/utils/trace_middleware.py
Normal file
@ -0,0 +1,37 @@
|
||||
"""TraceId 中间件：为每个请求生成/注入 trace_id。
|
||||
|
||||
优先从 X-Request-Id header 读取（支持上游链路传递），
|
||||
否则生成 uuid4。写入 ContextVar 供下游使用。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import Response
|
||||
from yuxi.utils.trace_context import (
|
||||
generate_trace_id,
|
||||
reset_trace_id,
|
||||
set_trace_id,
|
||||
)
|
||||
|
||||
_TRACE_ID_HEADER = "X-Request-Id"
|
||||
|
||||
|
||||
class TraceIdMiddleware(BaseHTTPMiddleware):
|
||||
"""为每个请求注入 trace_id 的中间件。
|
||||
|
||||
注册时应最后注册（LIFO 顺序中最先执行），确保所有后续中间件、
|
||||
异常处理器都能从 ContextVar 读取 trace_id。
|
||||
"""
|
||||
|
||||
async def dispatch(self, request: Request, call_next):
|
||||
trace_id = request.headers.get(_TRACE_ID_HEADER) or generate_trace_id()
|
||||
set_trace_id(trace_id)
|
||||
try:
|
||||
response: Response = await call_next(request)
|
||||
# 回写 trace_id 到响应头，前端可关联
|
||||
response.headers[_TRACE_ID_HEADER] = trace_id
|
||||
return response
|
||||
finally:
|
||||
reset_trace_id()
|
||||
Loading…
Reference in New Issue
Block a user