plaita.event.core

事件系统核心：Event、EventSubscription、EventBus、EventStorage、EventSubscriptionStorage、EventProcessingTracker、RetryPolicy。

core

事件系统的核心定义和接口

Event

Bases: BaseModel

事件对象，包含事件类型、数据和元数据

EventSubscription

Bases: BaseModel

事件订阅信息

mark_event_processed

mark_event_processed(event_id: str) -> None

Mark an event as processed by this subscription.

源代码位于： plaita/event/core.py

def mark_event_processed(self, event_id: str) -> None:
    """Mark an event as processed by this subscription."""
    self.processed_events.add(event_id)

is_event_processed

is_event_processed(event_id: str) -> bool

Check if an event has already been processed.

源代码位于： plaita/event/core.py

def is_event_processed(self, event_id: str) -> bool:
    """Check if an event has already been processed."""
    return event_id in self.processed_events

matches_event

matches_event(event: Event, context: Dict[str, Any]) -> bool

检查事件是否匹配此订阅

源代码位于： plaita/event/core.py

def matches_event(self, event: Event, context: Dict[str, Any]) -> bool:
    """检查事件是否匹配此订阅"""

    # 检查事件类型
    if self.event_type and event.event_type != self.event_type:
        return False

    # 检查correlation_id
    if self.correlation_id and event.correlation_id != self.correlation_id:
        return False

    express_prefix = context.get("EXPRESS_PREFIX", "$")
    flow_id = context.get(f"{express_prefix}FLOW_ID")
    node_id = context.get(f"{express_prefix}LAST_NODE")
    # 检查flow_id,在context中
    if self.flow_id and flow_id and self.flow_id != flow_id:
        return False

    # 检查node_id,在context中
    if self.node_id and node_id and self.node_id != node_id:
        return False

    # 检查过滤条件：顶层键做基本字典匹配；含 "." 的键按点路径遍历嵌套
    # payload（与 EventNode.can_handle_event 的遍历语义一致——挂起侧把
    # node.event_filter 原样存成 filter_condition，两处必须命中同一批事件，
    # 否则事件驱动 resume 静默失效）。中间层非 dict 视为不匹配。
    if self.filter_condition:
        for key, expected_value in self.filter_condition.items():
            if "." in key:
                current: Any = event.data
                for part in key.split("."):
                    if isinstance(current, dict) and part in current:
                        current = current[part]
                    else:
                        return False
                if current != expected_value:
                    return False
            elif key in event.data:
                if event.data[key] != expected_value:
                    return False
            elif hasattr(event, key):
                if getattr(event, key) != expected_value:
                    return False
            else:
                # 如果事件中没有指定的字段，则不匹配
                return False

    return True

EventStorage

Bases: ABC

事件存储接口，定义了如何存储和检索事件

store_event abstractmethod async

store_event(event: Event) -> str

存储事件并返回事件ID

源代码位于： plaita/event/core.py

@abstractmethod
async def store_event(self, event: Event) -> str:
    """存储事件并返回事件ID"""
    pass

get_event abstractmethod async

get_event(event_id: str) -> Optional[Event]

根据ID获取事件

源代码位于： plaita/event/core.py

@abstractmethod
async def get_event(self, event_id: str) -> Optional[Event]:
    """根据ID获取事件"""
    pass

list_events abstractmethod async

list_events(event_type: Optional[str] = None, start_time: Optional[float] = None, end_time: Optional[float] = None, limit: int = 100) -> List[Event]

列出符合条件的事件

源代码位于： plaita/event/core.py

@abstractmethod
async def list_events(self, 
                     event_type: Optional[str] = None, 
                     start_time: Optional[float] = None,
                     end_time: Optional[float] = None,
                     limit: int = 100) -> List[Event]:
    """列出符合条件的事件"""
    pass

delete_event abstractmethod async

delete_event(event_id: str) -> bool

删除事件

源代码位于： plaita/event/core.py

@abstractmethod
async def delete_event(self, event_id: str) -> bool:
    """删除事件"""
    pass

batch_store_events async

batch_store_events(events: List[Event]) -> List[str]

批量存储事件（默认实现，可被覆盖以提高性能）

源代码位于： plaita/event/core.py

