fix: preserve legacy event and db dto contracts

This commit is contained in:
jxxghp
2026-08-21 22:33:21 +08:00
parent 1b253510e1
commit bb0964622b
3 changed files with 57 additions and 15 deletions
+22 -5
View File
@@ -9,7 +9,6 @@
""" """
import time import time
from collections.abc import Awaitable, Callable from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any, Tuple, List, Optional from typing import Any, Tuple, List, Optional
from sqlalchemy import delete as sqlalchemy_delete, select from sqlalchemy import delete as sqlalchemy_delete, select
@@ -28,13 +27,31 @@ AfterCommitEffect = Callable[[int], None]
AsyncAfterCommitEffect = Callable[[int], Awaitable[None]] AsyncAfterCommitEffect = Callable[[int], Awaitable[None]]
@dataclass(frozen=True, slots=True)
class SubscribeStageResult: class SubscribeStageResult:
"""Oper 暂存结果,按 Application 端口需要暴露最小只读状态。""" """Oper 暂存结果,按 Application 端口需要暴露最小只读状态。"""
subscribe_id: int __slots__ = ("_subscribe_id", "_message", "_created")
message: str
created: bool def __init__(self, subscribe_id: int, message: str, created: bool) -> None:
"""保存写入后的订阅 ID、消息和是否创建标志。"""
self._subscribe_id = subscribe_id
self._message = message
self._created = created
@property
def subscribe_id(self) -> int:
"""返回已暂存或已存在的订阅 ID。"""
return self._subscribe_id
@property
def message(self) -> str:
"""返回暂存结果的人类可读消息。"""
return self._message
@property
def created(self) -> bool:
"""返回本次暂存是否创建了新记录。"""
return self._created
def _normalize_integer_flags(payload: dict, fields: Tuple[str, ...] = INTEGER_FLAG_FIELDS) -> dict: def _normalize_integer_flags(payload: dict, fields: Tuple[str, ...] = INTEGER_FLAG_FIELDS) -> dict:
+28 -5
View File
@@ -130,17 +130,40 @@ EVENT_CONTRACTS = {
} }
def get_event_contract(event_type: EventType | ChainEventType) -> EventContract: def normalize_event_type(
"""返回 enum 事件的完整登记契约。""" event_type: EventType | ChainEventType | str,
return EVENT_CONTRACTS[event_type] ) -> EventType | ChainEventType | str:
"""把旧 SDK 传入的已知字符串恢复为 enum,未知扩展值保持原样。"""
if not isinstance(event_type, str):
return event_type
for enum_type in (EventType, ChainEventType):
try:
return enum_type(event_type)
except ValueError:
continue
return event_type
def get_event_contract(
event_type: EventType | ChainEventType | str,
) -> EventContract:
"""返回已登记事件的完整契约,兼容传入 enum value 字符串。"""
normalized = normalize_event_type(event_type)
if isinstance(normalized, str):
raise KeyError(normalized)
return EVENT_CONTRACTS[normalized]
def validate_event_payload( def validate_event_payload(
event_type: EventType | ChainEventType, event_type: EventType | ChainEventType | str,
payload: Any, payload: Any,
) -> tuple[str, ...]: ) -> tuple[str, ...]:
"""在发送边界诊断首批 typed payload,保持原对象和插件 dict 形状不变。""" """在发送边界诊断首批 typed payload,保持原对象和插件 dict 形状不变。"""
model = get_event_contract(event_type).payload_model try:
model = get_event_contract(event_type).payload_model
except KeyError:
# 动态插件在旧 ABI 下可能使用宿主枚举之外的字符串事件。
return ()
if model is None or payload is None: if model is None or payload is None:
return () return ()
if isinstance(payload, model): if isinstance(payload, model):
+7 -5
View File
@@ -20,7 +20,7 @@ from app.runtime.event.binding import (
from app.runtime.event.dispatch import EventDispatcher from app.runtime.event.dispatch import EventDispatcher
from app.runtime.event.errors import EventErrorNotifier, EventErrorPolicy from app.runtime.event.errors import EventErrorNotifier, EventErrorPolicy
from app.runtime.event.registry import EventRegistry from app.runtime.event.registry import EventRegistry
from app.runtime.event.contracts import validate_event_payload from app.runtime.event.contracts import normalize_event_type, validate_event_payload
from app.runtime.correlation import get_correlation_id from app.runtime.correlation import get_correlation_id
from app.runtime.observability import record_metric from app.runtime.observability import record_metric
@@ -35,7 +35,7 @@ class Event:
事件类,封装事件的基本信息 事件类,封装事件的基本信息
""" """
def __init__(self, event_type: Union[EventType, ChainEventType], def __init__(self, event_type: Union[EventType, ChainEventType, str],
event_data: Optional[Union[Dict, ChainEventData]] = None, event_data: Optional[Union[Dict, ChainEventData]] = None,
priority: Optional[int] = DEFAULT_EVENT_PRIORITY, priority: Optional[int] = DEFAULT_EVENT_PRIORITY,
correlation_id: Optional[str] = None): correlation_id: Optional[str] = None):
@@ -45,11 +45,12 @@ class Event:
:param priority: 可选,事件的优先级,默认为 10 :param priority: 可选,事件的优先级,默认为 10
:param correlation_id: 生产事件时固化的请求关联 ID :param correlation_id: 生产事件时固化的请求关联 ID
""" """
event_type = normalize_event_type(event_type)
payload_problems = validate_event_payload(event_type, event_data) payload_problems = validate_event_payload(event_type, event_data)
if payload_problems: if payload_problems:
logger.warning( logger.warning(
"事件 %s payload 与登记契约不一致:%s;当前保留旧 payload 继续投递", "事件 %s payload 与登记契约不一致:%s;当前保留旧 payload 继续投递",
event_type.value, getattr(event_type, "value", event_type),
"; ".join(payload_problems), "; ".join(payload_problems),
) )
self.event_id = str(uuid.uuid4()) # 事件ID self.event_id = str(uuid.uuid4()) # 事件ID
@@ -63,7 +64,8 @@ class Event:
重写 __repr__ 方法,用于返回事件的详细信息,包括事件类型、事件ID和优先级 重写 __repr__ 方法,用于返回事件的详细信息,包括事件类型、事件ID和优先级
""" """
event_kind = Event.get_event_kind(self.event_type) event_kind = Event.get_event_kind(self.event_type)
return f"<{event_kind}: {self.event_type.value}, ID: {self.event_id}, Priority: {self.priority}>" event_name = getattr(self.event_type, "value", self.event_type)
return f"<{event_kind}: {event_name}, ID: {self.event_id}, Priority: {self.priority}>"
def __lt__(self, other): def __lt__(self, other):
""" """
@@ -73,7 +75,7 @@ class Event:
return self.priority < other.priority return self.priority < other.priority
@staticmethod @staticmethod
def get_event_kind(event_type: Union[EventType, ChainEventType]) -> str: def get_event_kind(event_type: Union[EventType, ChainEventType, str]) -> str:
""" """
根据事件类型判断事件是广播事件还是链式事件 根据事件类型判断事件是广播事件还是链式事件
:param event_type: 事件类型,支持 EventType 或 ChainEventType :param event_type: 事件类型,支持 EventType 或 ChainEventType