Completed
Push — master ( c0d344...c61c65 )
by Piotr
43s
created

Schema   A

Complexity

Total Complexity 1

Size/Duplication

Total Lines 3
Duplicated Lines 0 %

Importance

Changes 1
Bugs 0 Features 0
Metric Value
c 1
b 0
f 0
dl 0
loc 3
rs 10
wmc 1
1
import abc
2
import inspect
3
import re
4
5
from mountapi.core import exceptions
6
7
8
class AbstractConverter(metaclass=abc.ABCMeta):
9
    param_url: str = None
10
    param_regex: str = None
11
12
    @classmethod
13
    def path_to_regex(cls, path):
14
        return re.sub(cls.param_url, cls.param_regex, path) + '$'
15
16
17
class IntConverter(AbstractConverter):
18
    param_url = r'<(?P<param>\w+):int>'
19
    param_regex = r'(?P<\1>\d+)'
20
21
22
class StrConverter(AbstractConverter):
23
    param_url = r'<(?P<param>\w+):str>'
24
    param_regex = r'(?P<\1>\w+)'
25
26
27
class AbstractSchema(exceptions.NotImplementedMixin, metaclass=abc.ABCMeta):
28
    @abc.abstractmethod
29
    def build(self):
30
        self.not_implemented()
31
32
    @abc.abstractmethod
33
    def match(self, path):
34
        self.not_implemented()
35
36
37
class Schema(AbstractSchema):
38
    _converter_map = {int: IntConverter, str: StrConverter}
39
40
    def __init__(self, routes: list) -> None:
41
        self._routes = routes
42
        self._schema = None
43
44
    def build(self) -> None:
45
        if self._schema is None:
46
            self._schema = self._build_schema()
47
48
    def _build_schema(self):
49
        schema = {}
50
        for route in self._routes:
51
            schema[route.path] = {
52
                'endpoint': route.endpoint,
53
                'regex': self._get_path_regex(route.path),
54
                **self._get_schema_http_methods(route)
55
            }
56
        return schema
57
58
    def _get_path_regex(self, path):
59
        for converter in self._converter_map.values():
60
            path = converter.path_to_regex(path)
61
62
        return path
63
64
    def _get_schema_http_methods(self, route):
65
        return {
66
            http_method: {
67
                'handler': getattr(route.endpoint, http_method.lower()),
68
                'params': self._get_func_params(
69
                    getattr(route.endpoint, http_method.lower())
70
                )
71
            } for http_method in route.endpoint.get_allowed_methods()
72
        }
73
74
    def _get_func_params(self, func):
75
        return {
76
            param.name: self._converter_map[param.annotation]
77
            for param in inspect.signature(func).parameters.values()
78
            if param.annotation != inspect.Parameter.empty
79
        }
80
81
    def match(self, path):
82
        for route_path in self._schema:
83
            route_match = re.match(self._schema[route_path]['regex'], path)
84
            if route_match:
85
                return {
86
                    'endpoint': self._schema[route_path]['endpoint'],
87
                    'kwargs': route_match.groupdict()
88
                }
89
        else:
90
            raise exceptions.NotFound()
91