Spaces:
Sleeping
Sleeping
| """Auto-discovering task registry: any module here exposing `TASK` is registered.""" | |
| import importlib | |
| import pkgutil | |
| from app.tasks.base import Task | |
| REGISTRY: dict[str, Task] = {} | |
| for _mod_info in pkgutil.iter_modules(__path__): | |
| if _mod_info.name == "base": | |
| continue | |
| _mod = importlib.import_module(f"{__name__}.{_mod_info.name}") | |
| _task = getattr(_mod, "TASK", None) | |
| if _task is not None: | |
| REGISTRY[_task.id] = _task | |
| REGISTRY = dict(sorted(REGISTRY.items(), key=lambda kv: kv[1].order)) | |