aio-libs / aiohttp
1 10
import abc
2 10
import dataclasses
3 10
import os  # noqa
4 10
from typing import (
5
    TYPE_CHECKING,
6
    Any,
7
    Awaitable,
8
    Callable,
9
    Dict,
10
    Iterator,
11
    List,
12
    Optional,
13
    Sequence,
14
    Type,
15
    Union,
16
    overload,
17
)
18

19 10
from . import hdrs
20 10
from .abc import AbstractView
21 10
from .typedefs import PathLike
22

23
if TYPE_CHECKING:  # pragma: no cover
24
    from .web_request import Request
25
    from .web_response import StreamResponse
26
    from .web_urldispatcher import AbstractRoute, UrlDispatcher
27
else:
28 10
    Request = StreamResponse = UrlDispatcher = AbstractRoute = None
29

30

31 10
__all__ = (
32
    "AbstractRouteDef",
33
    "RouteDef",
34
    "StaticDef",
35
    "RouteTableDef",
36
    "head",
37
    "options",
38
    "get",
39
    "post",
40
    "patch",
41
    "put",
42
    "delete",
43
    "route",
44
    "view",
45
    "static",
46
)
47

48

49 10
class AbstractRouteDef(abc.ABC):
50 10
    @abc.abstractmethod
51 10
    def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
52
        pass  # pragma: no cover
53

54

55 10
_SimpleHandler = Callable[[Request], Awaitable[StreamResponse]]
56 10
_HandlerType = Union[Type[AbstractView], _SimpleHandler]
57

58

59 10
@dataclasses.dataclass(frozen=True, repr=False)
60 10
class RouteDef(AbstractRouteDef):
61 10
    method: str
62 10
    path: str
63 10
    handler: _HandlerType
64 10
    kwargs: Dict[str, Any]
65

66 10
    def __repr__(self) -> str:
67 10
        info = []
68 10
        for name, value in sorted(self.kwargs.items()):
69 10
            info.append(f", {name}={value!r}")
70 10
        return "<RouteDef {method} {path} -> {handler.__name__!r}" "{info}>".format(
71
            method=self.method, path=self.path, handler=self.handler, info="".join(info)
72
        )
73

74 10
    def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
75 10
        if self.method in hdrs.METH_ALL:
76 10
            reg = getattr(router, "add_" + self.method.lower())
77 10
            return [reg(self.path, self.handler, **self.kwargs)]
78
        else:
79 10
            return [
80
                router.add_route(self.method, self.path, self.handler, **self.kwargs)
81
            ]
82

83

84 10
@dataclasses.dataclass(frozen=True, repr=False)
85 10
class StaticDef(AbstractRouteDef):
86 10
    prefix: str
87 10
    path: PathLike
88 10
    kwargs: Dict[str, Any]
89

90 10
    def __repr__(self) -> str:
91 10
        info = []
92 10
        for name, value in sorted(self.kwargs.items()):
93 10
            info.append(f", {name}={value!r}")
94 10
        return "<StaticDef {prefix} -> {path}" "{info}>".format(
95
            prefix=self.prefix, path=self.path, info="".join(info)
96
        )
97

98 10
    def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
99 10
        resource = router.add_static(self.prefix, self.path, **self.kwargs)
100 10
        routes = resource.get_info().get("routes", {})
101 10
        return list(routes.values())
102

103

