airflow/models/dag.py
def dag(*dag_args, **dag_kwargs):
    """
    Python dag decorator. Wraps a function into an Airflow DAG.
    Accepts kwargs for operator kwarg. Can be used to parametrize DAGs.

    :param dag_args: Arguments for DAG object
    :type dag_args: Any
    :param dag_kwargs: Kwargs for DAG object.
    :type dag_kwargs: Any
    """

    def wrapper(f: Callable):
        # Get dag initializer signature and bind it to validate that dag_args, and dag_kwargs are correct
        dag_sig = signature(DAG.__init__)
        dag_bound_args = dag_sig.bind_partial(*dag_args, **dag_kwargs)

        @functools.wraps(f)
        def factory(*args, **kwargs):
            # Generate signature for decorated function and bind the arguments when called
            # we do this to extract parameters so we can annotate them on the DAG object.
            # In addition, this fails if we are missing any args/kwargs with TypeError as expected.
            f_sig = signature(f).bind(*args, **kwargs)
            # Apply defaults to capture default values if set.
            f_sig.apply_defaults()

            # Set function name as dag_id if not set
            dag_id = dag_bound_args.arguments.get('dag_id', f.__name__)
            dag_bound_args.arguments['dag_id'] = dag_id

            # Initialize DAG with bound arguments
            with DAG(*dag_bound_args.args, **dag_bound_args.kwargs) as dag_obj:
                # Set DAG documentation from function documentation.
                if f.__doc__:
                    dag_obj.doc_md = f.__doc__

                # Generate DAGParam for each function arg/kwarg and replace it for calling the function.
                # All args/kwargs for function will be DAGParam object and replaced on execution time.
                f_kwargs = {}
                for name, value in f_sig.arguments.items():
                    f_kwargs[name] = dag_obj.param(name, value)

                # set file location to caller source path
                back = sys._getframe().f_back
                dag_obj.fileloc = back.f_code.co_filename if back else ""

                # Invoke function to create operators in the DAG scope.
                f(**f_kwargs)

            # Return dag object such that it's accessible in Globals.
            return dag_obj

        return factory

    return wrapper

    Python dag decorator. Wraps a function into an Airflow DAG.
    Accepts kwargs for operator kwarg. Can be used to parametrize DAGs.

    :param dag_args: Arguments for DAG object
    :type dag_args: Any
    :param dag_kwargs: Kwargs for DAG object.
    :type dag_kwargs: Any
    
Python dag decorator. Wraps a function into an Airflow DAG.
Accepts kwargs for operator kwarg. Can be used to parametrize DAGs.
Python dag decorator . Wraps a function into an Airflow DAG . Accepts kwargs for operator kwarg . Can be used to parametrize DAGs .
def dag(*dag_args, **dag_kwargs):
    def wrapper(f: Callable):
        dag_sig = signature(DAG.__init__)
        dag_bound_args = dag_sig.bind_partial(*dag_args, **dag_kwargs)
        @functools.wraps(f)
        def factory(*args, **kwargs):
            f_sig = signature(f).bind(*args, **kwargs)
            f_sig.apply_defaults()
            dag_id = dag_bound_args.arguments.get('dag_id', f.__name__)
            dag_bound_args.arguments['dag_id'] = dag_id
            with DAG(*dag_bound_args.args, **dag_bound_args.kwargs) as dag_obj:
                if f.__doc__:
                    dag_obj.doc_md = f.__doc__
                f_kwargs = {}
                for name, value in f_sig.arguments.items():
                    f_kwargs[name] = dag_obj.param(name, value)
                back = sys._getframe().f_back
                dag_obj.fileloc = back.f_code.co_filename if back else ""
                f(**f_kwargs)
            return dag_obj
        return factory
    return wrapper
def dag ( * dag_args , ** dag_kwargs ) : def wrapper ( f : Callable ) : dag_sig = signature ( DAG . __init__ ) dag_bound_args = dag_sig . bind_partial ( * dag_args , ** dag_kwargs ) @ functools . wraps ( f ) def factory ( * args , ** kwargs ) : f_sig = signature ( f ) . bind ( * args , ** kwargs ) f_sig . apply_defaults ( ) dag_id = dag_bound_args . arguments . get ( 'dag_id' , f . __name__ ) dag_bound_args . arguments [ 'dag_id' ] = dag_id with DAG ( * dag_bound_args . args , ** dag_bound_args . kwargs ) as dag_obj : if f . __doc__ : dag_obj . doc_md = f . __doc__ f_kwargs = { } for name , value in f_sig . arguments . items ( ) : f_kwargs [ name ] = dag_obj . param ( name , value ) back = sys . _getframe ( ) . f_back dag_obj . fileloc = back . f_code . co_filename if back else "" f ( ** f_kwargs ) return dag_obj return factory return wrapper
Python dag decorator.
Python dag decorator .
"""
    Python dag decorator. Wraps a function into an Airflow DAG.
    Accepts kwargs for operator kwarg. Can be used to parametrize DAGs.

    :param dag_args: Arguments for DAG object
    :type dag_args: Any
    :param dag_kwargs: Kwargs for DAG object.
    :type dag_kwargs: Any
    """ # Get dag initializer signature and bind it to validate that dag_args, and dag_kwargs are correct # Generate signature for decorated function and bind the arguments when called # we do this to extract parameters so we can annotate them on the DAG object. # In addition, this fails if we are missing any args/kwargs with TypeError as expected. # Apply defaults to capture default values if set. # Set function name as dag_id if not set # Initialize DAG with bound arguments # Set DAG documentation from function documentation. # Generate DAGParam for each function arg/kwarg and replace it for calling the function. # All args/kwargs for function will be DAGParam object and replaced on execution time. # set file location to caller source path # Invoke function to create operators in the DAG scope. # Return dag object such that it's accessible in Globals.
dag_args
Arguments for DAG object
Arguments for DAG object
dag_kwargs
Kwargs for DAG object.
Kwargs for DAG object .