web_routedef.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215
  1. import abc
  2. import os # noqa
  3. from typing import (
  4. TYPE_CHECKING,
  5. Any,
  6. Awaitable,
  7. Callable,
  8. Dict,
  9. Iterator,
  10. List,
  11. Optional,
  12. Sequence,
  13. Type,
  14. Union,
  15. overload,
  16. )
  17. import attr
  18. from . import hdrs
  19. from .abc import AbstractView
  20. from .typedefs import PathLike
  21. if TYPE_CHECKING: # pragma: no cover
  22. from .web_request import Request
  23. from .web_response import StreamResponse
  24. from .web_urldispatcher import AbstractRoute, UrlDispatcher
  25. else:
  26. Request = StreamResponse = UrlDispatcher = AbstractRoute = None
  27. __all__ = (
  28. "AbstractRouteDef",
  29. "RouteDef",
  30. "StaticDef",
  31. "RouteTableDef",
  32. "head",
  33. "options",
  34. "get",
  35. "post",
  36. "patch",
  37. "put",
  38. "delete",
  39. "route",
  40. "view",
  41. "static",
  42. )
  43. class AbstractRouteDef(abc.ABC):
  44. @abc.abstractmethod
  45. def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
  46. pass # pragma: no cover
  47. _SimpleHandler = Callable[[Request], Awaitable[StreamResponse]]
  48. _HandlerType = Union[Type[AbstractView], _SimpleHandler]
  49. @attr.s(auto_attribs=True, frozen=True, repr=False, slots=True)
  50. class RouteDef(AbstractRouteDef):
  51. method: str
  52. path: str
  53. handler: _HandlerType
  54. kwargs: Dict[str, Any]
  55. def __repr__(self) -> str:
  56. info = []
  57. for name, value in sorted(self.kwargs.items()):
  58. info.append(f", {name}={value!r}")
  59. return "<RouteDef {method} {path} -> {handler.__name__!r}" "{info}>".format(
  60. method=self.method, path=self.path, handler=self.handler, info="".join(info)
  61. )
  62. def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
  63. if self.method in hdrs.METH_ALL:
  64. reg = getattr(router, "add_" + self.method.lower())
  65. return [reg(self.path, self.handler, **self.kwargs)]
  66. else:
  67. return [
  68. router.add_route(self.method, self.path, self.handler, **self.kwargs)
  69. ]
  70. @attr.s(auto_attribs=True, frozen=True, repr=False, slots=True)
  71. class StaticDef(AbstractRouteDef):
  72. prefix: str
  73. path: PathLike
  74. kwargs: Dict[str, Any]
  75. def __repr__(self) -> str:
  76. info = []
  77. for name, value in sorted(self.kwargs.items()):
  78. info.append(f", {name}={value!r}")
  79. return "<StaticDef {prefix} -> {path}" "{info}>".format(
  80. prefix=self.prefix, path=self.path, info="".join(info)
  81. )
  82. def register(self, router: UrlDispatcher) -> List[AbstractRoute]:
  83. resource = router.add_static(self.prefix, self.path, **self.kwargs)
  84. routes = resource.get_info().get("routes", {})
  85. return list(routes.values())
  86. def route(method: str, path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
  87. return RouteDef(method, path, handler, kwargs)
  88. def head(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
  89. return route(hdrs.METH_HEAD, path, handler, **kwargs)
  90. def options(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
  91. return route(hdrs.METH_OPTIONS, path, handler, **kwargs)
  92. def get(
  93. path: str,
  94. handler: _HandlerType,
  95. *,
  96. name: Optional[str] = None,
  97. allow_head: bool = True,
  98. **kwargs: Any,
  99. ) -> RouteDef:
  100. return route(
  101. hdrs.METH_GET, path, handler, name=name, allow_head=allow_head, **kwargs
  102. )
  103. def post(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
  104. return route(hdrs.METH_POST, path, handler, **kwargs)
  105. def put(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
  106. return route(hdrs.METH_PUT, path, handler, **kwargs)
  107. def patch(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
  108. return route(hdrs.METH_PATCH, path, handler, **kwargs)
  109. def delete(path: str, handler: _HandlerType, **kwargs: Any) -> RouteDef:
  110. return route(hdrs.METH_DELETE, path, handler, **kwargs)
  111. def view(path: str, handler: Type[AbstractView], **kwargs: Any) -> RouteDef:
  112. return route(hdrs.METH_ANY, path, handler, **kwargs)
  113. def static(prefix: str, path: PathLike, **kwargs: Any) -> StaticDef:
  114. return StaticDef(prefix, path, kwargs)
  115. _Deco = Callable[[_HandlerType], _HandlerType]
  116. class RouteTableDef(Sequence[AbstractRouteDef]):
  117. """Route definition table"""
  118. def __init__(self) -> None:
  119. self._items = [] # type: List[AbstractRouteDef]
  120. def __repr__(self) -> str:
  121. return "<RouteTableDef count={}>".format(len(self._items))
  122. @overload
  123. def __getitem__(self, index: int) -> AbstractRouteDef:
  124. ...
  125. @overload
  126. def __getitem__(self, index: slice) -> List[AbstractRouteDef]:
  127. ...
  128. def __getitem__(self, index): # type: ignore
  129. return self._items[index]
  130. def __iter__(self) -> Iterator[AbstractRouteDef]:
  131. return iter(self._items)
  132. def __len__(self) -> int:
  133. return len(self._items)
  134. def __contains__(self, item: object) -> bool:
  135. return item in self._items
  136. def route(self, method: str, path: str, **kwargs: Any) -> _Deco:
  137. def inner(handler: _HandlerType) -> _HandlerType:
  138. self._items.append(RouteDef(method, path, handler, kwargs))
  139. return handler
  140. return inner
  141. def head(self, path: str, **kwargs: Any) -> _Deco:
  142. return self.route(hdrs.METH_HEAD, path, **kwargs)
  143. def get(self, path: str, **kwargs: Any) -> _Deco:
  144. return self.route(hdrs.METH_GET, path, **kwargs)
  145. def post(self, path: str, **kwargs: Any) -> _Deco:
  146. return self.route(hdrs.METH_POST, path, **kwargs)
  147. def put(self, path: str, **kwargs: Any) -> _Deco:
  148. return self.route(hdrs.METH_PUT, path, **kwargs)
  149. def patch(self, path: str, **kwargs: Any) -> _Deco:
  150. return self.route(hdrs.METH_PATCH, path, **kwargs)
  151. def delete(self, path: str, **kwargs: Any) -> _Deco:
  152. return self.route(hdrs.METH_DELETE, path, **kwargs)
  153. def view(self, path: str, **kwargs: Any) -> _Deco:
  154. return self.route(hdrs.METH_ANY, path, **kwargs)
  155. def static(self, prefix: str, path: PathLike, **kwargs: Any) -> None:
  156. self._items.append(StaticDef(prefix, path, kwargs))