from __future__ import annotations
-from collections.abc import Coroutine
+from collections.abc import Callable, Coroutine
from typing import TYPE_CHECKING, Any, Final
from aiohttp import web
if TYPE_CHECKING:
import logging
- from collections.abc import Callable
from aiohttp.typedefs import Handler
MAX_CLIENT_SIZE: Final = 1024**2 * 16
MAX_LINE_SIZE: Final = 24570
+# Type alias for dynamic route handlers
+DynamicRouteHandler = Callable[
+ [web.Request], Coroutine[Any, Any, web.Response | web.StreamResponse]
+]
+
class Webserver:
"""Base Webserver logic for an HTTPServer that can handle dynamic routes."""
self._webapp: web.Application | None = None
self._tcp_site: web.TCPSite | None = None
self._static_routes: list[tuple[str, str, Handler]] | None = None
- self._dynamic_routes: dict[str, Callable] | None = {} if enable_dynamic_routes else None
+ self._dynamic_routes: dict[str, DynamicRouteHandler] | None = (
+ {} if enable_dynamic_routes else None
+ )
self._bind_port: int | None = None
self._ingress_tcp_site: web.TCPSite | None = None
path: str,
handler: Callable[[web.Request], Coroutine[Any, Any, web.Response | web.StreamResponse]],
method: str = "*",
- ) -> Callable:
+ ) -> Callable[[], None]:
"""Register a dynamic route on the webserver, returns handler to unregister."""
if self._dynamic_routes is None:
msg = "Dynamic routes are not enabled"
raise RuntimeError(msg)
self._dynamic_routes[key] = handler
- def _remove():
+ def _remove() -> None:
assert self._dynamic_routes is not None # for type checking
- return self._dynamic_routes.pop(key)
+ self._dynamic_routes.pop(key, None)
return _remove
headers = {"Cache-Control": "no-cache"}
return web.FileResponse(file_path, headers=headers)
- async def _handle_catch_all(self, request: web.Request) -> web.Response:
+ async def _handle_catch_all(self, request: web.Request) -> web.Response | web.StreamResponse:
"""Redirect request to correct destination."""
# find handler for the request
# Try exact match first
exclude = [
'^music_assistant/controllers/.*$',
'^music_assistant/helpers/app_vars.py',
- '^music_assistant/helpers/webserver.py',
'^music_assistant/models/player_provider.py',
'^music_assistant/providers/apple_music/.*$',
'^music_assistant/providers/bluesound/.*$',