from typing import Dict, List
import pendulum
from airflow.sdk import BaseOperator, dag, task, task_group
@dag(
    schedule=None,
    start_date=pendulum.datetime(2023, 1, 1),
)
def test_dag():
    _all_tasks: Dict[str, List[BaseOperator]] = {}
    def _generate_task_group_callable(group_id, tasks):
        @task_group(group_id=group_id)
        def _task_group():
            [t() for t in tasks]
        return _task_group
    def _generate_task_callable(task_id, task_set):
        def _generate():
            @task(task_id=task_id)
            def _task():
                pass
            if _all_tasks.get(task_set) is None:
                _all_tasks[task_set] = []
            _all_tasks[task_set].append(_task())
        return _generate
    _set_callables = []
    for _set in range(0, 2):
        _set_id = f"set_{_set}"
        _group_callables_a = []
        for _nga in range(0, 1):
            _nga_id = f"group_{_nga}"
            _group_callables_b = []
            for _ngb in range(0, 3):
                _ngb_id = f"group_{_ngb}"
                _task_callables = []
                for _task in range(0, 10):
                    _task_id = f"task_{_task}"
                    _callable = _generate_task_callable(_task_id, _set_id)
                    _task_callables.append(_callable)
                _group_callable_inner = _generate_task_group_callable(_ngb_id, _task_callables)
                _group_callables_b.append(_group_callable_inner)
            _group_callable_outer = _generate_task_group_callable(_nga_id, _group_callables_b)
            _group_callables_a.append(_group_callable_outer)
        _set_callable = _generate_task_group_callable(_set_id, _group_callables_a)
        _set_callables.append(_set_callable)
    [_s() for _s in _set_callables]
    for i in range(0, len(_all_tasks["set_0"])):
        _all_tasks["set_0"][i] >> _all_tasks["set_1"][i]
test_dag()