airflow/lineage/__init__.py
def prepare_lineage(func):
    """
    Prepares the lineage inlets and outlets. Inlets can be:

    * "auto" -> picks up any outlets from direct upstream tasks that have outlets defined, as such that
      if A -> B -> C and B does not have outlets but A does, these are provided as inlets.
    * "list of task_ids" -> picks up outlets from the upstream task_ids
    * "list of datasets" -> manually defined list of DataSet

    """
    @wraps(func)
    def wrapper(self, context, *args, **kwargs):
        self.log.debug("Preparing lineage inlets and outlets")

        task_ids = set(self._inlets['task_ids']).intersection(
            self.get_flat_relative_ids(upstream=True)
        )
        if task_ids:
            inlets = self.xcom_pull(context,
                                    task_ids=task_ids,
                                    dag_id=self.dag_id,
                                    key=PIPELINE_OUTLETS)
            inlets = [item for sublist in inlets if sublist for item in sublist]
            inlets = [DataSet.map_type(i['typeName'])(data=i['attributes'])
                      for i in inlets]
            self.inlets.extend(inlets)

        if self._inlets['auto']:
            # dont append twice
            task_ids = set(self._inlets['task_ids']).symmetric_difference(
                self.upstream_task_ids
            )
            inlets = self.xcom_pull(context,
                                    task_ids=task_ids,
                                    dag_id=self.dag_id,
                                    key=PIPELINE_OUTLETS)
            inlets = [item for sublist in inlets if sublist for item in sublist]
            inlets = [DataSet.map_type(i['typeName'])(data=i['attributes'])
                      for i in inlets]
            self.inlets.extend(inlets)

        if len(self._inlets['datasets']) > 0:
            self.inlets.extend(self._inlets['datasets'])

        # outlets
        if len(self._outlets['datasets']) > 0:
            self.outlets.extend(self._outlets['datasets'])

        self.log.debug("inlets: %s, outlets: %s", self.inlets, self.outlets)

        for dataset in chain(self.inlets, self.outlets):
            dataset.set_context(context)

        return func(self, context, *args, **kwargs)

    return wrapper
def prepare_lineage(func):
    """
    Prepares the lineage inlets and outlets. Inlets can be:

    * "auto" -> picks up any outlets from direct upstream tasks that have outlets defined, as such that
      if A -> B -> C and B does not have outlets but A does, these are provided as inlets.
    * "list of task_ids" -> picks up outlets from the upstream task_ids
    * "list of datasets" -> manually defined list of DataSet

    """
    @wraps(func)
    def wrapper(self, context, *args, **kwargs):
        self.log.debug("Preparing lineage inlets and outlets")

        task_ids = set(self._inlets['task_ids']).intersection(
            self.get_flat_relative_ids(upstream=True)
        )
        if task_ids:
            inlets = self.xcom_pull(context,
                                    task_ids=task_ids,
                                    dag_id=self.dag_id,
                                    key=PIPELINE_OUTLETS)
            inlets = [item for sublist in inlets if sublist for item in sublist]
            inlets = [DataSet.map_type(i['typeName'])(data=i['attributes'])
                      for i in inlets]
            self.inlets.extend(inlets)

        if self._inlets['auto']:
            # dont append twice
            task_ids = set(self._inlets['task_ids']).symmetric_difference(
                self.upstream_task_ids
            )
            inlets = self.xcom_pull(context,
                                    task_ids=task_ids,
                                    dag_id=self.dag_id,
                                    key=PIPELINE_OUTLETS)
            inlets = [item for sublist in inlets if sublist for item in sublist]
            inlets = [DataSet.map_type(i['typeName'])(data=i['attributes'])
                      for i in inlets]
            self.inlets.extend(inlets)

        if len(self._inlets['datasets']) > 0:
            self.inlets.extend(self._inlets['datasets'])

        # outlets
        if len(self._outlets['datasets']) > 0:
            self.outlets.extend(self._outlets['datasets'])

        self.log.debug("inlets: %s, outlets: %s", self.inlets, self.outlets)

        for dataset in chain(self.inlets, self.outlets):
            dataset.set_context(context)

        return func(self, context, *args, **kwargs)

    return wrapper
Prepares the lineage inlets and outlets. Inlets can be:

    * "auto" -> picks up any outlets from direct upstream tasks that have outlets defined, as such that
      if A -> B -> C and B does not have outlets but A does, these are provided as inlets.
    * "list of task_ids" -> picks up outlets from the upstream task_ids
    * "list of datasets" -> manually defined list of DataSet