async def batch_store_events(self, events: List[Event]) -> List[str]:
    """批量存储事件（默认实现，可被覆盖以提高性能）"""
    event_ids = []
    for event in events:
        event_id = await self.store_event(event)
        event_ids.append(event_id)
    return event_ids

EventSubscriptionStorage

Bases: ABC

事件订阅存储接口，定义了如何存储和检索事件订阅信息

store_subscription abstractmethod async

store_subscription(subscription: EventSubscription) -> str

存储事件订阅并返回订阅ID

源代码位于： plaita/event/core.py

@abstractmethod
async def store_subscription(self, subscription: EventSubscription) -> str:
    """存储事件订阅并返回订阅ID"""
    pass

get_subscription abstractmethod async

get_subscription(subscription_id: str) -> Optional[EventSubscription]

根据ID获取事件订阅

源代码位于： plaita/event/core.py

@abstractmethod
async def get_subscription(self, subscription_id: str) -> Optional[EventSubscription]:
    """根据ID获取事件订阅"""
    pass

list_subscriptions abstractmethod async

list_subscriptions(event_type: Optional[str] = None, correlation_id: Optional[str] = None, flow_id: Optional[str] = None, node_id: Optional[str] = None) -> List[EventSubscription]

列出符合条件的事件订阅

源代码位于： plaita/event/core.py

@abstractmethod
async def list_subscriptions(self, 
                           event_type: Optional[str] = None,
                           correlation_id: Optional[str] = None,
                           flow_id: Optional[str] = None,
                           node_id: Optional[str] = None) -> List[EventSubscription]:
    """列出符合条件的事件订阅"""
    pass

delete_subscription abstractmethod async

delete_subscription(subscription_id: str) -> bool

删除事件订阅

源代码位于： plaita/event/core.py

@abstractmethod
async def delete_subscription(self, subscription_id: str) -> bool:
    """删除事件订阅"""
    pass

find_matching_subscriptions async

find_matching_subscriptions(event: Event, context: Dict[str, Any] = None) -> List[EventSubscription]

查找与事件匹配的所有订阅

源代码位于： plaita/event/core.py

async def find_matching_subscriptions(self, event: Event, context: Dict[str, Any] = None) -> List[EventSubscription]:
    """查找与事件匹配的所有订阅"""
    subscriptions = await self.list_subscriptions(correlation_id=event.correlation_id, event_type=event.event_type)
    return [sub for sub in subscriptions if sub.matches_event(event, context)]

mark_event_processed abstractmethod async

mark_event_processed(subscription_id: str, event_id: str) -> bool

原子操作：标记事件为已处理状态

源代码位于： plaita/event/core.py

@abstractmethod
async def mark_event_processed(self, subscription_id: str, event_id: str) -> bool:
    """原子操作：标记事件为已处理状态"""
    pass

batch_mark_processed async

batch_mark_processed(subscription_id: str, event_ids: List[str]) -> bool

批量标记事件为已处理（默认实现）

源代码位于： plaita/event/core.py

async def batch_mark_processed(self, subscription_id: str, event_ids: List[str]) -> bool:
    """批量标记事件为已处理（默认实现）"""
    results = []
    for event_id in event_ids:
        result = await self.mark_event_processed(subscription_id, event_id)
        results.append(result)
    return all(results)

EventProcessingTracker

Bases: ABC

统一的事件处理记录接口，用于全局去重和状态追踪

mark_event_processed abstractmethod async

mark_event_processed(event_id: str, handler_id: str) -> bool

标记事件为已处理状态

源代码位于： plaita/event/core.py

@abstractmethod
async def mark_event_processed(self, event_id: str, handler_id: str) -> bool:
    """标记事件为已处理状态"""
    pass

is_event_processed abstractmethod async

is_event_processed(event_id: str, handler_id: str) -> bool

检查事件是否已被处理

源代码位于： plaita/event/core.py

@abstractmethod
async def is_event_processed(self, event_id: str, handler_id: str) -> bool:
    """检查事件是否已被处理"""
    pass

cleanup_old_records abstractmethod async

cleanup_old_records(max_age_seconds: int = 86400) -> int

清理旧的处理记录

源代码位于： plaita/event/core.py

@abstractmethod
async def cleanup_old_records(self, max_age_seconds: int = 86400) -> int:
    """清理旧的处理记录"""
    pass

record_processing_attempt abstractmethod async

record_processing_attempt(event_id: str, handler_id: str, status: str, error: Optional[str] = None) -> None

