src/ctp_slack_bot/controllers/base.py

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