| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051 |
- from functools import wraps
- from typing import Dict, Type, List
- # 存储所有注册的任务类
- registered_tasks: Dict[str, Type] = {}
- def register_task(task_name: str):
- """
- 任务注册装饰器,用于将任务类注册到系统中。
-
- Args:
- task_name: 任务的唯一标识名称
-
- Returns:
- 装饰器函数
- """
- def decorator(task_class: Type):
- @wraps(task_class)
- def wrapper(*args, **kwargs):
- return task_class(*args, **kwargs)
-
- # 将任务类注册到字典中
- registered_tasks[task_name] = task_class
- return wrapper
- return decorator
- def get_registered_tasks() -> Dict[str, Type]:
- """
- 获取所有注册的任务类。
-
- Returns:
- Dict[str, Type]: 包含所有注册任务的字典,键为任务名称,值为任务类
- """
- return registered_tasks
- def get_task_class(task_name: str) -> Type:
- """
- 根据任务名称获取对应的任务类。
-
- Args:
- task_name: 任务的唯一标识名称
-
- Returns:
- Type: 任务类
-
- Raises:
- KeyError: 如果任务名称不存在
- """
- if task_name not in registered_tasks:
- raise KeyError(f"任务 '{task_name}' 未注册")
- return registered_tasks[task_name]
|