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")