104 10
def route(method: str, path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
105 10
    return RouteDef(method, path, handler, kwargs)
106

107

108 10
def head(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
109 10
    return route(hdrs.METH_HEAD, path, handler, **kwargs)
110

111

112 10
def options(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
113 10
    return route(hdrs.METH_OPTIONS, path, handler, **kwargs)
114

115

116 10
def get(
117
    path: str,
118
    handler: _HandlerType,
119
    *,
120
    name: Optional[str] = None,
121
    allow_head: bool = True,
122
    **kwargs: Any,
123
) -> RouteDef:
124 10
    return route(
125
        hdrs.METH_GET, path, handler, name=name, allow_head=allow_head, **kwargs
126
    )
127

128

129 10
def post(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
130 10
    return route(hdrs.METH_POST, path, handler, **kwargs)
131

132

133 10
def put(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
134 10
    return route(hdrs.METH_PUT, path, handler, **kwargs)
135

136

137 10
def patch(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
138 10
    return route(hdrs.METH_PATCH, path, handler, **kwargs)
139

140

141 10
def delete(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
142 10
    return route(hdrs.METH_DELETE, path, handler, **kwargs)
143

144

145 10
def view(path: str, handler: Type[AbstractView], **kwargs: Any) -> RouteDef:
146 10
    return route(hdrs.METH_ANY, path, handler, **kwargs)
147

148

149 10
def static(prefix: str, path: PathLike, **kwargs: Any) -> StaticDef:
150 10
    return StaticDef(prefix, path, kwargs)
151

152

153 10
_Deco = Callable[[_HandlerType], _HandlerType]
154

155

156 10
class RouteTableDef(Sequence[AbstractRouteDef]):
157
    """Route definition table"""
158

159 10
    def __init__(self) -> None:
160 10
        self._items = []  # type: List[AbstractRouteDef]
161

162 10
    def __repr__(self) -> str:
163 10
        return "<RouteTableDef count={}>".format(len(self._items))
164

165 10
    @overload
166 10
    def __getitem__(self, index: int) -> AbstractRouteDef:
167 0
        ...
168

169 10
    @overload
170 10
    def __getitem__(self, index: slice) -> List[AbstractRouteDef]:
171 0
        ...
172

173 10
    def __getitem__(self, index):  # type: ignore[no-untyped-def]
174 10
        return self._items[index]
175

176 10
    def __iter__(self) -> Iterator[AbstractRouteDef]:
177 10
        return iter(self._items)
178

179 10
    def __len__(self) -> int:
180 10
        return len(self._items)
181

182 10
    def __contains__(self, item: object) -> bool:
183 10
        return item in self._items
184

185 10
    def route(self, method: str, path: str, **kwargs: Any) -> _Deco:
186 10
        def inner(handler: _HandlerType) -> _HandlerType:
187 10
            self._items.append(RouteDef(method, path, handler, kwargs))
188 10
            return handler
189

190 10
        return inner
191

192 10
    def head(self, path: str, **kwargs: Any) -> _Deco:
193 10
        return self.route(hdrs.METH_HEAD, path, **kwargs)
194

195 10
    def get(self, path: str, **kwargs: Any) -> _Deco:
196 10
        return self.route(hdrs.METH_GET, path, **kwargs)
197

198 10
    def post(self, path: str, **kwargs: Any) -> _Deco:
199 10
        return self.route(hdrs.METH_POST, path, **kwargs)
200

201 10
    def put(self, path: str, **kwargs: Any) -> _Deco:
202 10
        return self.route(hdrs.METH_PUT, path, **kwargs)
203

204 10
    def patch(self, path: str, **kwargs: Any) -> _Deco:
205 10
        return self.route(hdrs.METH_PATCH, path, **kwargs)
206

207 10
    def delete(self, path: str, **kwargs: Any) -> _Deco:
208 10
        return self.route(hdrs.METH_DELETE, path, **kwargs)
209

210 10
    def options(self, path: str, **kwargs: Any) -> _Deco:
211 10
        return self.route(hdrs.METH_OPTIONS, path, **kwargs)
212

213 10
    def view(self, path: str, **kwargs: Any) -> _Deco:
214 10
        return self.route(hdrs.METH_ANY, path, **kwargs)
215

216 10
    def static(self, prefix: str, path: PathLike, **kwargs: Any) -> None:
217 10
        self._items.append(StaticDef(prefix, path, kwargs))

Read our documentation on viewing source code .

Loading