from __future__ import annotations from typing import TYPE_CHECKING, Any, Sequence, Type from .errors import NotFound from .handler import APIHandler, api_handler from .middleware import ApiErrorMiddleware, LoggingMiddleware, MatchInfoMiddleware, Middleware, \ SchemaValidationMiddleware, ServerErrorMiddleware from .request import Request from .response import Response from .route import Route if TYPE_CHECKING: from src.logger import BoundLogger __all__ = ["Router"] @api_handler def _route_not_found(request: Request, **_kwargs): raise NotFound(f"No matching route found for {request.method} {request.path}") class Router: def __init__( self, routes: Sequence[Route], *, middlewares: Sequence[Type[Middleware]] = ( LoggingMiddleware, ServerErrorMiddleware, ApiErrorMiddleware, SchemaValidationMiddleware, MatchInfoMiddleware, ), logger: "BoundLogger", debug: bool = False, ): self._routes = routes self._middlewares = middlewares self._logger = logger self._debug = debug def _handle_request(self, request: Request) -> Response: for route in self._routes: match_info = route.match(request.method, request.path) if match_info is None: continue handler: APIHandler | Middleware = route.handler break else: handler = _route_not_found match_info = {} for middleware_cls in reversed(self._middlewares): handler = middleware_cls(handler, self._debug) return handler(request, logger=self._logger, **match_info) def handle_event(self, event: dict[str, Any]) -> dict[str, Any]: request = Request(event) return self._handle_request(request).to_dict()