记录处理尝试，包括成功和失败

源代码位于： plaita/event/core.py

@abstractmethod
async def record_processing_attempt(self, event_id: str, handler_id: str, 
                                  status: str, error: Optional[str] = None) -> None:
    """记录处理尝试，包括成功和失败"""
    pass

get_processing_history abstractmethod async

get_processing_history(event_id: str) -> List[Dict[str, Any]]

获取事件的处理历史

源代码位于： plaita/event/core.py

@abstractmethod
async def get_processing_history(self, event_id: str) -> List[Dict[str, Any]]:
    """获取事件的处理历史"""
    pass

RetryPolicy

Bases: BaseModel

事件处理重试策略

EventBus

Bases: ABC

事件总线接口，定义了事件发布和订阅的核心功能

matches_event_type staticmethod

matches_event_type(handler_event_type: Optional[str], actual_event_type: str) -> bool

检查事件类型是否匹配处理器的事件类型模式

支持的匹配规则： 1. None 或 "" - 匹配所有事件类型 2. 精确匹配 - "user.login" 只匹配 "user.login" 3. 前缀通配符 - "user." 匹配 "user.login", "user.logout" 等 4. 后缀通配符 - ".login" 匹配 "user.login", "admin.login" 等 5. 中间通配符 - ".user.*" 匹配 "app.user.login", "sys.user.update" 等

参数：

名称	类型	描述	默认
handler_event_type	Optional[str]	处理器注册的事件类型模式	必需
actual_event_type	str	实际事件类型	必需

返回：

名称	类型	描述
bool	bool	是否匹配

源代码位于： plaita/event/core.py

@staticmethod
def matches_event_type(handler_event_type: Optional[str], actual_event_type: str) -> bool:
    """
    检查事件类型是否匹配处理器的事件类型模式

    支持的匹配规则：
    1. None 或 "*" - 匹配所有事件类型
    2. 精确匹配 - "user.login" 只匹配 "user.login"
    3. 前缀通配符 - "user.*" 匹配 "user.login", "user.logout" 等
    4. 后缀通配符 - "*.login" 匹配 "user.login", "admin.login" 等
    5. 中间通配符 - "*.user.*" 匹配 "app.user.login", "sys.user.update" 等

    Args:
        handler_event_type: 处理器注册的事件类型模式
        actual_event_type: 实际事件类型

    Returns:
        bool: 是否匹配
    """
    # None 或 "*" 匹配所有事件
    if not handler_event_type or handler_event_type == "*":
        return True

    # 精确匹配
    if handler_event_type == actual_event_type:
        return True

    # 通配符匹配，使用fnmatch实现Unix shell风格的通配符
    return fnmatch.fnmatch(actual_event_type, handler_event_type)

publish abstractmethod async

publish(event: Union[Event, str, Dict[str, Any]], prevent_duplicate_consumption: bool = True, **kwargs) -> str

发布事件

源代码位于： plaita/event/core.py

@abstractmethod
async def publish(self, event: Union[Event, str, Dict[str, Any]], 
                prevent_duplicate_consumption: bool = True, 
                **kwargs) -> str:
    """发布事件"""
    pass

batch_publish async

batch_publish(events: List[Union[Event, str, Dict[str, Any]]], prevent_duplicate_consumption: bool = True) -> List[str]

批量发布事件（默认实现）

源代码位于： plaita/event/core.py

async def batch_publish(self, events: List[Union[Event, str, Dict[str, Any]]],
                      prevent_duplicate_consumption: bool = True) -> List[str]:
    """批量发布事件（默认实现）"""
    event_ids = []
    for event in events:
        event_id = await self.publish(event, prevent_duplicate_consumption)
        event_ids.append(event_id)
    return event_ids

register_subscription abstractmethod async

register_subscription(event_type: str, filter_condition: Optional[Dict[str, Any]] = None, correlation_id: Optional[str] = None, flow_id: Optional[str] = None, node_id: Optional[str] = None, timeout: Optional[float] = None, tenant_id: Optional[str] = None) -> str

注册事件订阅信息

源代码位于： plaita/event/core.py

@abstractmethod
async def register_subscription(self,
                              event_type: str,
                              filter_condition: Optional[Dict[str, Any]] = None,
                              correlation_id: Optional[str] = None,
                              flow_id: Optional[str] = None,
                              node_id: Optional[str] = None,
                              timeout: Optional[float] = None,
                              tenant_id: Optional[str] = None) -> str:
    """注册事件订阅信息"""
    pass

