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 2022/02/24 17:04:30 UTC
[GitHub] [airflow] ashb opened a new pull request #21798: WIP: Change BaseOperatorLink interface to take a ti_key, not a datetime
ashb opened a new pull request #21798:
URL: https://github.com/apache/airflow/pull/21798
More details to follow shortly.
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815922308
##########
File path: airflow/models/abstractoperator.py
##########
@@ -394,3 +405,7 @@ def _render_nested_template_fields(
# content has no inner template fields
return
self._do_render_template_fields(value, nested_template_fields, context, jinja_env, seen_oids)
+
+
+if TYPE_CHECKING:
+ from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
Review comment:
Done in a43f10836 -- it seemed to pass locally.
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
"""For an operator, gets the URLs that the ``extra_links`` entry points to.
:raise ValueError: The error message of a ValueError will be passed on through to
the fronted to show up as a tooltip on the disabled link.
- :param dttm: The datetime parsed execution date for the URL being searched for.
+ :param ti: The TaskInstance for the URL being searched for.
:param link_name: The name of the link we're looking for the URL for. Should be
one of the options specified in ``extra_links``.
"""
- if link_name in self.operator_extra_link_dict:
- return self.operator_extra_link_dict[link_name].get_link(self, dttm)
- elif link_name in self.global_operator_extra_link_dict:
- return self.global_operator_extra_link_dict[link_name].get_link(self, dttm)
+ link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
+ link_name
+ ) or self.global_operator_extra_link_dict.get(link_name)
Review comment:
Did the later in a43f10836
--
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] uranusjr commented on pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#issuecomment-1054738283
Would it be a good idea if `serialize_value` also take `ti_key` instead? cc @dstandish
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r814886878
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
"""For an operator, gets the URLs that the ``extra_links`` entry points to.
:raise ValueError: The error message of a ValueError will be passed on through to
the fronted to show up as a tooltip on the disabled link.
- :param dttm: The datetime parsed execution date for the URL being searched for.
+ :param ti: The TaskInstance for the URL being searched for.
:param link_name: The name of the link we're looking for the URL for. Should be
one of the options specified in ``extra_links``.
"""
- if link_name in self.operator_extra_link_dict:
- return self.operator_extra_link_dict[link_name].get_link(self, dttm)
- elif link_name in self.global_operator_extra_link_dict:
- return self.global_operator_extra_link_dict[link_name].get_link(self, dttm)
+ link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
+ link_name
+ ) or self.global_operator_extra_link_dict.get(link_name)
+ if not link:
+ return None
+ # Check for old function signature
+ parameters = inspect.signature(link.get_link).parameters
+ args = [name for name, p in parameters.items() if p.kind != p.VAR_KEYWORD]
+ if "ti_key" in args:
Review comment:
@uranusjr Do you think tis is enough of a check, or should we _force_ `ti_key` to be accepted only as a keyword argument?
Is it worth doing a `__init_subclass__` check on BaseOperatorLink to check the signature?
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
"""For an operator, gets the URLs that the ``extra_links`` entry points to.
:raise ValueError: The error message of a ValueError will be passed on through to
the fronted to show up as a tooltip on the disabled link.
- :param dttm: The datetime parsed execution date for the URL being searched for.
+ :param ti: The TaskInstance for the URL being searched for.
:param link_name: The name of the link we're looking for the URL for. Should be
one of the options specified in ``extra_links``.
"""
- if link_name in self.operator_extra_link_dict:
- return self.operator_extra_link_dict[link_name].get_link(self, dttm)
- elif link_name in self.global_operator_extra_link_dict:
- return self.global_operator_extra_link_dict[link_name].get_link(self, dttm)
+ link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
+ link_name
+ ) or self.global_operator_extra_link_dict.get(link_name)
+ if not link:
+ return None
+ # Check for old function signature
+ parameters = inspect.signature(link.get_link).parameters
+ args = [name for name, p in parameters.items() if p.kind != p.VAR_KEYWORD]
+ if "ti_key" in args:
Review comment:
@uranusjr Do you think this is enough of a check, or should we _force_ `ti_key` to be accepted only as a keyword argument?
Is it worth doing a `__init_subclass__` check on BaseOperatorLink to check the signature?
--
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] ashb commented on pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#issuecomment-1055262790
> Would it be a good idea if `serialize_value` also take `ti_key` instead? cc @dstandish
I agree, I've done it as part of https://github.com/apache/airflow/pull/21710/commits/2edebe2b3ff10b01bb416337b7e0ca5a19ef8dfc in #21710 (for no real reason, other than we are making larger changes there) -- this PR was concerned with just the changes needed for OperatorLink interface.
--
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] ashb commented on a change in pull request #21798: WIP: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r814269212
##########
File path: airflow/providers/qubole/operators/qubole.py
##########
@@ -39,7 +41,13 @@ class QDSLink(BaseOperatorLink):
name = 'Go to QDS'
- def get_link(self, operator: BaseOperator, dttm: datetime) -> str:
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
Oh not quite -- I can't use the "AbstractOperator" type as that only exists in 2.3.
Updated now.
--
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] ashb commented on a change in pull request #21798: WIP: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815184164
##########
File path: tests/serialization/test_dag_serialization.py
##########
@@ -915,68 +944,6 @@ def execute(self, context: Context):
)
assert expected_err_msg in caplog.text
- def test_extra_serialized_field_and_multiple_operator_links(self, dag_maker):
Review comment:
This test was broadly a duplicate of test_extra_serialized_field_and_operator_links -- so I made that parameterized instead.
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815922009
##########
File path: airflow/models/baseoperator.py
##########
@@ -1730,11 +1731,17 @@ def name(self) -> str:
"""
@abstractmethod
- def get_link(self, operator: BaseOperator, dttm: datetime) -> str:
+ def get_link(
+ self,
+ operator: AbstractOperator,
+ dttm: Optional[datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
Done in a43f10836
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815899497
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
Review comment:
No, I don't think so really -- the only use I can think of is by the webserver to show registered links.
--
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] ashb commented on pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#issuecomment-1055503573
Failure is just a timeout cos of
> error pulling image configuration: Get https://eastus.data.mcr.microsoft.com/aba285c624a04409823b708c7a50e7b9-jttfjm99vo//docker/registry/v2/blobs/sha256/d7/d78e982c2f2bdabbf34c3fd1b898ade9a25a22a5872ef23ea0c4fde1e6e3d7c8/data?se=2022-03-01T14%3A15%3A33Z&sig=UZQOZHXU5yR1CzSAY9MEBkjb0OWqR5hkCbL5TB%2By0r0%3D&sp=r&spr=https&sr=b&sv=2016-05-31®id=aba285c624a04409823b708c7a50e7b9: dial tcp 131.253.33.219:443: i/o timeout
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r816157523
##########
File path: airflow/operators/trigger_dagrun.py
##########
@@ -43,14 +48,27 @@ class TriggerDagRunLink(BaseOperatorLink):
name = 'Triggered DAG'
- def get_link(self, operator, dttm):
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime.datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
I'm trying to remove the overload, if that doesn't cause typing problems for the providers I'll leave it just with the new signature.
--
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] ashb merged pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb merged pull request #21798:
URL: https://github.com/apache/airflow/pull/21798
--
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] uranusjr commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815943384
##########
File path: airflow/operators/trigger_dagrun.py
##########
@@ -43,14 +48,27 @@ class TriggerDagRunLink(BaseOperatorLink):
name = 'Triggered DAG'
- def get_link(self, operator, dttm):
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime.datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
Continuing the backward compatibility topic, this particular class may not need to implement the compatibility signature? If we don’t support getting the link through the Python API, this one can only be called by Airflow core and only with the new API. (The same does not apply for operators in providers, of course.)
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815897260
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
"""For an operator, gets the URLs that the ``extra_links`` entry points to.
:raise ValueError: The error message of a ValueError will be passed on through to
the fronted to show up as a tooltip on the disabled link.
- :param dttm: The datetime parsed execution date for the URL being searched for.
+ :param ti: The TaskInstance for the URL being searched for.
:param link_name: The name of the link we're looking for the URL for. Should be
one of the options specified in ``extra_links``.
"""
- if link_name in self.operator_extra_link_dict:
- return self.operator_extra_link_dict[link_name].get_link(self, dttm)
- elif link_name in self.global_operator_extra_link_dict:
- return self.global_operator_extra_link_dict[link_name].get_link(self, dttm)
+ link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
+ link_name
+ ) or self.global_operator_extra_link_dict.get(link_name)
Review comment:
No, sadly not :(
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815944185
##########
File path: airflow/operators/trigger_dagrun.py
##########
@@ -43,14 +48,27 @@ class TriggerDagRunLink(BaseOperatorLink):
name = 'Triggered DAG'
- def get_link(self, operator, dttm):
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime.datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
You make a good point. I was just doing all of them without paying that much attention where the operator was defined.
--
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] uranusjr commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815938223
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
Review comment:
I mean potentially a third-party tool can rely on this to show registered links as well? This is listed in documentation, after all. (Although I’d definitely not object to less compatibility code personally.)
--
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 #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#issuecomment-1055275408
The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r816138782
##########
File path: airflow/operators/trigger_dagrun.py
##########
@@ -43,14 +48,27 @@ class TriggerDagRunLink(BaseOperatorLink):
name = 'Triggered DAG'
- def get_link(self, operator, dttm):
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime.datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
Oh,
```
airflow/operators/trigger_dagrun.py:51: error: Signature of "get_link"
incompatible with supertype "BaseOperatorLink" [override]
def get_link(
^
airflow/operators/trigger_dagrun.py:59: error: Argument "ti_key" to "get_one"
of "BaseXCom" has incompatible type "Optional[TaskInstanceKey]"; expected
"TaskInstanceKey" [arg-type]
when = XCom.get_one(ti_key=ti_key, key=XCOM_EXECUTION_DATE_ISO...
```
--
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] ashb commented on a change in pull request #21798: WIP: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815184495
##########
File path: airflow/providers/google/cloud/links/base.py
##########
@@ -0,0 +1,51 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from datetime import datetime
+from typing import TYPE_CHECKING, ClassVar, Optional
+
+from airflow.models import BaseOperatorLink, XCom
+
+if TYPE_CHECKING:
+ from airflow.models.taskinstance import TaskInstanceKey
+
+
+class BaseGoogleLink(BaseOperatorLink):
Review comment:
New base class for as almost all of the links in Google provider follow this simple pattern.
--
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] ashb commented on a change in pull request #21798: WIP: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r814886878
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
"""For an operator, gets the URLs that the ``extra_links`` entry points to.
:raise ValueError: The error message of a ValueError will be passed on through to
the fronted to show up as a tooltip on the disabled link.
- :param dttm: The datetime parsed execution date for the URL being searched for.
+ :param ti: The TaskInstance for the URL being searched for.
:param link_name: The name of the link we're looking for the URL for. Should be
one of the options specified in ``extra_links``.
"""
- if link_name in self.operator_extra_link_dict:
- return self.operator_extra_link_dict[link_name].get_link(self, dttm)
- elif link_name in self.global_operator_extra_link_dict:
- return self.global_operator_extra_link_dict[link_name].get_link(self, dttm)
+ link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
+ link_name
+ ) or self.global_operator_extra_link_dict.get(link_name)
+ if not link:
+ return None
+ # Check for old function signature
+ parameters = inspect.signature(link.get_link).parameters
+ args = [name for name, p in parameters.items() if p.kind != p.VAR_KEYWORD]
+ if "ti_key" in args:
Review comment:
@uranusjr Do you think tis is enough of a check, or should we _force_ `ti_key` to be accepted only as a keyword argument?
Is it worth doing a __init_subclass__ check on BaseOperatorLink to check the signature?
--
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] ashb commented on a change in pull request #21798: WIP: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r814107578
##########
File path: airflow/providers/qubole/operators/qubole.py
##########
@@ -39,7 +41,13 @@ class QDSLink(BaseOperatorLink):
name = 'Go to QDS'
- def get_link(self, operator: BaseOperator, dttm: datetime) -> str:
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
I think this signature and logic should be backwards and forwards compatible.
--
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] uranusjr commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815885054
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
Review comment:
Does this need to be considered public API?
##########
File path: airflow/models/abstractoperator.py
##########
@@ -394,3 +405,7 @@ def _render_nested_template_fields(
# content has no inner template fields
return
self._do_render_template_fields(value, nested_template_fields, context, jinja_env, seen_oids)
+
+
+if TYPE_CHECKING:
+ from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
Review comment:
Why are these not at the top?
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
"""For an operator, gets the URLs that the ``extra_links`` entry points to.
:raise ValueError: The error message of a ValueError will be passed on through to
the fronted to show up as a tooltip on the disabled link.
- :param dttm: The datetime parsed execution date for the URL being searched for.
+ :param ti: The TaskInstance for the URL being searched for.
:param link_name: The name of the link we're looking for the URL for. Should be
one of the options specified in ``extra_links``.
"""
- if link_name in self.operator_extra_link_dict:
- return self.operator_extra_link_dict[link_name].get_link(self, dttm)
- elif link_name in self.global_operator_extra_link_dict:
- return self.global_operator_extra_link_dict[link_name].get_link(self, dttm)
+ link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
+ link_name
+ ) or self.global_operator_extra_link_dict.get(link_name)
Review comment:
```suggestion
link: Optional["BaseOperatorLink"] = (
self.operator_extra_link_dict.get(link_name)
or self.global_operator_extra_link_dict.get(link_name)
)
```
Would Black be happy with this? The current format is… bad…
##########
File path: airflow/models/baseoperator.py
##########
@@ -1730,11 +1731,17 @@ def name(self) -> str:
"""
@abstractmethod
- def get_link(self, operator: BaseOperator, dttm: datetime) -> str:
+ def get_link(
+ self,
+ operator: AbstractOperator,
+ dttm: Optional[datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
`@overload` this?
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
"""For an operator, gets the URLs that the ``extra_links`` entry points to.
:raise ValueError: The error message of a ValueError will be passed on through to
the fronted to show up as a tooltip on the disabled link.
- :param dttm: The datetime parsed execution date for the URL being searched for.
+ :param ti: The TaskInstance for the URL being searched for.
:param link_name: The name of the link we're looking for the URL for. Should be
one of the options specified in ``extra_links``.
"""
- if link_name in self.operator_extra_link_dict:
- return self.operator_extra_link_dict[link_name].get_link(self, dttm)
- elif link_name in self.global_operator_extra_link_dict:
- return self.global_operator_extra_link_dict[link_name].get_link(self, dttm)
+ link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
+ link_name
+ ) or self.global_operator_extra_link_dict.get(link_name)
+ if not link:
+ return None
+ # Check for old function signature
+ parameters = inspect.signature(link.get_link).parameters
+ args = [name for name, p in parameters.items() if p.kind != p.VAR_KEYWORD]
+ if "ti_key" in args:
Review comment:
I think this should be good enough. Can’t think of all the future possibilities but this should be strict enough, if we always _call_ `get_link` with keyword `ti_key` (we already do here).
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815897878
##########
File path: airflow/models/abstractoperator.py
##########
@@ -394,3 +405,7 @@ def _render_nested_template_fields(
# content has no inner template fields
return
self._do_render_template_fields(value, nested_template_fields, context, jinja_env, seen_oids)
+
+
+if TYPE_CHECKING:
+ from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
Review comment:
Oh probably can be now. I was having trouble with what I think was an import cycle in typing.
--
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] uranusjr commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r816305634
##########
File path: airflow/operators/trigger_dagrun.py
##########
@@ -43,14 +48,27 @@ class TriggerDagRunLink(BaseOperatorLink):
name = 'Triggered DAG'
- def get_link(self, operator, dttm):
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime.datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
Just implementing the new signature should work fine. We could also implement the compatibility signature and ignore the compatibility arguments (with a note explaining this is fine).
--
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] ashb commented on a change in pull request #21798: WIP: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r814269212
##########
File path: airflow/providers/qubole/operators/qubole.py
##########
@@ -39,7 +41,13 @@ class QDSLink(BaseOperatorLink):
name = 'Go to QDS'
- def get_link(self, operator: BaseOperator, dttm: datetime) -> str:
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
Oh not quite -- I can't use the "AbstractOperator" type as that only exists in 2.3
--
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] ashb commented on a change in pull request #21798: WIP: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r814243513
##########
File path: airflow/models/operator.py
##########
@@ -16,11 +16,24 @@
# specific language governing permissions and limitations
# under the License.
+import datetime
from typing import Union
+from airflow.configuration import conf
from airflow.models.baseoperator import BaseOperator
from airflow.models.mappedoperator import MappedOperator
+from airflow.utils.trigger_rule import TriggerRule
+from airflow.utils.weight_rule import WeightRule
Operator = Union[BaseOperator, MappedOperator]
__all__ = ["Operator"]
+
+DEFAULT_OWNER = conf.get("operators", "default_owner")
+DEFAULT_POOL_SLOTS = 1
+DEFAULT_PRIORITY_WEIGHT = 1
+DEFAULT_QUEUE = conf.get("operators", "default_queue")
+DEFAULT_RETRIES = conf.getint("core", "default_task_retries", fallback=0)
+DEFAULT_RETRY_DELAY = datetime.timedelta(seconds=300)
+DEFAULT_WEIGHT_RULE = conf.get("core", "default_task_weight_rule", fallback=WeightRule.DOWNSTREAM)
+DEFAULT_TRIGGER_RULE = TriggerRule.ALL_SUCCESS
Review comment:
These had to be moved because I think mypy got in to an import loop and couldn't work out the type of these variables!
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815943682
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
Review comment:
True. I think it is vanishingly likely they are doing it directly in code and not via the API, so I'm calling this one as "not public API" and I'll add a `:meta private` to make it so.
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r816138782
##########
File path: airflow/operators/trigger_dagrun.py
##########
@@ -43,14 +48,27 @@ class TriggerDagRunLink(BaseOperatorLink):
name = 'Triggered DAG'
- def get_link(self, operator, dttm):
+ def get_link(
+ self,
+ operator: "AbstractOperator",
+ dttm: Optional[datetime.datetime] = None,
+ *,
+ ti_key: Optional["TaskInstanceKey"] = None,
+ ) -> str:
Review comment:
Oh,
```
airflow/operators/trigger_dagrun.py:51: error: Signature of "get_link"
incompatible with supertype "BaseOperatorLink" [override]
def get_link(
^
```
Maybe the override doesn't actually hurts us here?
--
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] ashb commented on a change in pull request #21798: Change BaseOperatorLink interface to take a ti_key, not a datetime
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #21798:
URL: https://github.com/apache/airflow/pull/21798#discussion_r815902580
##########
File path: airflow/models/abstractoperator.py
##########
@@ -239,19 +242,27 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
def extra_links(self) -> List[str]:
return list(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
- def get_extra_links(self, dttm: datetime.datetime, link_name: str) -> Optional[Dict[str, Any]]:
+ def get_extra_links(self, ti: "TaskInstance", link_name: str) -> Optional[str]:
"""For an operator, gets the URLs that the ``extra_links`` entry points to.
:raise ValueError: The error message of a ValueError will be passed on through to
the fronted to show up as a tooltip on the disabled link.
- :param dttm: The datetime parsed execution date for the URL being searched for.
+ :param ti: The TaskInstance for the URL being searched for.
:param link_name: The name of the link we're looking for the URL for. Should be
one of the options specified in ``extra_links``.
"""
- if link_name in self.operator_extra_link_dict:
- return self.operator_extra_link_dict[link_name].get_link(self, dttm)
- elif link_name in self.global_operator_extra_link_dict:
- return self.global_operator_extra_link_dict[link_name].get_link(self, dttm)
+ link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
+ link_name
+ ) or self.global_operator_extra_link_dict.get(link_name)
Review comment:
I could do it like this:
```python
link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(
link_name, self.global_operator_extra_link_dict.get(link_name, None)
)
```
(Which will do a unnecessary lookup to the global links)
Or this
```python
link: Optional["BaseOperatorLink"] = self.operator_extra_link_dict.get(link_name)
if not link:
link = self.global_operator_extra_link_dict.get(link_name)
```
--
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