diff --git a/libervia/backend/tools/trigger.py b/libervia/backend/tools/trigger.py index 8face230..9314c801 100644 --- a/libervia/backend/tools/trigger.py +++ b/libervia/backend/tools/trigger.py @@ -20,7 +20,7 @@ """Misc usefull classes""" import inspect -from typing import Callable +from typing import Any, Callable from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.core.core_types import SatXMPPEntity @@ -52,23 +52,14 @@ class TriggerManager(object): def __init__(self): self.__triggers = {} - def is_available(self, args: list): - """Check if plugin used in a client context, and if it's available""" - if not args or not isinstance(args[0], SatXMPPEntity): - # we are not in the context of a client - return True - client = args[0] - if not client.is_component: - # plugins are always avaialble for normal clients - return True - - def add(self, point_name, callback: Callable, priority=0): - """Add a trigger to a point + def add( + self, point_name: str, callback: Callable[..., Any], priority: int = 0 + ) -> None: + """Add a trigger to a point. @param point_name: name of the point when the trigger should be run @param callback: method to call at the trigger point - @param priority: callback will be called in priority order, biggest - first + @param priority: callback will be called in priority order, biggest first """ if point_name not in self.__triggers: self.__triggers[point_name] = [] @@ -88,9 +79,13 @@ class TriggerManager(object): ) def add_with_check( - self, point_name: str, plugin, callback: Callable, priority: int = 0 + self, + point_name: str, + plugin: object, + callback: Callable[..., Any], + priority: int = 0, ) -> None: - """Like [Add], but check session before running the trigger + """Like [add], but checks session before running the trigger. This method is to be used for triggers which can run in components and are expecting a ``client``: as all plugins are not run for component, a check will be @@ -124,11 +119,11 @@ class TriggerManager(object): self.add(point_name, sync_wrapper, priority) - def remove(self, point_name, callback): - """Remove a trigger from a point + def remove(self, point_name: str, callback: Callable[..., Any]) -> None: + """Remove a trigger from a point. @param point_name: name of the point when the trigger should be run - @param callback: method to remove, must exists in the trigger point + @param callback: method to remove; must exist in the trigger point """ for trigger_tuple in self.__triggers[point_name]: if trigger_tuple[1] == callback: @@ -136,17 +131,16 @@ class TriggerManager(object): return raise TriggerException("Trying to remove an unexisting trigger") - def point(self, point_name, *args, **kwargs): - """This put a trigger point + def point(self, point_name: str, *args: Any, **kwargs: Any) -> bool: + """Put a trigger point. - All the triggers for that point will be run + All the triggers for that point will be run. @param point_name: name of the trigger point - @param *args: args to transmit to trigger - @param *kwargs: kwargs to transmit to trigger - if "triggers_no_cancel" is present, it will be popup out - when set to True, this argument don't let triggers stop - the workflow - @return: True if the action must be continued, False else + @param args: arguments to transmit to triggers + @param kwargs: keyword arguments to transmit to triggers. If + ``triggers_no_cancel`` is present and set to True, it will be + popped and triggers cannot stop the workflow. + @return: True if the action must be continued, False otherwise """ if point_name not in self.__triggers: return True @@ -161,18 +155,23 @@ class TriggerManager(object): break return True - def return_point(self, point_name, *args, **kwargs): - """Like point but trigger must return (continue, return_value) + def return_point( + self, point_name: str, *args: Any, **kwargs: Any + ) -> tuple[bool, Any | None]: + """Like [point] but triggers must return (continue, return_value). All triggers for that point must return a tuple with 2 values: - - continue, same as for point, if False action must be finished - - return_value: value to return ONLY IF CONTINUE IS FALSE + - continue: same as for [point], if False action must be finished + - return_value: value to return ONLY IF continue is False @param point_name: name of the trigger point - @return: True if the action must be continued, False else + @param args: arguments to transmit to triggers + @param kwargs: keyword arguments to transmit to triggers + @return: tuple of (continue, return_value); True/None if continued, + False/value if a trigger stopped the workflow """ if point_name not in self.__triggers: - return True + return True, None for priority, trigger in self.__triggers[point_name]: try: