Python 本身不提供运行时反射式查找“被某装饰器修饰的类”的内置机制，但可通过装饰器自维护注册表实现类似 Spring Boot @ComponentScan 的自动发现能力。核心思路是让装饰器在修饰时主动记录目标对象。
python 本身不提供运行时反射式查找“被某装饰器修饰的类”的内置机制，但可通过装饰器自维护注册表实现类似 spring boot `@componentscan` 的自动发现能力。核心思路是让装饰器在修饰时主动记录目标对象。
在 Python 中，装饰器本质上是函数调用（如 @workflow.defn 等价于 workflow.defn(cls)），而非元数据标签。这意味着：装饰器执行后，原始对象可能已被替换（如返回 wrapper），且标准 inspect 模块无法回溯“谁被哪个装饰器处理过”——正如你在 kafka_listener 函数中观察到的：__dict__、__annotations__ 等均不保留装饰器参数信息，因为装饰器逻辑若未显式保存上下文，该信息即告丢失。
✅ 正确解法不是“事后扫描”，而是事前注册：改造装饰器，使其在应用时主动将被修饰对象（类或函数）添加到一个全局或模块级注册表中。
以下是一个生产就绪的注册式装饰器实现示例：
# registry.py
from typing import List, Type, Callable, Any
# 全局注册表：按装饰器类型分组
_REGISTERED_CLASSES: dict[str, List[Type]] = {}
_REGISTERED_FUNCTIONS: dict[str, List[Callable]] = {}
def register_by_name(name: str):
"""工厂装饰器：为指定名称创建可注册的类装饰器"""
def decorator(cls: Type) -> Type:
if name not in _REGISTERED_CLASSES:
_REGISTERED_CLASSES[name] = []
_REGISTERED_CLASSES[name].append(cls)
return cls
return decorator
def get_registered_classes(name: str) -> List[Type]:
"""获取指定名称下所有已注册的类"""
return _REGISTERED_CLASSES.get(name, []).copy()
# 使用示例
@register_by_name("workflow.defn")
class PaymentWorkflow:
pass
@register_by_name("activity.defn")
class ValidateCardActivity:
pass
# 启动时自动注册到 Temporal Worker
if __name__ == "__main__":
workflows = get_registered_classes("workflow.defn")
activities = get_registered_classes("activity.defn")
print(f"Found workflows: {[w.__name__ for w in workflows]}") # ['PaymentWorkflow']
print(f"Found activities: {[a.__name__ for a in activities]}") # ['ValidateCardActivity']? 对函数装饰器（如 @kafka_listener(topic="foo")）同样适用：
立即学习“Python免费学习笔记（深入）”；
def kafka_listener(topic: str):
def decorator(func: Callable) -> Callable:
# 注册函数及其配置
key = f"kafka_listener.{topic}"
if key not in _REGISTERED_FUNCTIONS:
_REGISTERED_FUNCTIONS[key] = []
_REGISTERED_FUNCTIONS[key].append(func)
return func
return decorator
@kafka_listener(topic="user_events")
def handle_user_event(record):
print("Handling:", record)⚠️ 注意事项：
- 装饰器必须是“有状态”的：需显式维护注册列表（推荐使用模块级变量或单例管理器，避免闭包导致的多实例隔离问题）；
-
导入顺序影响注册时机：确保所有被装饰的模块在调用
get_registered_*()前已被导入（可借助pkgutil.iter_modules自动导入子包）； -
避免重复注册：可在装饰器内增加
if cls not in lst:判断（注意is或==语义）； -
不适用于第三方装饰器：若无法修改
workflow.defn源码，则无法注入注册逻辑——此时需结合 AST 静态分析（复杂，不推荐用于常规场景）。
总结：Python 的装饰器是“主动行为”而非“被动标记”。要实现自动发现，关键在于让装饰器承担注册职责，而非依赖运行时反射。这种模式简洁、高效、可控，也是 FastAPI、SQLModel、Temporal Python SDK 等主流框架的实际采用方案。
