File size: 5,341 Bytes
079c32c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
import random
import string
from abc import ABCMeta, abstractmethod
from typing import Optional, Callable, Mapping, Any, Dict

_LENGTH_OF_RANDOM_TOKEN = 64


def random_token(length: Optional[int] = None) -> str:
    """
    Overview:
        Generate random hex token
    Arguments:
        - length (:obj:`Optional[int]`): Length of the random token (`None` means `64`)
    Returns:
        - token (:obj:`str`): Generated random token
    Example:
        >>> random_token()  # '4eAbd5218e3d0da5e7AAFcBF48Ea0Df2dadED1bdDF0B8724FdE1569AA78F24A7'
        >>> random_token(24)  # 'Cd1CdD98caAb8602ac6501aC'
    """
    return ''.join([random.choice(string.hexdigits) for _ in range(length or _LENGTH_OF_RANDOM_TOKEN)])


class ControllableContext(metaclass=ABCMeta):
    """
    Overview:
        Basic context-supported class structure
    Example:
        - Common usage

        >>> c = MyControllableContext()  # One of the superclasses if ControllableContext
        >>> c.start()
        >>> try:
        >>>     pass  # do anything you like
        >>> finally:
        >>>     c.close()

        - Use with keyword (the same as code above)

        >>> c = MyControllableContext()  # One of the superclasses if ControllableContext
        >>> with c as cc:   # cc is c, have the same id
        >>>     pass  # do anything you like
    """

    @abstractmethod
    def start(self):
        """
        Overview:
            Start the context
        """
        raise NotImplementedError  # pragma: no cover

    @abstractmethod
    def close(self):
        """
        Overview:
            Close the context
        """
        raise NotImplementedError  # pragma: no cover

    def __enter__(self):
        """
        Overview:
            Enter the context
        Returns:
            - self (:obj:`ControllableContext`): Context object itself
        """
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Overview:
            Exit the context
        """
        self.close()


class ControllableService(ControllableContext, metaclass=ABCMeta):
    """
    Overview:
        Controllable service with context support, usually has concurrent feature.
    Example:
        - A common usage

        >>> c = MyControllableService()  # One of its superclasses is ControllableService
        >>> c.start()
        >>> try:
        >>>     pass  # do anything you like
        >>> finally:
        >>>     c.shutdown()  # shutdown the service
        >>>     c.join()  # wait until service is down

        - Use with keyword (the same as code above)

        >>> c = MyControllableService()  # One of its superclasses is ControllableService
        >>> with c as cc:   # cc is c, have the same id
        >>>     pass  # do anything you like
    """

    @abstractmethod
    def start(self):
        """
        Overview:
            Start the service
        """
        raise NotImplementedError  # pragma: no cover

    @abstractmethod
    def shutdown(self):
        """
        Overview:
            Shutdown the service (but service will not down immediately)
        """
        raise NotImplementedError  # pragma: no cover

    @abstractmethod
    def join(self):
        """
        Overview:
            Wait until the service is completely down
        """
        raise NotImplementedError  # pragma: no cover

    def close(self):
        """
        Overview:
            Close the service, wait until the service is down.
        """
        self.shutdown()
        self.join()


def translate_dict_func(d: Mapping[str, Callable[..., Any]]) -> Callable[..., Dict[str, Any]]:
    """
    Overview:
        Transform dict with funcs to function generating dict.
    Arguments:
        - d (:obj:`Mapping[str, Callable[..., Any]]`): Dict with funcs
    Returns:
        - func (:obj:`Callable[..., Dict[str, Any]]`): Function generating dict
    Example:
        >>> f1 = lambda x, y: x + y
        >>> f2 = lambda x, y: x - y
        >>> f3 = lambda x, y: x * y
        >>> fx = translate_dict_func({'a': f1, 'b': f2, 'c': f3})
        >>> fx(2, 3)  # {'a': 5, 'b': -1, 'c': 6}
        >>> fx(5, 11)  # ('a': 16, 'b': -6, 'c': 55}
    """

    def _func(*args, **kwargs) -> Dict[str, Any]:
        return {k: f(*args, **kwargs) for k, f in d.items()}

    return _func


def default_func(return_value=None) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """
    Overview:
        Transform optional function (maybe `None`) to function with default value
    Argument:
        - return_value (:obj:): Return value of the default function
    Returns:
        - decorator (:obj:`Callable[[Callable[..., Any]], Callable[..., Any]]`): A decorator function \
            that can decorator optional function to real function (must be not None)
    Example:
        >>> f1 = None
        >>> f2 = lambda x, y: x + y
        >>> ff1 = default_func()(f1)
        >>> ft1 = default_func(0)(f1)
        >>> ff2 = default_func()(f2)
        >>> ff1(2, 3)  # None
        >>> ft1(2, 3)  # 0
        >>> ff2(2, 3)  # 5
    """

    def _decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        # noinspection PyUnusedLocal
        def _func(*args, **kwargs):
            return return_value

        return func or _func

    return _decorator