68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
"""Routes implementation."""
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass, field
|
|
import logging
|
|
|
|
from aiohttp.web import Request, Response
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
Handler = Callable[[Request], Awaitable[Response]]
|
|
|
|
|
|
@dataclass
|
|
class RouteInfo:
|
|
"""Route struct."""
|
|
|
|
url_path: str
|
|
handler: Handler
|
|
enabled: bool = False
|
|
fallback: Handler = field(default_factory=lambda: unregistred)
|
|
|
|
|
|
class Routes:
|
|
"""Routes class."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Init."""
|
|
self.routes: dict[str, RouteInfo] = {}
|
|
|
|
async def dispatch(self, request: Request) -> Response:
|
|
"""Dispatch."""
|
|
info = self.routes.get(request.path)
|
|
if not info:
|
|
_LOGGER.debug("Route %s is not registered!")
|
|
return await unregistred(request)
|
|
handler = info.handler if info.enabled else info.fallback
|
|
return await handler(request)
|
|
|
|
def switch_route(self, url_path: str) -> None:
|
|
"""Switch route to new handler."""
|
|
for path, info in self.routes.items():
|
|
info.enabled = path == url_path
|
|
|
|
def add_route(
|
|
self, url_path: str, handler: Handler, *, enabled: bool = False
|
|
) -> None:
|
|
"""Add route to dispatcher."""
|
|
|
|
self.routes[url_path] = RouteInfo(url_path, handler, enabled=enabled)
|
|
_LOGGER.debug("Registered dispatcher for route %s", url_path)
|
|
|
|
def show_enabled(self) -> str:
|
|
"""Show info of enabled route."""
|
|
for url, route in self.routes.items():
|
|
if route.enabled:
|
|
return (
|
|
f"Dispatcher enabled for URL: {url}, with handler: {route.handler}"
|
|
)
|
|
return "No routes is enabled."
|
|
|
|
|
|
async def unregistred(request: Request) -> Response:
|
|
"""Return unregistred error."""
|
|
_ = request
|
|
_LOGGER.debug("Received data to unregistred or disabled webhook.")
|
|
return Response(text="Unregistred webhook. Check your settings.", status=400)
|