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
def prepare_lineage ( func ) : @ 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
Prepares the lineage inlets and outlets . Inlets can be :