[airflow] @task 속성에 대해서 알아보자

 

 

def _run_task_by_selected_method(args, dag: DAG, ti: TaskInstance) -> None | TaskReturnCode:
    """
    Run the task based on a mode.

    Any of the 3 modes are available:

    - using LocalTaskJob
    - as raw task
    - by executor
    """
    if args.local:
        return _run_task_by_local_task_job(args, ti)
    if args.raw:
        return _run_raw_task(args, ti)
    _run_task_by_executor(args, dag, ti)
    return None

 

 

 

class TaskDecoratorCollection:
    """Implementation to provide the ``@task`` syntax."""

    python = staticmethod(python_task)
    virtualenv = staticmethod(virtualenv_task)
    external_python = staticmethod(external_python_task)
    branch = staticmethod(branch_task)
    branch_virtualenv = staticmethod(branch_virtualenv_task)
    branch_external_python = staticmethod(branch_external_python_task)
    short_circuit = staticmethod(short_circuit_task)
    sensor = staticmethod(sensor_task)
    bash = staticmethod(bash_task)
    run_if = staticmethod(run_if)
    skip_if = staticmethod(skip_if)

    __call__: Any = python  # Alias '@task' to '@task.python'.

    def __getattr__(self, name: str) -> TaskDecorator:
        """Dynamically get provider-registered task decorators, e.g. ``@task.docker``."""
        if name.startswith("__"):
            raise AttributeError(f"{type(self).__name__} has no attribute {name!r}")
        decorators = ProvidersManager().taskflow_decorators
        if name not in decorators:
            raise AttributeError(f"task decorator {name!r} not found")
        return decorators[name]


task = TaskDecoratorCollection()
setup: Callable = setup_task
teardown: Callable = teardown_task