feat: on action decorator

src/viur/core/decorators.py

@@ -1,4 +1,8 @@
+import abc
+import collections
+import logging
 import typing as t
+
 from viur.core.module import Method
 
 __all__ = [
@@ -142,3 +146,50 @@ def decorator(func):
         return decorator
 
     return decorator(func)
+
+
+class OnAction(abc.ABC):
+    @property
+    @abc.abstractmethod
+    def action_name(self) -> str:
+        ...
+
+    def __init__(self, fn: t.Callable):
+        self.fn = fn
+
+    def __call__(self, *args, **kwargs):
+        return self.fn(*args, **kwargs)
+
+    def __set_name__(self, owner, name):
+        self.fn.class_name = owner.__name__
+        if not hasattr(owner, "on_handlers"):
+            owner.on_handlers: dict[str, list[t.Callable]] = collections.defaultdict(list)
+        owner.on_handlers[self.action_name].append(self.fn)
+
+    @classmethod
+    def dispatch(cls, caller, *args, raise_error: bool = True, **kwargs):
+        from viur.core import conf
+        for handler in caller.on_handlers[cls.action_name]:
+            if conf.debug.trace:
+                logging.debug(f"Calling {handler=} with {args=} and {kwargs=}")
+            try:
+                handler(caller, *args, **kwargs)
+            except Exception as exc:
+                if raise_error:
+                    raise
+                else:
+                    logging.exception(f"Calling {handler=} with {args=} and {kwargs=} failed: {exc}")
+
+    @classmethod
+    def create(cls, action_name: str) -> "t.Self":
+        return type(f"{action_name}Action", (cls,), {"action_name": action_name})
+
+
+on_add = OnAction.create("add")
+on_added = OnAction.create("added")
+on_edit = OnAction.create("edit")
+on_edited = OnAction.create("edited")
+on_delete = OnAction.create("delete")
+on_deleted = OnAction.create("deleted")
+on_clone = OnAction.create("clone")
+on_cloned = OnAction.create("cloned")

src/viur/core/prototypes/list.py

@@ -5,6 +5,7 @@
 from viur.core.cache import flushCache
 from viur.core.skeleton import SkeletonInstance
 from .skelmodule import SkelModule, DEFAULT_ORDER_TYPE
+from ..decorators import OnAction, on_add, on_added, on_clone, on_cloned, on_edit, on_edited
 
 
 class List(SkelModule):
@@ -565,6 +566,7 @@ def onAdd(self, skel: SkeletonInstance):
 
             .. seealso:: :func:`add`, :func:`onAdded`
         """
+        on_add.dispatch(self, skel)
         pass
 
     def onAdded(self, skel: SkeletonInstance):
@@ -582,6 +584,7 @@ def onAdded(self, skel: SkeletonInstance):
         flushCache(kind=skel.kindName)
         if user := current.user.get():
             logging.info(f"""User: {user["name"]!r} ({user["key"]!r})""")
+        on_added.dispatch(self, skel)
 
     def onEdit(self, skel: SkeletonInstance):
         """
@@ -593,6 +596,7 @@ def onEdit(self, skel: SkeletonInstance):
 
             .. seealso:: :func:`edit`, :func:`onEdited`
         """
+        on_edit.dispatch(self, skel)
         pass
 
     def onEdited(self, skel: SkeletonInstance):
@@ -610,6 +614,7 @@ def onEdited(self, skel: SkeletonInstance):
         flushCache(key=skel["key"])
         if user := current.user.get():
             logging.info(f"""User: {user["name"]!r} ({user["key"]!r})""")
+        on_edited.dispatch(self, skel)
 
     def onView(self, skel: SkeletonInstance):
         """
@@ -664,6 +669,7 @@ def onClone(self, skel: SkeletonInstance, src_skel: SkeletonInstance):
         .. seealso:: :func:`clone`, :func:`onCloned`
         """
         pass
+        on_clone.dispatch(self, skel)
 
     def onCloned(self, skel: SkeletonInstance, src_skel: SkeletonInstance):
         """
@@ -681,6 +687,7 @@ def onCloned(self, skel: SkeletonInstance, src_skel: SkeletonInstance):
 
         if user := utils.getCurrentUser():
             logging.info(f"""User: {user["name"]!r} ({user["key"]!r})""")
+        on_cloned.dispatch(self, skel)
 
 
 List.admin = True