1import collections 2from copy import deepcopy 3from typing import Any 4from typing import Callable 5from typing import DefaultDict 6from typing import Optional 7from typing import Set 8 9import attr 10 11from .internal.logger import get_logger 12 13 14log = get_logger(__name__) 15 16 17@attr.s(slots=True) 18class Hooks(object): 19 """ 20 Hooks configuration object is used for registering and calling hook functions 21 22 Example:: 23 24 @config.falcon.hooks.on('request') 25 def on_request(span, request, response): 26 pass 27 """ 28 29 _hooks = attr.ib(init=False, factory=lambda: collections.defaultdict(set), type=DefaultDict[str, Set]) 30 31 def __deepcopy__(self, memodict=None): 32 hooks = Hooks() 33 hooks._hooks = deepcopy(self._hooks, memodict) 34 return hooks 35 36 def register( 37 self, 38 hook, # type: Any 39 func=None, # type: Optional[Callable] 40 ): 41 # type: (...) -> Optional[Callable[..., Any]] 42 """ 43 Function used to register a hook for the provided name. 44 45 Example:: 46 47 def on_request(span, request, response): 48 pass 49 50 config.falcon.hooks.register('request', on_request) 51 52 53 If no function is provided then a decorator is returned:: 54 55 @config.falcon.hooks.register('request') 56 def on_request(span, request, response): 57 pass 58 59 :param hook: The name of the hook to register the function for 60 :type hook: object 61 :param func: The function to register, or ``None`` if a decorator should be returned 62 :type func: function, None 63 :returns: Either a function decorator if ``func is None``, otherwise ``None`` 64 :rtype: function, None 65 """ 66 # If they didn't provide a function, then return a decorator 67 if not func: 68 69 def wrapper(func): 70 self.register(hook, func) 71 return func 72 73 return wrapper 74 self._hooks[hook].add(func) 75 return None 76 77 # Provide shorthand `on` method for `register` 78 # >>> @config.falcon.hooks.on('request') 79 # def on_request(span, request, response): 80 # pass 81 on = register 82 83 def deregister( 84 self, 85 hook, # type: Any 86 func, # type: Callable 87 ): 88 # type: (...) -> None 89 """ 90 Function to deregister a function from a hook it was registered under 91 92 Example:: 93 94 @config.falcon.hooks.on('request') 95 def on_request(span, request, response): 96 pass 97 98 config.falcon.hooks.deregister('request', on_request) 99 100 :param hook: The name of the hook to register the function for 101 :type hook: object 102 :param func: Function hook to register 103 :type func: function 104 """ 105 if hook in self._hooks: 106 try: 107 self._hooks[hook].remove(func) 108 except KeyError: 109 pass 110 111 def emit( 112 self, 113 hook, # type: Any 114 *args, # type: Any 115 **kwargs # type: Any 116 ): 117 # type: (...) -> None 118 """ 119 Function used to call registered hook functions. 120 121 :param hook: The hook to call functions for 122 :type hook: str 123 :param args: Positional arguments to pass to the hook functions 124 :type args: list 125 :param kwargs: Keyword arguments to pass to the hook functions 126 :type kwargs: dict 127 """ 128 # Call registered hooks 129 for func in self._hooks.get(hook, ()): 130 try: 131 func(*args, **kwargs) 132 except Exception: 133 log.error("Failed to run hook %s function %s", hook, func, exc_info=True) 134