unregister_subscription abstractmethod async

unregister_subscription(subscription_id: str) -> bool

取消事件订阅

源代码位于： plaita/event/core.py

@abstractmethod
async def unregister_subscription(self, subscription_id: str) -> bool:
    """取消事件订阅"""
    pass

wait_for_event abstractmethod async

wait_for_event(event_type: str, timeout: Optional[float] = None, condition: Optional[Callable[[Event], bool]] = None) -> Event

等待特定类型的事件发生

源代码位于： plaita/event/core.py

@abstractmethod
async def wait_for_event(self, event_type: str, 
                       timeout: Optional[float] = None,
                       condition: Optional[Callable[[Event], bool]] = None) -> Event:
    """等待特定类型的事件发生"""
    pass

register_handler abstractmethod async

register_handler(event_type: Optional[str] = None, handler: EventHandler = None, filter_condition: Optional[Dict[str, Any]] = None, retry_policy: Optional[RetryPolicy] = None) -> str

注册事件处理器

参数：

名称	类型	描述	默认
event_type	Optional[str]	事件类型模式，支持通配符： - None 或 "": 监听所有事件 - "prefix.": 监听prefix开头的所有事件 - ".suffix": 监听以suffix结尾的所有事件 - ".middle.*": 监听包含middle的所有事件	None
handler	EventHandler	事件处理器函数	None
filter_condition	Optional[Dict[str, Any]]	额外的过滤条件	None
retry_policy	Optional[RetryPolicy]	重试策略	None

返回：

名称	类型	描述
str	str	处理器ID

源代码位于： plaita/event/core.py

@abstractmethod
async def register_handler(self, event_type: Optional[str] = None, 
                         handler: EventHandler = None,
                         filter_condition: Optional[Dict[str, Any]] = None,
                         retry_policy: Optional[RetryPolicy] = None) -> str:
    """
    注册事件处理器

    Args:
        event_type: 事件类型模式，支持通配符：
                   - None 或 "*": 监听所有事件
                   - "prefix.*": 监听prefix开头的所有事件
                   - "*.suffix": 监听以suffix结尾的所有事件
                   - "*.middle.*": 监听包含middle的所有事件
        handler: 事件处理器函数
        filter_condition: 额外的过滤条件
        retry_policy: 重试策略

    Returns:
        str: 处理器ID
    """
    pass

get_event abstractmethod async

get_event(event_id: str) -> Event

获取事件

源代码位于： plaita/event/core.py

@abstractmethod
async def get_event(self, event_id: str) -> Event:
    """获取事件"""
    pass

publish_sync

publish_sync(event: Union[Event, str, Dict[str, Any]], prevent_duplicate_consumption: bool = True, **kwargs) -> str

Synchronous publish — delegates to the async publish method.

Detects whether an event loop is already running and uses _run_async_from_sync to bridge appropriately.

On backends that dispatch handlers from publish (InMemory), the bridge drains this publish's dispatch tasks — bounded by publish_sync_drain_timeout — before asyncio.run closes the private loop. Without that drain a slow async handler is cancelled by loop shutdown and silently never runs (plaita#218).

publish itself stays fire-and-forget: async callers still return before handlers finish.

源代码位于： plaita/event/core.py

def publish_sync(self, event: Union[Event, str, Dict[str, Any]],
                 prevent_duplicate_consumption: bool = True,
                 **kwargs) -> str:
    """Synchronous publish — delegates to the async ``publish`` method.

    Detects whether an event loop is already running and uses
    ``_run_async_from_sync`` to bridge appropriately.

    On backends that dispatch handlers from ``publish`` (InMemory), the
    bridge drains this publish's dispatch tasks — bounded by
    ``publish_sync_drain_timeout`` — before ``asyncio.run`` closes the
    private loop.  Without that drain a slow async handler is cancelled
    by loop shutdown and silently never runs (plaita#218).

    ``publish`` itself stays fire-and-forget: async callers still return
    before handlers finish.
    """
    return _run_async_from_sync(
        self._publish_and_drain(event, prevent_duplicate_consumption, **kwargs)
    )

bind_trusted_tenant

