ashb commented on code in PR #46613:
URL: https://github.com/apache/airflow/pull/46613#discussion_r1950764591
##########
airflow/models/baseoperatorlink.py:
##########
@@ -20,14 +20,47 @@
from abc import ABCMeta, abstractmethod
from typing import TYPE_CHECKING, ClassVar
-import attr
+import attrs
+
+from airflow.models.xcom import BaseXCom
+from airflow.utils.log.logging_mixin import LoggingMixin
if TYPE_CHECKING:
from airflow.models.baseoperator import BaseOperator
from airflow.models.taskinstancekey import TaskInstanceKey
[email protected](auto_attribs=True)
[email protected]()
+class GenericOperatorLink(LoggingMixin):
+ """A generic operator link class that can retrieve link only using XCOMs.
Used while deserializing operators."""
+
+ name: str
+ xcom_key: str
+
+ def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) ->
str:
+ """
+ Retrieve the link from the XComs.
+
+ :param operator: The Airflow operator object this link is associated
to.
+ :param ti_key: TaskInstance ID to return link for.
+ :return: link to external system, but by pulling it from XComs
+ """
+ self.log.info(
+ "Attempting to retrieve link from XComs with key: %s for task id:
%s", self.xcom_key, ti_key
+ )
+ value = BaseXCom.get_one(key=self.xcom_key, run_id=ti_key.run_id)
+ if not value:
+ self.log.debug(
+ "No link with name: %s present in XCom as key: %s, returning
empty link",
+ self.name,
+ self.xcom_key,
+ )
+ return ""
Review Comment:
I'd probably say that this specific case should fail/except here, but maybe
catch that in the API layer? (so that you can view an old TI even if the link
xcom has been deleted by a db clean process?)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
