[GitHub] [airflow] dstandish commented on a diff in pull request #31863: Add default Trigger.serialize implementation

["dstandish (via GitHub)" <gi...@apache.org>]

dstandish commented on code in PR #31863:
URL: https://github.com/apache/airflow/pull/31863#discussion_r1230100045


##########
airflow/triggers/base.py:
##########
@@ -50,14 +51,55 @@ def _set_context(self, context):
         """
         raise NotImplementedError
 
-    @abc.abstractmethod
+    def _serialize_classpath(self):
+        class_name = self.__class__.__name__
+        mod = inspect.getmodule(self)
+
+        # easy case
+        if mod and mod.__name__ != "__main__":
+            return f"{mod.__name__}.{class_name}"
+
+        # tougher: the file being run defines the method, or we're in repl
+        mod_file = inspect.getfile(self.__class__)
+
+        # in repl: impossible
+        if mod_file == "<input>":
+            raise ValueError(
+                "Logic for automated serialization of trigger does not work "
+                "when defining the class in a repl; please define the class "
+                "in another module and import it; alternatively, implement "
+                "the `serialize` method and hardcode the import path."
+            )
+
+        from airflow import AIRFLOW_ROOT
+
+        # entrypoint file defines the class
+        try:
+            rel_path = Path(mod_file).relative_to(AIRFLOW_ROOT)
+        except ValueError:
+            # entrypoint file not relative to airflow dir; can't guess classpath
+            raise ValueError(
+                f"Class {class_name} is not located in a submodule of "
+                f"Airflow. You must implement its `serialize` method and hardcode "
+                f"the fully qualified class name."
+            )
+        mod_full_name = ".".join([*rel_path.parts[:-1], rel_path.stem])
+        return f"airflow.{mod_full_name}.{class_name}"
+
+    def _serialize_kwargs(self):
+        params = inspect.signature(self.__init__).parameters.keys()  # type: ignore[misc]
+        return {k: getattr(self, k) for k in params}
+
     def serialize(self) -> tuple[str, dict[str, Any]]:
         """
         Returns the information needed to reconstruct this Trigger.
 
+        In most cases the default implementation should work but there may be some cases where
+        you will need to implement.

Review Comment:
   ```suggestion
           you will need to override this method.
   ```



-- 
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