如何在Python中查找被指定装饰器修饰的类？（含Temporal场景）

疑问 #

Python有没有内置机制可以自动找到被Temporal的@workflow.defn和@activity.defn修饰的类？比如类似importlib.find_all_classes_decorated_by(workflow.defn)的方法，这样就能无需显式配置自动注册到Worker。

解答 #

Python标准库没有直接提供这种开箱即用的扫描API，但可以通过以下方式实现：

• 遍历模块+检查装饰器标记
  用importlib遍历指定包下的所有模块，导入后用inspect遍历模块内的类，通过Temporal装饰器给类添加的特殊标记来识别：

  import importlib
  import inspect
  from pathlib import Path
  from temporalio import workflow, activity
  
  def find_temporal_decorated_classes(package_name: str):
      workflows = []
      activities = []
      # 导入根包
      root_package = importlib.import_module(package_name)
      root_path = Path(root_package.__file__).parent
  
      # 遍历所有.py文件
      for py_file in root_path.rglob("*.py"):
          # 跳过__pycache__和__init__.py（可选）
          if py_file.parent.name == "__pycache__" or py_file.name == "__init__.py":
              continue
          # 生成模块路径
          module_rel_path = py_file.relative_to(root_path).with_suffix("")
          module_path = f"{package_name}.{'.'.join(module_rel_path.parts)}"
          try:
              module = importlib.import_module(module_path)
              # 遍历模块内的所有类
              for _, cls in inspect.getmembers(module, inspect.isclass):
                  # 检查Temporal工作流标记（根据SDK实际实现调整）
                  if hasattr(cls, "_is_workflow"):
                      workflows.append(cls)
                  # 检查Temporal活动标记
                  elif hasattr(cls, "_is_activity"):
                      activities.append(cls)
          except ImportError:
              continue
      return workflows, activities
  
  # 使用示例：扫描你的业务包
  workflows, activities = find_temporal_decorated_classes("your_business_package")
  # 注册到Worker
  # worker = Worker(..., workflows=workflows, activities=activities)
  

  注：Temporal的@workflow.defn和@activity.defn会给类添加内部标记（比如_is_workflow），具体可以查看SDK源码确认属性名。
• 结合Temporal SDK工具
  Temporal Python SDK本身没有自动扫描功能，但可以将上述遍历逻辑和Worker注册逻辑结合，实现自动注册。

问题场景 #

用带参数的装饰器@kafka_listener(topic="foo")修饰函数consume_foo后，通过inspect查看函数成员，过滤掉__doc__、__globals__、__builtins__后，看不到topic参数或装饰器的任何标识。

原因 #

你的装饰器实现没有给被修饰的函数绑定任何标记。带参数的装饰器本质是返回装饰器的高阶函数，如果内部没有把装饰器参数或标识附加到被装饰函数上，inspect自然无法查到。

解决方案 #

修改装饰器实现，将装饰器参数和标识绑定到被装饰函数的自定义属性上，同时用functools.wraps保留原函数的元信息：

import functools

def kafka_listener(topic: str):
    def decorator(func):
        # 添加自定义属性存储装饰器参数和标识
        func._kafka_listener_topic = topic
        func._is_kafka_listener = True

        # 用functools.wraps保留原函数的名称、文档等元信息
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 这里可以添加装饰器的逻辑，比如预处理消息
            return func(*args, **kwargs)
        return wrapper
    return decorator

修改后，就可以通过以下方式获取信息：

# 检查是否被kafka_listener修饰
if hasattr(consume_foo, "_is_kafka_listener"):
    # 获取topic参数
    print(consume_foo._kafka_listener_topic)  # 输出: foo

内容的提问来源于stack exchange，提问作者Archimedes Trajano