Kris
b88c0ae29e
新增完整的 channels 限界上下文模块，包含契约层、领域核心层、应用服务、管道编排、插件体系、基础设施组合根等全层级代码，新增飞书与微信 iLink 渠道插件基础结构，补充各类 DTO、端口协议与领域服务实现。
80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
"""错误基类定义。
|
||
|
||
定义所有契约层错误的根类型 ``Error``，提供统一的 ``error_code`` / ``message`` /
|
||
``trace_id`` 字段与 ``to_dict()`` 序列化方法。所有跨边界错误 **必须** 继承自
|
||
``Error``，禁止原生异常穿透至核心层（INV-7 / FF-ERR-01）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from abc import ABC
|
||
from typing import Any
|
||
|
||
|
||
class Error(Exception, ABC):
|
||
"""错误基类。
|
||
|
||
所有契约层错误的根类型。子类通过类变量 ``error_code`` 声明稳定的错误码，
|
||
便于驱动适配器映射为外部协议响应（HTTP 状态码、错误码）。
|
||
|
||
继承 ``Exception`` 以支持 ``raise`` 语句，继承 ``ABC`` 以保留抽象基类
|
||
语义（禁止直接实例化 ``Error``）。
|
||
|
||
字段：
|
||
error_code: 稳定错误码（由子类以类变量覆盖）。
|
||
message: 人类可读错误信息。
|
||
trace_id: 调用链路追踪 ID，由调用方显式注入，用于跨链路关联（INV-10）。
|
||
未注入时为空字符串，契约层不再从 ContextVar 反向解析，以避免
|
||
违反 §6.1（契约层禁止依赖任何实现层）。
|
||
|
||
属性：
|
||
status_code: HTTP 状态码，由 ``error_code`` 查 ``CHANNEL_ERROR_STATUS_MAP``
|
||
得出，未命中时默认 500。
|
||
details: 业务字段字典，从 ``to_dict()`` 提取并排除 ``error_code`` /
|
||
``message`` / ``trace_id``，子类覆写 ``to_dict()`` 后自动生效。
|
||
"""
|
||
|
||
error_code: str = "ERROR"
|
||
|
||
def __init__(self, message: str, *, trace_id: str = "") -> None:
|
||
super().__init__(message)
|
||
self.message = message
|
||
# trace_id 由调用方显式注入：契约层禁止依赖实现层（§6.1），
|
||
# 因此不再从 ContextVar 反向解析。adapter/model 层抛错时由其
|
||
# 自行注入当前 trace_id（P1-5/P1-6）。
|
||
self.trace_id = trace_id
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
"""序列化为字典，便于跨层传递、日志记录与 API 响应。"""
|
||
return {
|
||
"error_code": self.error_code,
|
||
"message": self.message,
|
||
"trace_id": self.trace_id,
|
||
}
|
||
|
||
@property
|
||
def status_code(self) -> int:
|
||
"""HTTP 状态码：由 error_code 查表得出。
|
||
|
||
查询 ``CHANNEL_ERROR_STATUS_MAP``，未命中时返回 500。使用延迟 import
|
||
以避免核心层与状态码映射表之间的循环依赖。
|
||
"""
|
||
from yuxi.channels.contract.errors._status_map import (
|
||
CHANNEL_ERROR_STATUS_MAP,
|
||
)
|
||
|
||
return CHANNEL_ERROR_STATUS_MAP.get(self.error_code, 500)
|
||
|
||
@property
|
||
def details(self) -> dict[str, Any]:
|
||
"""业务字段：从 to_dict() 提取，排除 error_code/message/trace_id。
|
||
|
||
子类覆写的 to_dict() 自动生效，业务字段（resource/id/retry_after_ms 等）
|
||
自动进入 details。
|
||
"""
|
||
full_dict = self.to_dict()
|
||
return {k: v for k, v in full_dict.items() if k not in ("error_code", "message", "trace_id")}
|
||
|
||
def __str__(self) -> str:
|
||
return f"[{self.error_code}] {self.message}"
|