Skip to content

api.common

AmbiguousParameters

AmbiguousParameters

Source code in api/common/exceptions.py 
6
7
class AmbiguousParameters(SapiBaseException):
    """AmbiguousParameters"""

ClsNotRegisteredException

ClsNotRegisteredException

Source code in api/common/exceptions.py 
18
19
class ClsNotRegisteredException(SerializationException):
    """ClsNotRegisteredException"""

ConfigException

ConfigException

Source code in api/common/exceptions.py 
24
25
class ConfigException(SapiBaseException):
    """ConfigException"""

FaultyRoleException

FaultyRoleException

Source code in api/common/exceptions.py 
27
28
class FaultyRoleException(ConfigException):
    """FaultyRoleException"""

InsufficientParameters

InsufficientParameters

Source code in api/common/exceptions.py 
3
4
class InsufficientParameters(SapiBaseException):
    """InsufficientParameters"""

InvalidParameters

InvalidParameters

Source code in api/common/exceptions.py 
12
13
class InvalidParameters(SapiBaseException):
    """InvalidParameters"""

NonStrKeyException

NonStrKeyException

Source code in api/common/exceptions.py 
21
22
class NonStrKeyException(SerializationException):
    """NonStrKeyException"""

NotFound

NotFound

Source code in api/common/exceptions.py 
 9
10
class NotFound(SapiBaseException):
    """NotFound"""

SerializationException

SerializationException

Source code in api/common/exceptions.py 
15
16
class SerializationException(SapiBaseException):
    """SerializationException"""

decorators

async_router_decorator(role, *, func)

Async Router decorator

Parameters:

Name Type Description Default
role ROLE_TYPE

role of this process

 required
func

function to be queued, or called, based on role

 required

Raises: AssertionError: if wrapped function is not async TimeoutError: if the async process took more than 600sec to complete FaultyRoleException: if role is set to other than all or server

Source code in api/common/decorators.py 
 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
def async_router_decorator(role: ROLE_TYPE, *, func):
    """Async Router decorator

    Args:
        role: role of this process
        func: function to be queued, or called, based on role
    Raises:
        AssertionError: if wrapped function is not async
        TimeoutError: if the async process took more than 600sec to complete
        FaultyRoleException: if role is set to other than `all` or `server`
    """
    def outer(fn):
        global name_to_fns_map

        assert fn.__name__ not in name_to_fns_map, f"{fn.__name__} already in name_to_fns_map"
        name_to_fns_map[fn.__name__] = (fn, func)

        assert inspect.iscoroutinefunction(fn), f"async_router_decorator can only be used to decorate async functions"

        return_func = None
        if role == "all":
            async def async_get_direct_result(*args, **kwargs):
                return func(*args, **kwargs)
            return_func = async_get_direct_result
        if role == "server":
            async def async_get_result(*args, **kwargs):
                if isinstance(func, partial):
                    func.func.s(*func.args, **func.keywords)
                    _func = func.func
                    args=[*args, *func.args]
                    kwargs={**kwargs, **func.keywords}
                else:
                    _func=func

                async_result = _func.apply_async(args, kwargs)
                wait=0.1
                timestamp=time.time()
                while True:
                    if async_result.status == "SUCCESS":
                        return async_result.get()
                    if async_result.status == "FAILURE":
                        # on failure, getting the result will raise the exception
                        async_result.get()
                        raise Exception("unknown exception")
                    if (time.time() - timestamp) > 600:
                        async_result.revoke()
                        raise TimeoutError
                    await asyncio.sleep(wait)
                    wait = min(wait + wait, 1) # increase sleep duration, but at most 1 sec
            return_func = async_get_result

        if return_func is None:
            raise FaultyRoleException(f"router_decorator can only be used in roles: all, server, but you selected {role}")

        @wraps(fn)
        async def inner(*args, **kwargs):
            return await fn(*args, **kwargs, func=return_func)
        return inner
    return outer

data_decorator(role)

data decorator

Extensively used in api.common.data_handlers. Most of the business logic, including data fetching, serialization, etc.

Parameters:

Name Type Description Default
role ROLE_TYPE

Role of this process

 required

Raises:

Type Description
ImportError

Celery not installed, but role is set to either worker or server
Source code in api/common/decorators.py 
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
def data_decorator(role: ROLE_TYPE):
    """data decorator

    Extensively used in api.common.data_handlers. Most of the business logic, including data fetching, serialization, etc.

    Args:
        role: Role of this process

    Raises:
        ImportError: Celery not installed, but role is set to either `worker` or `server`
    """
    def outer_wrapper(fn):
        if role == "all":
            return fn
        if role == "worker" or role == "server":
            try:
                from api.worker import app

                def celery_task_wrapper(self, *args, **kwargs):
                    return fn(*args, **kwargs)

                return app.task(bind=True)(
                    wraps(fn)(celery_task_wrapper)
                )
            except ImportError as e:
                errmsg = f"For worker role, celery must be installed as a dep"
                logger.critical(errmsg)
                raise ImportError(errmsg) from e
        raise FaultyRoleException(f"role must be 'all', 'server' or 'worker', but it is {role}")
    return outer_wrapper

router_decorator(role, *, func, queue_as_async=False, **kwargs)

Sync Router Decorator

Parameters:

Name Type Description Default
role ROLE_TYPE

role of this process

 required
func

function to be queued, or called, based on role

 required
queue_as_async bool

if set, will return id of the task, instead of task itself. Query the id for result.

 False

Raises:

Type Description
FaultyRoleException

if role is set to all and queue_as_async is set
FaultyRoleException

