Team Ai
Datasetpublic

codekingpro/portable-devtools

sourceHugging Faceupdated 5mo agoView on Hugging Face
1likes15kdownloads
web_routedef.py215 linesDownload Raw Back to aiohttp
1import abc
2import os  # noqa
3from typing import (
4    TYPE_CHECKING,
5    Any,
6    Callable,
7    Dict,
8    Iterator,
9    List,
10    Optional,
11    Sequence,
12    Type,
13    Union,
14    overload,
15)
16
17import attr
18
19from . import hdrs
20from .abc import AbstractView
21from .typedefs import Handler, PathLike
22
23if TYPE_CHECKING:
24    from .web_request import Request
25    from .web_response import StreamResponse
26    from .web_urldispatcher import AbstractRoute, UrlDispatcher
27else:
28    Request = StreamResponse = UrlDispatcher = AbstractRoute = None
29
30
31__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
49class AbstractRouteDef(abc.ABC):
50    @abc.abstractmethod
51    def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
52        pass  # pragma: no cover
53
54
55_HandlerType = Union[Type[AbstractView], Handler]
56
57
58@attr.s(auto_attribs=True, frozen=True, repr=False, slots=True)
59class RouteDef(AbstractRouteDef):
60    method: str
61    path: str
62    handler: _HandlerType
63    kwargs: Dict[str, Any]
64
65    def __repr__(self) -> str:
66        info = []
67        for name, value in sorted(self.kwargs.items()):
68            info.append(f", {name}={value!r}")
69        return "<RouteDef {method} {path} -> {handler.__name__!r}{info}>".format(
70            method=self.method, path=self.path, handler=self.handler, info="".join(info)
71        )
72
73    def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
74        if self.method in hdrs.METH_ALL:
75            reg = getattr(router, "add_" + self.method.lower())
76            return [reg(self.path, self.handler, **self.kwargs)]
77        else:
78            return [
79                router.add_route(self.method, self.path, self.handler, **self.kwargs)
80            ]
81
82
83@attr.s(auto_attribs=True, frozen=True, repr=False, slots=True)
84class StaticDef(AbstractRouteDef):
85    prefix: str
86    path: PathLike
87    kwargs: Dict[str, Any]
88
89    def __repr__(self) -> str:
90        info = []
91        for name, value in sorted(self.kwargs.items()):
92            info.append(f", {name}={value!r}")
93        return "<StaticDef {prefix} -> {path}{info}>".format(
94            prefix=self.prefix, path=self.path, info="".join(info)
95        )
96
97    def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
98        resource = router.add_static(self.prefix, self.path, **self.kwargs)
99        routes = resource.get_info().get("routes", {})
100        return list(routes.values())
101
102
103def route(method: str, path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
104    return RouteDef(method, path, handler, kwargs)
105
106
107def head(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
108    return route(hdrs.METH_HEAD, path, handler, **kwargs)
109
110
111def options(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
112    return route(hdrs.METH_OPTIONS, path, handler, **kwargs)
113
114
115def get(
116    path: str,
117    handler: _HandlerType,
118    *,
119    name: Optional[str] = None,
120    allow_head: bool = True,
121    **kwargs: Any,
122) -> RouteDef:
123    return route(
124        hdrs.METH_GET, path, handler, name=name, allow_head=allow_head, **kwargs
125    )
126
127
128def post(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
129    return route(hdrs.METH_POST, path, handler, **kwargs)
130
131
132def put(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
133    return route(hdrs.METH_PUT, path, handler, **kwargs)
134
135
136def patch(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
137    return route(hdrs.METH_PATCH, path, handler, **kwargs)
138
139
140def delete(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
141    return route(hdrs.METH_DELETE, path, handler, **kwargs)
142
143
144def view(path: str, handler: Type[AbstractView], **kwargs: Any) -> RouteDef:
145    return route(hdrs.METH_ANY, path, handler, **kwargs)
146
147
148def static(prefix: str, path: PathLike, **kwargs: Any) -> StaticDef:
149    return StaticDef(prefix, path, kwargs)
150
151
152_Deco = Callable[[_HandlerType], _HandlerType]
153
154
155class RouteTableDef(Sequence[AbstractRouteDef]):
156    """Route definition table"""
157
158    def __init__(self) -> None:
159        self._items: List[AbstractRouteDef] = []
160
161    def __repr__(self) -> str:
162        return f"<RouteTableDef count={len(self._items)}>"
163
164    @overload
165    def __getitem__(self, index: int) -> AbstractRouteDef: ...
166
167    @overload
168    def __getitem__(self, index: slice) -> List[AbstractRouteDef]: ...
169
170    def __getitem__(self, index):  # type: ignore[no-untyped-def]
171        return self._items[index]
172
173    def __iter__(self) -> Iterator[AbstractRouteDef]:
174        return iter(self._items)
175
176    def __len__(self) -> int:
177        return len(self._items)
178
179    def __contains__(self, item: object) -> bool:
180        return item in self._items
181
182    def route(self, method: str, path: str, **kwargs: Any) -> _Deco:
183        def inner(handler: _HandlerType) -> _HandlerType:
184            self._items.append(RouteDef(method, path, handler, kwargs))
185            return handler
186
187        return inner
188
189    def head(self, path: str, **kwargs: Any) -> _Deco:
190        return self.route(hdrs.METH_HEAD, path, **kwargs)
191
192    def get(self, path: str, **kwargs: Any) -> _Deco:
193        return self.route(hdrs.METH_GET, path, **kwargs)
194
195    def post(self, path: str, **kwargs: Any) -> _Deco:
196        return self.route(hdrs.METH_POST, path, **kwargs)
197
198    def put(self, path: str, **kwargs: Any) -> _Deco:
199        return self.route(hdrs.METH_PUT, path, **kwargs)
200
201    def patch(self, path: str, **kwargs: Any) -> _Deco:
202        return self.route(hdrs.METH_PATCH, path, **kwargs)
203
204    def delete(self, path: str, **kwargs: Any) -> _Deco:
205        return self.route(hdrs.METH_DELETE, path, **kwargs)
206
207    def options(self, path: str, **kwargs: Any) -> _Deco:
208        return self.route(hdrs.METH_OPTIONS, path, **kwargs)
209
210    def view(self, path: str, **kwargs: Any) -> _Deco:
211        return self.route(hdrs.METH_ANY, path, **kwargs)
212
213    def static(self, prefix: str, path: PathLike, **kwargs: Any) -> None:
214        self._items.append(StaticDef(prefix, path, kwargs))
215 
codekingpro/portable-devtools · Team Ai