78 lines
2.1 KiB
Python
78 lines
2.1 KiB
Python
"""Store routes info."""
|
|
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from logging import getLogger
|
|
|
|
from aiohttp.web import AbstractRoute, Response
|
|
|
|
_LOGGER = getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class Route:
|
|
"""Store route info."""
|
|
|
|
url_path: str
|
|
route: AbstractRoute
|
|
handler: Callable
|
|
enabled: bool = False
|
|
|
|
def __str__(self):
|
|
"""Return string representation."""
|
|
return f"{self.url_path} -> {self.handler}"
|
|
|
|
|
|
class Routes:
|
|
"""Store routes info."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize routes."""
|
|
self.routes = {}
|
|
|
|
def switch_route(self, coordinator: Callable, url_path: str):
|
|
"""Switch route."""
|
|
|
|
for url, route in self.routes.items():
|
|
if url == url_path:
|
|
_LOGGER.info("New coordinator to route: %s", route.url_path)
|
|
route.enabled = True
|
|
route.handler = coordinator
|
|
route.route._handler = coordinator # noqa: SLF001
|
|
else:
|
|
route.enabled = False
|
|
route.handler = unregistred
|
|
route.route._handler = unregistred # noqa: SLF001
|
|
|
|
def add_route(
|
|
self,
|
|
url_path: str,
|
|
route: AbstractRoute,
|
|
handler: Callable,
|
|
enabled: bool = False,
|
|
):
|
|
"""Add route."""
|
|
self.routes[url_path] = Route(url_path, route, handler, enabled)
|
|
|
|
def get_route(self, url_path: str) -> Route:
|
|
"""Get route."""
|
|
return self.routes.get(url_path, Route)
|
|
|
|
def get_enabled(self) -> str:
|
|
"""Get enabled routes."""
|
|
enabled_routes = [
|
|
route.url_path for route in self.routes.values() if route.enabled
|
|
]
|
|
return "".join(enabled_routes) if enabled_routes else "None"
|
|
|
|
def __str__(self):
|
|
"""Return string representation."""
|
|
return "\n".join([str(route) for route in self.routes.values()])
|
|
|
|
|
|
async def unregistred(*args, **kwargs):
|
|
"""Unregister path to handle incoming data."""
|
|
|
|
_LOGGER.error("Recieved data to unregistred webhook. Check your settings")
|
|
return Response(body=f"{'Unregistred webhook.'}", status=404)
|