You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/11/18 15:30:29 UTC
[GitHub] [airflow] davidavdav opened a new pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
davidavdav opened a new pull request #19683:
URL: https://github.com/apache/airflow/pull/19683
This small PR uses the type of a `BaseOperator` task, rather than its property `task_type`, to detect DAG dependence.
This helps in situations like described in #19582, where an operator is derived from `ExternalTaskSensor`. Using the `task_type` prevents such a sensor from being detected as a DAG dependence. This PR will allow automatic detection of derived classes.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#issuecomment-1038487475
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] closed pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #19683:
URL: https://github.com/apache/airflow/pull/19683
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#issuecomment-1038487475
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] davidavdav commented on pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
davidavdav commented on pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#issuecomment-976865653
Is that purely formatting / style? I should be able to fix that (but so should the automatic checking I suppose?).
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a change in pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
dstandish commented on a change in pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#discussion_r763222713
##########
File path: tests/serialization/test_dag_serialization.py
##########
@@ -1233,6 +1233,66 @@ def test_deps_sorted(self):
'airflow.ti_deps.deps.trigger_rule_dep.TriggerRuleDep',
]
+ def test_derived_dag_deps(self):
+ """
+ Tests DAG dependency detection, including derived classes
+ """
+ from airflow.operators.dummy import DummyOperator
+ from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+ from airflow.sensors.external_task import ExternalTaskSensor
+
+ class DerivedSensor(ExternalTaskSensor):
+ template_fields = ['external_dag_id', 'run_id']
+
+ def __init__(self, *, external_dag_id: str, run_id: str = None, **kwargs):
+ super().__init__(
+ external_dag_id=external_dag_id, execution_delta=None, execution_date_fn=None, **kwargs
+ )
+ self.run_id = run_id
Review comment:
```suggestion
class DerivedSensor(ExternalTaskSensor):
pass
```
would this not accoplish the same thing? is there a reason we need to add the other stuff for this test?
##########
File path: tests/serialization/test_dag_serialization.py
##########
@@ -1233,6 +1233,66 @@ def test_deps_sorted(self):
'airflow.ti_deps.deps.trigger_rule_dep.TriggerRuleDep',
]
+ def test_derived_dag_deps(self):
+ """
+ Tests DAG dependency detection, including derived classes
+ """
+ from airflow.operators.dummy import DummyOperator
+ from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+ from airflow.sensors.external_task import ExternalTaskSensor
+
+ class DerivedSensor(ExternalTaskSensor):
+ template_fields = ['external_dag_id', 'run_id']
+
+ def __init__(self, *, external_dag_id: str, run_id: str = None, **kwargs):
+ super().__init__(
+ external_dag_id=external_dag_id, execution_delta=None, execution_date_fn=None, **kwargs
+ )
+ self.run_id = run_id
+
+ execution_date = datetime(2020, 1, 1)
+ for class_ in [ExternalTaskSensor, DerivedSensor]:
+ with DAG(dag_id="test_derived_dag_deps_sensor", start_date=execution_date) as dag:
+ task1 = class_(
+ task_id="task1",
+ external_dag_id="external_dag_id",
+ mode="reschedule",
+ )
+ task2 = DummyOperator(task_id="task2")
+ task1 >> task2
+
+ dag = SerializedDAG.to_dict(dag)
+ assert dag['dag']['dag_dependencies'] == [
+ {
+ 'source': 'external_dag_id',
+ 'target': 'test_derived_dag_deps_sensor',
+ 'dependency_type': 'sensor',
+ 'dependency_id': 'task1',
+ }
+ ]
+
+ class DerivedOperator(TriggerDagRunOperator):
Review comment:
could you please split this test into two tests:
`test_derived_dag_deps_sensor` and `test_derived_dag_deps_operator`?
there's not really any shared logic or interaction, so separating would help readability
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a change in pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
dstandish commented on a change in pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#discussion_r758624943
##########
File path: tests/serialization/test_dag_serialization.py
##########
@@ -1233,6 +1233,67 @@ def test_deps_sorted(self):
'airflow.ti_deps.deps.trigger_rule_dep.TriggerRuleDep',
]
+ def test_derived_dag_deps(self):
+ """
+ Tests DAG dependency detection, including derived classes
+ """
+ from airflow.operators.dummy import DummyOperator
+ from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+ from airflow.sensors.external_task import ExternalTaskSensor
+
+ class DerivedSensor(ExternalTaskSensor):
+ template_fields = ['external_dag_id', 'run_id']
+
+ def __init__(self, *, external_dag_id: str, run_id: str = None, **kwargs):
+ super().__init__(
+ external_dag_id=external_dag_id, execution_delta=None, execution_date_fn=None, **kwargs
+ )
+ self.run_id = run_id
+
+ execution_date = datetime(2020, 1, 1)
+ for Sensor in [ExternalTaskSensor, DerivedSensor]:
+ with DAG(dag_id="test_derived_dag_deps_sensor", start_date=execution_date) as dag:
+ task1 = Sensor(
+ task_id="task1",
+ external_dag_id="external_dag_id",
+ mode="reschedule",
+ )
+ task2 = DummyOperator(task_id="task2")
+ task1 >> task2
+
+ dag = SerializedDAG.to_dict(dag)
+ assert dag['dag']['dag_dependencies'] == [
+ {
+ 'source': 'external_dag_id',
+ 'target': 'test_derived_dag_deps_sensor',
+ 'dependency_type': 'sensor',
+ 'dependency_id': 'task1',
+ }
+ ]
+
+ class DerivedTrigger(TriggerDagRunOperator):
Review comment:
please call this `DerivedOperator` instead of `DerivedTrigger`
`Trigger` is a _thing_ in airflow so best not to use it here (could be confusing, and produce false positives in search)
```suggestion
class DerivedOperator(TriggerDagRunOperator):
```
##########
File path: tests/serialization/test_dag_serialization.py
##########
@@ -1233,6 +1233,67 @@ def test_deps_sorted(self):
'airflow.ti_deps.deps.trigger_rule_dep.TriggerRuleDep',
]
+ def test_derived_dag_deps(self):
+ """
+ Tests DAG dependency detection, including derived classes
+ """
+ from airflow.operators.dummy import DummyOperator
+ from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+ from airflow.sensors.external_task import ExternalTaskSensor
+
+ class DerivedSensor(ExternalTaskSensor):
+ template_fields = ['external_dag_id', 'run_id']
+
+ def __init__(self, *, external_dag_id: str, run_id: str = None, **kwargs):
+ super().__init__(
+ external_dag_id=external_dag_id, execution_delta=None, execution_date_fn=None, **kwargs
+ )
+ self.run_id = run_id
+
+ execution_date = datetime(2020, 1, 1)
+ for Sensor in [ExternalTaskSensor, DerivedSensor]:
Review comment:
```suggestion
for class_ in [ExternalTaskSensor, DerivedSensor]:
```
##########
File path: tests/serialization/test_dag_serialization.py
##########
@@ -1233,6 +1233,67 @@ def test_deps_sorted(self):
'airflow.ti_deps.deps.trigger_rule_dep.TriggerRuleDep',
]
+ def test_derived_dag_deps(self):
+ """
+ Tests DAG dependency detection, including derived classes
+ """
+ from airflow.operators.dummy import DummyOperator
+ from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+ from airflow.sensors.external_task import ExternalTaskSensor
+
+ class DerivedSensor(ExternalTaskSensor):
+ template_fields = ['external_dag_id', 'run_id']
+
+ def __init__(self, *, external_dag_id: str, run_id: str = None, **kwargs):
+ super().__init__(
+ external_dag_id=external_dag_id, execution_delta=None, execution_date_fn=None, **kwargs
+ )
+ self.run_id = run_id
+
+ execution_date = datetime(2020, 1, 1)
+ for Sensor in [ExternalTaskSensor, DerivedSensor]:
+ with DAG(dag_id="test_derived_dag_deps_sensor", start_date=execution_date) as dag:
+ task1 = Sensor(
+ task_id="task1",
+ external_dag_id="external_dag_id",
+ mode="reschedule",
+ )
+ task2 = DummyOperator(task_id="task2")
+ task1 >> task2
+
+ dag = SerializedDAG.to_dict(dag)
+ assert dag['dag']['dag_dependencies'] == [
+ {
+ 'source': 'external_dag_id',
+ 'target': 'test_derived_dag_deps_sensor',
+ 'dependency_type': 'sensor',
+ 'dependency_id': 'task1',
+ }
+ ]
+
+ class DerivedTrigger(TriggerDagRunOperator):
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ for Trigger in [TriggerDagRunOperator, DerivedTrigger]:
Review comment:
```suggestion
for class_ in [TriggerDagRunOperator, DerivedTrigger]:
```
##########
File path: airflow/serialization/serialized_objects.py
##########
@@ -444,14 +446,14 @@ class DependencyDetector:
@staticmethod
def detect_task_dependencies(task: BaseOperator) -> Optional['DagDependency']:
"""Detects dependencies caused by tasks"""
- if task.task_type == "TriggerDagRunOperator":
+ if isinstance(task, TriggerDagRunOperator):
Review comment:
@kaxil do you think that perhaps there should be an attribute on task sensor (like a class attribute that is set on it, and which could also be set in other implementations) that we look at (instead of the class itself) so that there could be other implementations of task-sensor-like operators which do not inherit from external task sensor but also have dag dependencies?
so that it needn't necessarily be a subclass. maybe it's a bit too magical, and just thinking aloud, but perhaps we could look at the arguments of the operator to infer whether DagDependency should be created?
##########
File path: tests/serialization/test_dag_serialization.py
##########
@@ -1233,6 +1233,67 @@ def test_deps_sorted(self):
'airflow.ti_deps.deps.trigger_rule_dep.TriggerRuleDep',
]
+ def test_derived_dag_deps(self):
+ """
+ Tests DAG dependency detection, including derived classes
+ """
+ from airflow.operators.dummy import DummyOperator
+ from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+ from airflow.sensors.external_task import ExternalTaskSensor
+
+ class DerivedSensor(ExternalTaskSensor):
+ template_fields = ['external_dag_id', 'run_id']
+
+ def __init__(self, *, external_dag_id: str, run_id: str = None, **kwargs):
+ super().__init__(
+ external_dag_id=external_dag_id, execution_delta=None, execution_date_fn=None, **kwargs
+ )
+ self.run_id = run_id
+
+ execution_date = datetime(2020, 1, 1)
+ for Sensor in [ExternalTaskSensor, DerivedSensor]:
+ with DAG(dag_id="test_derived_dag_deps_sensor", start_date=execution_date) as dag:
+ task1 = Sensor(
+ task_id="task1",
+ external_dag_id="external_dag_id",
+ mode="reschedule",
+ )
+ task2 = DummyOperator(task_id="task2")
+ task1 >> task2
+
+ dag = SerializedDAG.to_dict(dag)
+ assert dag['dag']['dag_dependencies'] == [
+ {
+ 'source': 'external_dag_id',
+ 'target': 'test_derived_dag_deps_sensor',
+ 'dependency_type': 'sensor',
+ 'dependency_id': 'task1',
+ }
+ ]
+
+ class DerivedTrigger(TriggerDagRunOperator):
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
Review comment:
you can omit `__init__` when you do not change it
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#issuecomment-972973773
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/BREEZE.rst) for testing locally, itโs a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#coding-style-and-best-practices).
Apache Airflow is a community-driven project and together we are making it better ๐.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on a change in pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#discussion_r752362440
##########
File path: airflow/serialization/serialized_objects.py
##########
@@ -444,14 +446,14 @@ class DependencyDetector:
@staticmethod
def detect_task_dependencies(task: BaseOperator) -> Optional['DagDependency']:
"""Detects dependencies caused by tasks"""
- if task.task_type == "TriggerDagRunOperator":
+ if isinstance(task, TriggerDagRunOperator):
Review comment:
Since this is done during serialization time this should be ok, however can you please add test(s)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] davidavdav commented on a change in pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
davidavdav commented on a change in pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#discussion_r752533080
##########
File path: airflow/serialization/serialized_objects.py
##########
@@ -444,14 +446,14 @@ class DependencyDetector:
@staticmethod
def detect_task_dependencies(task: BaseOperator) -> Optional['DagDependency']:
"""Detects dependencies caused by tasks"""
- if task.task_type == "TriggerDagRunOperator":
+ if isinstance(task, TriggerDagRunOperator):
Review comment:
Sure I'll look into how tests work in airflow, and add one, I suppose, for testing that both `TriggerDagRunOperator` and inherited classes give the expected result.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] davidavdav commented on pull request #19683: In DAG dependency detector, use class type instead of class name, related #19582
Posted by GitBox <gi...@apache.org>.
davidavdav commented on pull request #19683:
URL: https://github.com/apache/airflow/pull/19683#issuecomment-1002997948
bump
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org