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