You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python监听器模式下文件未导入致事件未注册的替代解决方法

事件驱动系统的处理器注册问题

我希望构建一个事件驱动系统,让函数可以订阅并处理特定事件。我选择使用装饰器实现该功能,相关代码如下:

代码实现

decorators.py

from collections import defaultdict
from typing import Callable

EVENT_HANDLERS: dict[str, set[Callable]] = defaultdict(set)


def my_event_listener(event_type: str):
    """
    A decorator to subscribe to events with the given event_type.
    """

    def decorator(callback_fn):
        EVENT_HANDLERS[event_type].add(callback_fn)
        return callback_fn

    return decorator

events.py

def create_event(event: dict):
    # 注:原代码存在变量未定义问题,此处修正为取event字典中的"event_type"键
    for handler_fn in EVENT_HANDLERS[event["event_type"]]:
        handler_fn(event)

handlers.py

import logging

logger = logging.getLogger(__name__)

@my_event_listener(event_type="test")
def handle_test_event(event):
    logger.info(f"Test event received with payload {event['payload']}")

该实现运行正常,但当handlers.py未被代码库中其他模块导入时会出现问题。Python仅在文件被其他地方导入时才会加载它,此时handlers.py没有被导入的理由,因此永远不会被加载,装饰器也不会执行,回调函数无法完成注册。

我认为这并非装饰器实现方式的问题,即便使用类等其他方式,也会遇到相同的导入问题。

除了在某处维护处理器注册表之外,是否有其他解决办法?


解决方案

1. 自动扫描并导入处理器模块

在项目入口文件(如main.py)中添加自动扫描逻辑,遍历处理器目录并导入所有相关模块,无需手动维护导入列表:

import os
import importlib

def load_handlers():
    # 处理器所在目录,根据实际项目路径调整
    handlers_dir = os.path.join(os.path.dirname(__file__), "handlers")
    # 遍历目录下所有非初始化py文件
    for filename in os.listdir(handlers_dir):
        if filename.endswith(".py") and not filename.startswith("__"):
            # 构造模块名
            module_name = f"handlers.{filename[:-3]}"
            # 动态导入模块
            importlib.import_module(module_name)

# 系统启动时执行加载
if __name__ == "__main__":
    load_handlers()
    # 后续启动逻辑

2. 核心模块显式导入处理器

在事件系统的核心模块(如events.py)中直接导入所有处理器模块,确保模块被加载:

# events.py
from handlers import handlers  # 导入整个处理器模块
# 或者导入特定处理器函数
from handlers.handlers import handle_test_event

def create_event(event: dict):
    for handler_fn in EVENT_HANDLERS[event["event_type"]]:
        handler_fn(event)

这种方式适合小型项目,缺点是新增处理器时需要手动更新导入语句。

3. 利用包的__init__.py自动导入

在handlers目录下创建__init__.py文件,在其中导入所有处理器函数:

# handlers/__init__.py
from .handlers import handle_test_event

然后在项目入口处导入整个handlers包:

import handlers

当包被导入时,__init__.py中的代码会执行,触发处理器模块的加载和装饰器注册。


内容的提问来源于stack exchange,提问作者Tom

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.01 06:25:33