bind_trusted_tenant(event: Event) -> Event

发布时绑定受信租户（plaita#191）。

租户只从进程内 ContextVar 取（worker / console 请求上下文已 set）， 绝不读 event.data["tenant_id"]——那是发布者/客户端可控载荷，正是 历史跨租户 resume 注入的来源。已显式设置 tenant_id 的事件不覆盖： 服务内部补偿链（delay / approval / http_callback 的 resume 事件）由发布 方从服务端任务配置显式写入受信租户。

源代码位于： plaita/event/core.py

def bind_trusted_tenant(event: Event) -> Event:
    """发布时绑定受信租户（plaita#191）。

    租户**只**从进程内 ContextVar 取（worker / console 请求上下文已 set），
    **绝不**读 ``event.data["tenant_id"]``——那是发布者/客户端可控载荷，正是
    历史跨租户 resume 注入的来源。已显式设置 ``tenant_id`` 的事件不覆盖：
    服务内部补偿链（delay / approval / http_callback 的 resume 事件）由发布
    方从服务端任务配置显式写入受信租户。
    """
    if event.tenant_id is None:
        # 局部 import：tenant_context 是顶层无依赖模块，事件层引用安全；
        # 局部化只为让 event.core 的 import 面保持最小。
        from plaita.tenant_context import current_tenant

        event.tenant_id = current_tenant()
    return event

event_handler

event_handler(event_bus: EventBus, event_type: Optional[str] = None, filter_condition: Optional[Dict[str, Any]] = None, retry_policy: Optional[RetryPolicy] = None)

事件处理器装饰器

注册是异步的 (EventBus.register_handler 是 coroutine)。本装饰器在 模块导入期被触发, 当时未必有 running loop, 直接 asyncio.create_task 会抛 RuntimeError 或 (即使成功) task 引用丢失被 GC 回收。

处理策略: 1. 检测到 running loop —— create_task 并把引用存进模块级集合, 由 loop 自然驱动; 任务完成后从集合移除 (done_callback)。 2. 没 running loop (典型场景: 模块导入期) —— 把 register coroutine 存进 模块级待办列表, 暴露 flush_pending_handler_registrations() 让用户 在 loop 起来后显式 await。

源代码位于： plaita/event/core.py

def event_handler(event_bus: EventBus,
                 event_type: Optional[str] = None,
                 filter_condition: Optional[Dict[str, Any]] = None,
                 retry_policy: Optional[RetryPolicy] = None):
    """事件处理器装饰器

    注册是异步的 (``EventBus.register_handler`` 是 coroutine)。本装饰器在
    模块导入期被触发, 当时未必有 running loop, 直接 ``asyncio.create_task``
    会抛 ``RuntimeError`` 或 (即使成功) task 引用丢失被 GC 回收。

    处理策略:
    1. 检测到 running loop —— ``create_task`` 并把引用存进模块级集合, 由 loop
       自然驱动; 任务完成后从集合移除 (``done_callback``)。
    2. 没 running loop (典型场景: 模块导入期) —— 把 register coroutine 存进
       模块级待办列表, 暴露 ``flush_pending_handler_registrations()`` 让用户
       在 loop 起来后显式 await。
    """

    def decorator(func: EventHandler):
        async def register():
            await event_bus.register_handler(
                event_type=event_type,
                handler=func,
                filter_condition=filter_condition,
                retry_policy=retry_policy
            )

        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            # 没运行中的 loop —— 暂存, 等用户在 async 上下文里 flush
            _pending_handler_registrations.append(register)
            return func

        task = loop.create_task(register())
        _handler_registration_tasks.add(task)
        task.add_done_callback(_handler_registration_tasks.discard)
        return func

    return decorator

flush_pending_handler_registrations async

flush_pending_handler_registrations() -> None

驱动所有在无 running loop 期间被 @event_handler 排队的注册任务。

典型用法: 应用启动、asyncio.run 入口里 await flush_...() 一次。

源代码位于： plaita/event/core.py

async def flush_pending_handler_registrations() -> None:
    """驱动所有在无 running loop 期间被 ``@event_handler`` 排队的注册任务。

    典型用法: 应用启动、``asyncio.run`` 入口里 ``await flush_...()`` 一次。
    """
    pending = list(_pending_handler_registrations)
    _pending_handler_registrations.clear()
    for register in pending:
        await register()