File size: 522 Bytes
910dadd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
"""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))