Spaces:
Runtime error
Runtime error
File size: 2,818 Bytes
f0fe0fd a1a6d79 f0fe0fd a1a6d79 f0fe0fd a1a6d79 f0fe0fd a1a6d79 f0fe0fd a1a6d79 f0fe0fd a1a6d79 f0fe0fd a1a6d79 f0fe0fd a1a6d79 f0fe0fd |
1 2 3 4 5 6 7 8 9 10 11 12 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 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 |
from abc import abstractmethod
from aiohttp.web import Request, Response
from functools import partial, wraps
from importlib import import_module
from inspect import getmembers, ismethod
from pydantic import BaseModel, ConfigDict
from typing import Awaitable, Callable, ClassVar, Collection, Mapping, Optional, overload, ParamSpec, Self, Sequence, TypeVar
from ctp_slack_bot.core import ApplicationComponentBase
AsyncHandler = Callable[[Request], Awaitable[Response]]
class Route(BaseModel):
model_config = ConfigDict(frozen=True)
method: str
path: str
handler: AsyncHandler
class ControllerBase(ApplicationComponentBase):
def get_routes(self: Self) -> Sequence[Route]:
return tuple(Route(method=method._http_method,
path="/".join(filter(None, (self.prefix, method._http_path))),
handler=method)
for name, method in getmembers(self, predicate=ismethod)
if name != 'get_routes' and name != 'prefix' and hasattr(method, "_http_method") and hasattr(method, "_http_path"))
@property
@abstractmethod
def prefix(self: Self) -> str:
pass
T = TypeVar('T', bound=ControllerBase)
class ControllerRegistry:
__registry: ClassVar[list[T]] = []
@classmethod
def get_registry(cls) -> Collection[T]:
import_module(__package__)
return tuple(cls.__registry)
@classmethod
def register(cls, controller_cls: T) -> None:
cls.__registry.append(controller_cls)
@overload
def controller(cls: T) -> T: ...
@overload
def controller(prefix: str = "/") -> Callable[[T], T]: ...
def controller(cls_or_prefix=None):
def implement_prefix_property_and_register_controller(cls: T, prefix: Optional[str] = "/") -> T:
def prefix_getter(self: T) -> str:
return prefix
setattr(cls, 'prefix', property(prefix_getter))
if hasattr(cls, '__abstractmethods__'):
cls.__abstractmethods__ = frozenset(method for method in cls.__abstractmethods__ if method != 'prefix')
ControllerRegistry.register(cls)
return cls
if isinstance(cls_or_prefix, type):
return implement_prefix_property_and_register_controller(cls_or_prefix)
def decorator(cls: T) -> T:
return implement_prefix_property_and_register_controller(cls, cls_or_prefix)
return decorator
def route(method: str, path: str = "") -> Callable[[AsyncHandler], AsyncHandler]:
def decorator(function: AsyncHandler) -> AsyncHandler:
function._http_method = method
function._http_path = path
return function
return decorator
get = partial(route, "GET")
post = partial(route, "POST")
put = partial(route, "PUT")
delete = partial(route, "DELETE")
patch = partial(route, "PATCH")
|