if role is set to other than all or server
Source code in api/common/decorators.py 
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
def router_decorator(role: ROLE_TYPE, *, func, queue_as_async: bool=False, **kwargs):
    """Sync Router Decorator

    Args:
        role: role of this process
        func: function to be queued, or called, based on role
        queue_as_async: if set, will return id of the task, instead of task itself. Query the id for result.

    Raises:
        FaultyRoleException: if role is set to `all` and `queue_as_async` is set
        FaultyRoleException: if role is set to other than `all` or `server`
    """
    def outer(fn):

        global name_to_fns_map

        assert fn.__name__ not in name_to_fns_map, f"{fn.__name__} already in name_to_fns_map"
        name_to_fns_map[fn.__name__] = (fn, func)

        return_func = None
        if role == "all":
            if queue_as_async:
                raise FaultyRoleException(f"If role is set to all, cannot queue_as_async")
            return_func = func
        if role == "server":
            def sync_get_result(*args, **kwargs):
                if isinstance(func, partial):
                    func.func.s(*func.args, **func.keywords)
                    _func = func.func
                    args=[*args, *func.args]
                    kwargs={**kwargs, **func.keywords}
                else:
                    _func=func

                async_result = _func.apply_async(args, kwargs)
                if queue_as_async:
                    return async_result.id
                else:
                    return async_result.get()
            return_func = sync_get_result
        if return_func is None:
            raise FaultyRoleException(f"router_decorator can only be used in roles: all, server, but you selected {role}")

        @wraps(fn)
        def inner(*args, **kwargs):
            return fn(*args, **kwargs, func=return_func)
        return inner
    return outer

exceptions

AmbiguousParameters

AmbiguousParameters

Source code in api/common/exceptions.py 
6
7
class AmbiguousParameters(SapiBaseException):
    """AmbiguousParameters"""

ClsNotRegisteredException

ClsNotRegisteredException

Source code in api/common/exceptions.py 
18
19
class ClsNotRegisteredException(SerializationException):
    """ClsNotRegisteredException"""

ConfigException

ConfigException

Source code in api/common/exceptions.py 
24
25
class ConfigException(SapiBaseException):
    """ConfigException"""

FaultyRoleException

FaultyRoleException

Source code in api/common/exceptions.py 
27
28
class FaultyRoleException(ConfigException):
    """FaultyRoleException"""

InsufficientParameters

InsufficientParameters

Source code in api/common/exceptions.py 
3
4
class InsufficientParameters(SapiBaseException):
    """InsufficientParameters"""

InvalidParameters

InvalidParameters

Source code in api/common/exceptions.py 
12
13
class InvalidParameters(SapiBaseException):
    """InvalidParameters"""

NonStrKeyException

NonStrKeyException

Source code in api/common/exceptions.py 
21
22
class NonStrKeyException(SerializationException):
    """NonStrKeyException"""

NotFound

NotFound

Source code in api/common/exceptions.py 
 9
10
class NotFound(SapiBaseException):
    """NotFound"""

SerializationException

SerializationException

Source code in api/common/exceptions.py 
15
16
class SerializationException(SapiBaseException):
    """SerializationException"""

logger

Logging module.

if LOGGER_DIR is defined in config, will also use TimedRotatingFileHandler to write to LOGGER_DIR.

storage

get_filename(*args, ext=None)

Get a hashed filename based on positional arguments.

Will also honor SIIBRA_API_SHARED_DIR in config, if defined.

Parameters:

Name Type Description Default
args List[str]

positional arguments

 ()
ext str

extension

 None

Returns:

Type Description
str

hashed path, in the form of {SIIBRA_API_SHARED_DIR}/{hash(*args)} + ('.{ext}' if ext else '')
Source code in api/common/storage.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
def get_filename(*args: List[str], ext:str=None) -> str:
    """Get a hashed filename based on positional arguments.

    Will also honor `SIIBRA_API_SHARED_DIR` in config, if defined.

    Args:
        args: positional arguments
        ext: extension

    Returns:
        hashed path, in the form of `{SIIBRA_API_SHARED_DIR}/{hash(*args)} + ('.{ext}' if ext else '')`
    """
    assert all(isinstance(arg, str) for arg in args), f"all args to get_filename must be str"
    return os.path.join(SIIBRA_API_SHARED_DIR, hashlib.md5("".join(args).encode("utf-8")).hexdigest() + (f".{ext.lstrip('.')}") if ext else "")

timer

Cron

Source code in api/common/timer.py 
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
class Cron:
    def __init__(self) -> None:
        self._minutely_fns: List[Callable] = []
        self._ten_minutely_fns: List[Callable] = []

        self._timers: List[RepeatTimer] = [
            RepeatTimer(60, self._run_minutely),
            RepeatTimer(600, self._run_ten_minutely)
        ]

    def _run_minutely(self):
        for fn in self._minutely_fns:
            fn()

    def _run_ten_minutely(self):
        for fn in self._ten_minutely_fns:
            fn()

    def minutely(self, fn: Callable):
        self._minutely_fns.append(fn)
        return fn

    def ten_minutely(self, fn: Callable):
        self._ten_minutely_fns.append(fn)
        return fn

    def run_all(self):
        self._run_ten_minutely()
        self._run_minutely()

    def start(self):
        for timer in self._timers:
            timer.start()

    def stop(self):
        """On terminate"""
        for timer in self._timers:
            timer.cancel()

stop()

On terminate

Source code in api/common/timer.py 
47
48
49
50
def stop(self):
    """On terminate"""
    for timer in self._timers:
        timer.cancel()

RepeatTimer

RepeatTimer

runs self.function every self.interval

Source code in api/common/timer.py 
 4
 5
 6
 7
 8
 9
10
11
class RepeatTimer(Timer):
    """RepeatTimer

    runs `self.function` every `self.interval`
    """
    def run(self):
        while not self.finished.wait(self.interval):
            self.function(*self.args, **self.kwargs)