airflow-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] [airflow] kaxil commented on a change in pull request #6564: [AIRFLOW-5911] Simplify lineage API and improve robustness
Date Mon, 25 Nov 2019 19:42:04 GMT
kaxil commented on a change in pull request #6564: [AIRFLOW-5911] Simplify lineage API and
improve robustness
URL: https://github.com/apache/airflow/pull/6564#discussion_r350384356
 
 

 ##########
 File path: airflow/lineage/__init__.py
 ##########
 @@ -92,52 +120,49 @@ def prepare_lineage(func):
     * "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
+    * "list of datasets" -> manually defined list of data
 
     """
     @wraps(func)
     def wrapper(self, context, *args, **kwargs):
         self.log.debug("Preparing lineage inlets and outlets")
 
-        task_ids = set(self._inlets['task_ids']).intersection(  # pylint:disable=protected-access
-            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']:  # pylint:disable=protected-access
-            # dont append twice
-            task_ids = set(self._inlets['task_ids']).symmetric_difference(  # pylint:disable=protected-access
-                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 self._inlets['datasets']:  # pylint:disable=protected-access
-            self.inlets.extend(self._inlets['datasets'])  # pylint:disable=protected-access
-
-        # outlets
-        if self._outlets['datasets']:  # pylint:disable=protected-access
-            self.outlets.extend(self._outlets['datasets'])  # pylint:disable=protected-access
+        if isinstance(self._inlets, (str, Operator)) or attr.has(self._inlets):         
   # pylint:disable=protected-access, line-too-long # noqa: E501
+            self._inlets = [self._inlets, ]                                             
   # pylint:disable=protected-access, line-too-long # noqa: E501
 
-        self.log.debug("inlets: %s, outlets: %s", self.inlets, self.outlets)
+        if self._inlets and isinstance(self._inlets, list):                             
   # pylint:disable=protected-access, line-too-long # noqa: E501
 
 Review comment:
   Can we add a comment on what is happening in this if condition with all the `union` and
`intersection` stuff.
   
   We should also try and get rid of `line-too-long` if possible as they are causing noise
between code (atleast in the Github diff :D ) 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message