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
Saves the lineage to XCom and if configured to do so sends it
    to the backend.