fixup! fixup! Optimize subclasses of DummyOperator for Scheduling

kaxil · kaxil · commit 0b47a1b9d251 · 2020-12-01T20:32:03.000Z
diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
@@ -1371,6 +1371,11 @@ def is_smart_sensor_compatible(self):
         """Return if this operator can use smart service. Default False."""
         return False
 
+    @property
+    def inherits_from_dummy_operator(self):
+        """Used to determine if an Operator is inherited from DummyOperator"""
+        return getattr(self, '_is_dummy', False)
+
 
 def chain(*tasks: Union[BaseOperator, Sequence[BaseOperator]]):
     r"""
diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
@@ -724,7 +724,7 @@ def schedule_tis(self, schedulable_tis: Iterable[TI], session: Session = None) -
             ti
             for ti in schedulable_tis
             if (
-                (ti.task.task_type == "DummyOperator" or getattr(ti.task, "_is_dummy", False))
+                ti.task.inherits_from_dummy_operator
                 and not ti.task.on_execute_callback
                 and not ti.task.on_success_callback
             )
diff --git a/airflow/operators/dummy_operator.py b/airflow/operators/dummy_operator.py
@@ -29,6 +29,7 @@ class DummyOperator(BaseOperator):
     """
 
     ui_color = '#e8f7e4'
+    inherits_from_dummy_operator = True
 
     @apply_defaults
     def __init__(self, **kwargs) -> None:
diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
@@ -31,7 +31,6 @@
 from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
 from airflow.models.connection import Connection
 from airflow.models.dag import DAG
-from airflow.operators.dummy_operator import DummyOperator
 from airflow.serialization.enums import DagAttributeTypes as DAT, Encoding
 from airflow.serialization.helpers import serialize_template_field
 from airflow.serialization.json_schema import Validator, load_dag_schema
@@ -353,7 +352,7 @@ def serialize_operator(cls, op: BaseOperator) -> dict:
         serialize_op['_task_module'] = op.__class__.__module__
 
         # Used to determine if an Operator is inherited from DummyOperator
-        serialize_op['_is_dummy'] = cls._is_inherited_from_dummy_operator(op)
+        serialize_op['_is_dummy'] = op.inherits_from_dummy_operator
 
         if op.operator_extra_links:
             serialize_op['_operator_extra_links'] = cls._serialize_operator_extra_links(
@@ -540,13 +539,6 @@ def _serialize_operator_extra_links(cls, operator_extra_links: Iterable[BaseOper
 
         return serialize_operator_extra_links
 
-    @classmethod
-    def _is_inherited_from_dummy_operator(cls, op: BaseOperator) -> bool:
-        """Used to determine if an Operator is inherited from DummyOperator"""
-        if op.task_type == "DummyOperator" or isinstance(op, DummyOperator):
-            return True
-        return False
-
 
 class SerializedDAG(DAG, BaseSerialization):
     """

Original file line number	Diff line number	Diff line change
`@@ -724,7 +724,7 @@ def schedule_tis(self, schedulable_tis: Iterable[TI], session: Session = None) -`
`724`	`724`	`ti`
`725`	`725`	`for ti in schedulable_tis`
`726`	`726`	`if (`
`727`		`- (ti.task.task_type == "DummyOperator" or getattr(ti.task, "_is_dummy", False))`
	`727`	`+ ti.task.inherits_from_dummy_operator`
`728`	`728`	`and not ti.task.on_execute_callback`
`729`	`729`	`and not ti.task.on_success_callback`
`730`	`730`	`)`