airflow/lineage/__init__.py
def apply_lineage(func):
    """
    Saves the lineage to XCom and if configured to do so sends it
    to the backend.
    """
    backend = _get_backend()

    @wraps(func)
    def wrapper(self, context, *args, **kwargs):
        self.log.debug("Backend: %s, Lineage called with inlets: %s, outlets: %s",
                       backend, self.inlets, self.outlets)
        ret_val = func(self, context, *args, **kwargs)

        outlets = [x.as_dict() for x in self.outlets]
        inlets = [x.as_dict() for x in self.inlets]

        if len(self.outlets) > 0:
            self.xcom_push(context,
                           key=PIPELINE_OUTLETS,
                           value=outlets,
                           execution_date=context['ti'].execution_date)

        if len(self.inlets) > 0:
            self.xcom_push(context,
                           key=PIPELINE_INLETS,
                           value=inlets,
                           execution_date=context['ti'].execution_date)

        if backend:
            backend.send_lineage(operator=self, inlets=self.inlets,
                                 outlets=self.outlets, context=context)

        return ret_val

    return wrapper
def apply_lineage(func):
    """
    Saves the lineage to XCom and if configured to do so sends it
    to the backend.
    """
    backend = _get_backend()

    @wraps(func)
    def wrapper(self, context, *args, **kwargs):
        self.log.debug("Backend: %s, Lineage called with inlets: %s, outlets: %s",
                       backend, self.inlets, self.outlets)
        ret_val = func(self, context, *args, **kwargs)

        outlets = [x.as_dict() for x in self.outlets]
        inlets = [x.as_dict() for x in self.inlets]

        if len(self.outlets) > 0:
            self.xcom_push(context,
                           key=PIPELINE_OUTLETS,
                           value=outlets,
                           execution_date=context['ti'].execution_date)

        if len(self.inlets) > 0:
            self.xcom_push(context,
                           key=PIPELINE_INLETS,
                           value=inlets,
                           execution_date=context['ti'].execution_date)

        if backend:
            backend.send_lineage(operator=self, inlets=self.inlets,
                                 outlets=self.outlets, context=context)

        return ret_val

    return wrapper
def apply_lineage ( func ) : backend = _get_backend ( ) @ wraps ( func ) def wrapper ( self , context , * args , * * kwargs ) : self . log . debug ( "Backend: %s, Lineage called with inlets: %s, outlets: %s" , backend , self . inlets , self . outlets ) ret_val = func ( self , context , * args , * * kwargs ) outlets = [ x . as_dict ( ) for x in self . outlets ] inlets = [ x . as_dict ( ) for x in self . inlets ] if len ( self . outlets ) > 0 : self . xcom_push ( context , key = PIPELINE_OUTLETS , value = outlets , execution_date = context [ 'ti' ] . execution_date ) if len ( self . inlets ) > 0 : self . xcom_push ( context , key = PIPELINE_INLETS , value = inlets , execution_date = context [ 'ti' ] . execution_date ) if backend : backend . send_lineage ( operator = self , inlets = self . inlets , outlets = self . outlets , context = context ) return ret_val return wrapper
Saves the lineage to XCom and if configured to do so sends it
    to the backend.
Saves the lineage to XCom and if configured to do so sends it to the backend .