diff --git a/docs/en/docs/release-notes.md b/docs/en/docs/release-notes.md index 5d8cc18..915b193 100644 --- a/docs/en/docs/release-notes.md +++ b/docs/en/docs/release-notes.md @@ -10,6 +10,9 @@ hide: ### Added - Add `ReceiveSendSniffer`. This sniffer allows to detect communication events and to replay receive messages. +- `Include` and `BaseLilya` (application) have now a ClassVar `router_class` to provide a custom router. +- Subclasses of `BaseLilya` (application) can set the `router_class` to None to provide a completely custom router + which initialization parameters aren't required to match the ones of `Router`. - Expose `fall_through` on `StaticFile`. ### Changed diff --git a/lilya/apps.py b/lilya/apps.py index a8c27f6..7476fc7 100644 --- a/lilya/apps.py +++ b/lilya/apps.py @@ -3,7 +3,7 @@ import sys from collections.abc import Awaitable, Callable, Mapping, Sequence from functools import cached_property -from typing import Annotated, Any, cast +from typing import Annotated, Any, ClassVar, cast from lilya._internal._module_loading import import_string from lilya._utils import is_class_and_subclass @@ -44,6 +44,9 @@ class BaseLilya: + router_class: ClassVar[type[Router] | None] = Router + router: Router + def __init__( self, debug: Annotated[bool, Doc("Enable or disable debug mode. Defaults to False.")] = False, @@ -392,16 +395,17 @@ async def create_user(request: Request): self.state = State() self.middleware_stack: ASGIApp | None = None - self.router: Router = Router( - routes=routes, - redirect_slashes=redirect_slashes, - permissions=self.custom_permissions, - on_startup=on_startup, - on_shutdown=on_shutdown, - lifespan=lifespan, - include_in_schema=include_in_schema, - settings_module=self.settings, - ) + if self.router_class is not None: + self.router = self.router_class( + routes=routes, + redirect_slashes=redirect_slashes, + permissions=self.custom_permissions, + on_startup=on_startup, + on_shutdown=on_shutdown, + lifespan=lifespan, + include_in_schema=include_in_schema, + settings_module=self.settings, + ) self.__set_settings_app(self.settings, self) @property diff --git a/lilya/routing.py b/lilya/routing.py index 87bc8c3..351a055 100644 --- a/lilya/routing.py +++ b/lilya/routing.py @@ -5,7 +5,7 @@ import re import traceback from collections.abc import Awaitable, Callable, Mapping, Sequence -from typing import Annotated, Any, TypeVar, cast +from typing import Annotated, Any, ClassVar, TypeVar, cast from lilya import status from lilya._internal._events import AsyncLifespan, handle_lifespan_events @@ -708,93 +708,34 @@ def __repr__(self) -> str: return f"{self.__class__.__name__}(path={self.path!r}, name={self.name!r})" -class Include(BasePath): +class Host(BasePath): __slots__ = ( - "path", + "host", "app", - "namespace", - "pattern", + "__base_app__", "name", - "exception_handlers", "middleware", "permissions", "exception_handlers", - "deprecated", ) def __init__( self, - path: str, - app: ASGIApp | str | None = None, - routes: Sequence[BasePath] | None = None, - namespace: str | None = None, - pattern: str | None = None, + host: str, + app: ASGIApp, name: str | None = None, *, middleware: Sequence[DefineMiddleware] | None = None, permissions: Sequence[DefinePermission] | None = None, exception_handlers: Mapping[Any, ExceptionHandler] | None = None, - include_in_schema: bool = True, - deprecated: bool = False, - redirect_slashes: bool = True, ) -> None: - """ - Initialize the router with specified parameters. - - Args: - path (str): The path associated with the router. - app (Union[ASGIApp, str, None]): The ASGI app. - routes (Union[Sequence[BasePath], None]): The routes. - namespace (Union[str, None]): The namespace. - pattern (Union[str, None]): The pattern. - name (Union[str, None]): The name. - middleware (Union[Sequence[DefineMiddleware], None]): The middleware. - permissions (Union[Sequence[DefinePermission], None]): The permissions. - include_in_schema (bool): Flag to include in the schema. - redirect_slashes (bool): (Only namespace or routes) Redirect slashes on mismatch. - - Returns: - None - """ - assert path == "" or path.startswith("/"), "Routed paths must start with '/'" - assert ( - app is not None or routes is not None or namespace is not None - ), "Either 'app=...', or 'routes=...', or 'namespace=...' must be specified" - self.path = clean_path(path) - - assert ( - namespace is None or routes is None - ), "Either 'namespace=...' or 'routes=', not both." - - if namespace and not isinstance(namespace, str): - raise ImproperlyConfigured("Namespace must be a string. Example: 'myapp.routes'.") - - if pattern and not isinstance(pattern, str): - raise ImproperlyConfigured("Pattern must be a string. Example: 'route_patterns'.") - - if pattern and routes: - raise ImproperlyConfigured("Pattern must be used only with namespace.") - - if namespace is not None: - routes = include(namespace, pattern) - - self.__base_app__: ASGIApp | Router - if isinstance(app, str): - self.__base_app__ = import_string(app) - self.handle_not_found = self.handle_not_found_fallthrough # type: ignore - elif app is not None: - self.handle_not_found = self.handle_not_found_fallthrough # type: ignore - self.__base_app__ = app - else: - self.__base_app__ = Router( - routes=routes, - default=self.handle_not_found_fallthrough, - is_sub_router=True, - redirect_slashes=redirect_slashes, - ) - - self.app = self.__base_app__ - + assert not host.startswith("/"), "Host must not start with '/'" + self.host = host + self.app = self.__base_app__ = app + self.name = name + self.host_regex, self.host_format, self.param_convertors, self.path_start = compile_path( + host + ) self.middleware = middleware if middleware is not None else [] self.permissions = permissions if permissions is not None else [] self.exception_handlers = {} if exception_handlers is None else dict(exception_handlers) @@ -802,14 +743,6 @@ def __init__( self._apply_middleware(middleware) self._apply_permissions(permissions) - self.name = name - self.include_in_schema = include_in_schema - self.deprecated = deprecated - - self.path_regex, self.path_format, self.param_convertors, self.path_start = compile_path( - clean_path(self.path + "/{path:path}") - ) - def _apply_middleware(self, middleware: Sequence[DefineMiddleware] | None) -> None: """ Apply middleware to the app. @@ -858,18 +791,15 @@ def search(self, scope: Scope) -> tuple[Match, Scope]: Tuple[Match, Scope]: The match result and child scope. """ if scope["type"] in {ScopeType.HTTP, ScopeType.WEBSOCKET}: - root_path = scope.get("root_path", "") - route_path = get_route_path(scope) - match = self.path_regex.match(route_path) - + headers = Header.from_scope(scope=scope) + host = headers.get("host", "").split(":")[0] + match = self.host_regex.match(host) if match: - return self.handle_match(scope, match, route_path, root_path) + return self.handle_match(scope, match) return Match.NONE, {} - def handle_match( - self, scope: Scope, match: re.Match, route_path: str, root_path: str - ) -> tuple[Match, Scope]: + def handle_match(self, scope: Scope, match: re.Match) -> tuple[Match, Scope]: """ Handles the case when a match is found in the route patterns. @@ -884,14 +814,9 @@ def handle_match( key: self.param_convertors[key].transform(value) for key, value in match.groupdict().items() } - remaining_path = f'/{matched_params.pop("path", "")}' - matched_path = route_path[: -len(remaining_path)] - path_params = {**scope.get("path_params", {}), **matched_params} child_scope = { "path_params": path_params, - "app_root_path": scope.get("app_root_path", root_path), - "root_path": root_path + matched_path, "handler": self.app, } return Match.FULL, child_scope @@ -927,20 +852,36 @@ def path_for(self, name: str, /, **path_params: Any) -> URLPath: Raises: NoMatchFound: If no matching route is found for the given name and parameters. """ - if self.name is not None and name == self.name and "path" in path_params: - path_params["path"] = path_params["path"].lstrip("/") - path, remaining_params = replace_params( - self.path_format, self.param_convertors, path_params - ) - if not remaining_params: - return URLPath(path=path) + return self.path_for_with_name(path_params) elif self.name is None or name.startswith(self.name + ":"): - return self._path_for_without_name(name, path_params) + return self.path_for_without_name(name, path_params) + else: + raise NoMatchFound(name, path_params) - raise NoMatchFound(name, path_params) + def path_for_with_name(self, path_params: dict) -> URLPath: + """ + Generate a URLPath for a route with a specific name and path parameters. - def _path_for_without_name(self, name: str, path_params: dict) -> URLPath: + Args: + path_params: Path parameters for route substitution. + + Returns: + URLPath: The generated URLPath. + + Raises: + NoMatchFound: If no matching route is found for the given parameters. + """ + path = path_params.pop("path") + host, remaining_params = replace_params( + self.host_format, self.param_convertors, path_params, is_host=True + ) + if not remaining_params: + return URLPath(path=path, host=host) + + raise NoMatchFound(self.name, path_params) + + def path_for_without_name(self, name: str, path_params: dict) -> URLPath: """ Generate a URLPath for a route without a specific name and with path parameters. @@ -959,19 +900,14 @@ def _path_for_without_name(self, name: str, path_params: dict) -> URLPath: else: remaining_name = name[len(self.name) + 1 :] - path_kwarg = path_params.get("path") - path_params["path"] = "" - path_prefix, remaining_params = replace_params( - self.path_format, self.param_convertors, path_params + host, remaining_params = replace_params( + self.host_format, self.param_convertors, path_params, is_host=True ) - if path_kwarg is not None: - remaining_params["path"] = path_kwarg - for route in self.routes or []: try: url = route.path_for(remaining_name, **remaining_params) - return URLPath(path=path_prefix.rstrip("/") + str(url), protocol=url.protocol) + return URLPath(path=str(url), protocol=url.protocol, host=host) except NoMatchFound: pass @@ -979,43 +915,94 @@ def _path_for_without_name(self, name: str, path_params: dict) -> URLPath: def __repr__(self) -> str: name = self.name or "" - return f"{self.__class__.__name__}(path={self.path!r}, name={name!r}, app={self.app!r})" + return f"{self.__class__.__name__}(host={self.host!r}, name={name!r}, app={self.app!r})" -class Host(BasePath): +class BaseRouter: + """ + A Lilya base router object. + """ + __slots__ = ( - "host", - "app", - "__base_app__", - "name", + "routes", + "redirect_slashes", + "default", + "on_startup", + "on_shutdown", "middleware", "permissions", - "exception_handlers", + "include_in_schema", + "deprecated", + "lifespan_context", + "middleware_stack", + "permission_started", + "settings_module", + "is_sub_router", ) def __init__( self, - host: str, - app: ASGIApp, - name: str | None = None, + routes: Sequence[BasePath] | None = None, + redirect_slashes: bool = True, + default: ASGIApp | None = None, + on_startup: Sequence[Callable[[], Any]] | None = None, + on_shutdown: Sequence[Callable[[], Any]] | None = None, + lifespan: Lifespan[Any] | None = None, *, middleware: Sequence[DefineMiddleware] | None = None, permissions: Sequence[DefinePermission] | None = None, - exception_handlers: Mapping[Any, ExceptionHandler] | None = None, + settings_module: Annotated[ + Settings | None, + Doc( + """ + Alternative settings parameter. This parameter is an alternative to + `LILYA_SETTINGS_MODULE` way of loading your settings into a Lilya application. + + When the `settings_module` is provided, it will make sure it takes priority over + any other settings provided for the instance. + """ + ), + ] = None, + include_in_schema: bool = True, + deprecated: bool = False, + is_sub_router: bool = False, ) -> None: - assert not host.startswith("/"), "Host must not start with '/'" - self.host = host - self.app = self.__base_app__ = app - self.name = name - self.host_regex, self.host_format, self.param_convertors, self.path_start = compile_path( - host + assert lifespan is None or ( + on_startup is None and on_shutdown is None + ), "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both." + + if inspect.isasyncgenfunction(lifespan) or inspect.isgeneratorfunction(lifespan): + raise ImproperlyConfigured( + "async function generators are not allowed. " + "Use @contextlib.asynccontextmanager instead." + ) + + self.on_startup = [] if on_startup is None else list(on_startup) + self.on_shutdown = [] if on_shutdown is None else list(on_shutdown) + + self.lifespan_context = handle_lifespan_events( + on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan ) + + if self.lifespan_context is None: + self.lifespan_context = AsyncLifespan(self) + + self.routes = [] if routes is None else list(routes) + self.redirect_slashes = redirect_slashes + self.default = self.handle_not_found if default is None else default + self.include_in_schema = include_in_schema + self.deprecated = deprecated + self.middleware = middleware if middleware is not None else [] self.permissions = permissions if permissions is not None else [] - self.exception_handlers = {} if exception_handlers is None else dict(exception_handlers) + self.settings_module = settings_module + self.middleware_stack = self.app + self.permission_started = False + self.is_sub_router = is_sub_router - self._apply_middleware(middleware) - self._apply_permissions(permissions) + self._apply_middleware(self.middleware) + self._apply_permissions(self.permissions) + self._set_settings_app(self.settings_module, self) def _apply_middleware(self, middleware: Sequence[DefineMiddleware] | None) -> None: """ @@ -1029,7 +1016,7 @@ def _apply_middleware(self, middleware: Sequence[DefineMiddleware] | None) -> No """ if middleware is not None: for cls, args, options in reversed(middleware): - self.app = cls(app=self.app, *args, **options) + self.middleware_stack = cls(app=self.middleware_stack, *args, **options) def _apply_permissions(self, permissions: Sequence[DefinePermission] | None) -> None: """ @@ -1042,271 +1029,10 @@ def _apply_permissions(self, permissions: Sequence[DefinePermission] | None) -> None """ if permissions is not None: - for cls, args, options in reversed(permissions): - self.app = cls(app=self.app, *args, **options) + for cls, args, options in reversed(self.permissions): + self.middleware_stack = cls(app=self.middleware_stack, *args, **options) - @property - def routes(self) -> list[BasePath]: - """ - Returns a list of declared path objects. - """ - return getattr(self.__base_app__, "routes", []) - - def search(self, scope: Scope) -> tuple[Match, Scope]: - """ - Searches within the route patterns and matches against the regex. - - If found, then dispatches the request to the handler of the object. - - Args: - scope (Scope): The request scope. - - Returns: - Tuple[Match, Scope]: The match result and child scope. - """ - if scope["type"] in {ScopeType.HTTP, ScopeType.WEBSOCKET}: - headers = Header.from_scope(scope=scope) - host = headers.get("host", "").split(":")[0] - match = self.host_regex.match(host) - if match: - return self.handle_match(scope, match) - - return Match.NONE, {} - - def handle_match(self, scope: Scope, match: re.Match) -> tuple[Match, Scope]: - """ - Handles the case when a match is found in the route patterns. - - Args: - scope (Scope): The request scope. - match: The match object from the regex. - - Returns: - Tuple[Match, Scope]: The match result and child scope. - """ - matched_params = { - key: self.param_convertors[key].transform(value) - for key, value in match.groupdict().items() - } - path_params = {**scope.get("path_params", {}), **matched_params} - child_scope = { - "path_params": path_params, - "handler": self.app, - } - return Match.FULL, child_scope - - async def handle_dispatch(self, scope: Scope, receive: Receive, send: Send) -> None: - """ - Handles the dispatch of the request to the appropriate handler. - - Args: - scope (Scope): The request scope. - receive (Receive): The receive channel. - send (Send): The send channel. - - Returns: - None - """ - try: - await self.app(scope, receive, send) - except Exception as ex: - await self.handle_exception_handlers(scope, receive, send, ex) - - def path_for(self, name: str, /, **path_params: Any) -> URLPath: - """ - Generate a URLPath for a given route name and path parameters. - - Args: - name (str): The name of the route. - path_params: Path parameters for route substitution. - - Returns: - URLPath: The generated URLPath. - - Raises: - NoMatchFound: If no matching route is found for the given name and parameters. - """ - if self.name is not None and name == self.name and "path" in path_params: - return self.path_for_with_name(path_params) - elif self.name is None or name.startswith(self.name + ":"): - return self.path_for_without_name(name, path_params) - else: - raise NoMatchFound(name, path_params) - - def path_for_with_name(self, path_params: dict) -> URLPath: - """ - Generate a URLPath for a route with a specific name and path parameters. - - Args: - path_params: Path parameters for route substitution. - - Returns: - URLPath: The generated URLPath. - - Raises: - NoMatchFound: If no matching route is found for the given parameters. - """ - path = path_params.pop("path") - host, remaining_params = replace_params( - self.host_format, self.param_convertors, path_params, is_host=True - ) - if not remaining_params: - return URLPath(path=path, host=host) - - raise NoMatchFound(self.name, path_params) - - def path_for_without_name(self, name: str, path_params: dict) -> URLPath: - """ - Generate a URLPath for a route without a specific name and with path parameters. - - Args: - name (str): The name of the route. - path_params: Path parameters for route substitution. - - Returns: - URLPath: The generated URLPath. - - Raises: - NoMatchFound: If no matching route is found for the given name and parameters. - """ - if self.name is None: - remaining_name = name - else: - remaining_name = name[len(self.name) + 1 :] - - host, remaining_params = replace_params( - self.host_format, self.param_convertors, path_params, is_host=True - ) - - for route in self.routes or []: - try: - url = route.path_for(remaining_name, **remaining_params) - return URLPath(path=str(url), protocol=url.protocol, host=host) - except NoMatchFound: - pass - - raise NoMatchFound(name, path_params) - - def __repr__(self) -> str: - name = self.name or "" - return f"{self.__class__.__name__}(host={self.host!r}, name={name!r}, app={self.app!r})" - - -class BaseRouter: - """ - A Lilya base router object. - """ - - __slots__ = ( - "routes", - "redirect_slashes", - "default", - "on_startup", - "on_shutdown", - "middleware", - "permissions", - "include_in_schema", - "deprecated", - "lifespan_context", - "middleware_stack", - "permission_started", - "settings_module", - "is_sub_router", - ) - - def __init__( - self, - routes: Sequence[BasePath] | None = None, - redirect_slashes: bool = True, - default: ASGIApp | None = None, - on_startup: Sequence[Callable[[], Any]] | None = None, - on_shutdown: Sequence[Callable[[], Any]] | None = None, - lifespan: Lifespan[Any] | None = None, - *, - middleware: Sequence[DefineMiddleware] | None = None, - permissions: Sequence[DefinePermission] | None = None, - settings_module: Annotated[ - Settings | None, - Doc( - """ - Alternative settings parameter. This parameter is an alternative to - `LILYA_SETTINGS_MODULE` way of loading your settings into a Lilya application. - - When the `settings_module` is provided, it will make sure it takes priority over - any other settings provided for the instance. - """ - ), - ] = None, - include_in_schema: bool = True, - deprecated: bool = False, - is_sub_router: bool = False, - ) -> None: - assert lifespan is None or ( - on_startup is None and on_shutdown is None - ), "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both." - - if inspect.isasyncgenfunction(lifespan) or inspect.isgeneratorfunction(lifespan): - raise ImproperlyConfigured( - "async function generators are not allowed. " - "Use @contextlib.asynccontextmanager instead." - ) - - self.on_startup = [] if on_startup is None else list(on_startup) - self.on_shutdown = [] if on_shutdown is None else list(on_shutdown) - - self.lifespan_context = handle_lifespan_events( - on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan - ) - - if self.lifespan_context is None: - self.lifespan_context = AsyncLifespan(self) - - self.routes = [] if routes is None else list(routes) - self.redirect_slashes = redirect_slashes - self.default = self.handle_not_found if default is None else default - self.include_in_schema = include_in_schema - self.deprecated = deprecated - - self.middleware = middleware if middleware is not None else [] - self.permissions = permissions if permissions is not None else [] - self.settings_module = settings_module - self.middleware_stack = self.app - self.permission_started = False - self.is_sub_router = is_sub_router - - self._apply_middleware(self.middleware) - self._apply_permissions(self.permissions) - self._set_settings_app(self.settings_module, self) - - def _apply_middleware(self, middleware: Sequence[DefineMiddleware] | None) -> None: - """ - Apply middleware to the app. - - Args: - middleware (Union[Sequence[DefineMiddleware], None]): The middleware. - - Returns: - None - """ - if middleware is not None: - for cls, args, options in reversed(middleware): - self.middleware_stack = cls(app=self.middleware_stack, *args, **options) - - def _apply_permissions(self, permissions: Sequence[DefinePermission] | None) -> None: - """ - Apply permissions to the app. - - Args: - permissions (Union[Sequence[DefinePermission], None]): The permissions. - - Returns: - None - """ - if permissions is not None: - for cls, args, options in reversed(self.permissions): - self.middleware_stack = cls(app=self.middleware_stack, *args, **options) - - def _set_settings_app(self, settings_module: Settings, app: ASGIApp) -> None: + def _set_settings_app(self, settings_module: Settings, app: ASGIApp) -> None: """ Sets the main `app` of the settings module. This is particularly useful for reversing urls. @@ -1589,63 +1315,223 @@ def add_route( ) self.routes.append(route) - def add_websocket_route( + def add_websocket_route( + self, + path: str, + handler: Callable[[WebSocket], Awaitable[None]], + name: str | None = None, + middleware: Sequence[DefineMiddleware] | None = None, + permissions: Sequence[DefinePermission] | None = None, + exception_handlers: Mapping[Any, ExceptionHandler] | None = None, + ) -> None: + """ + Manually creates a `WebSocketPath` from a given handler. + """ + route = WebSocketPath( + path, + handler=handler, + middleware=middleware, + permissions=permissions, + name=name, + exception_handlers=exception_handlers, + ) + self.routes.append(route) + + def add_event_handler( + self, event_type: str, func: Callable[[], Any] + ) -> None: # pragma: no cover + assert event_type in ( + EventType.ON_STARTUP, + EventType.ON_SHUTDOWN, + EventType.STARTUP, + EventType.SHUTDOWN, + ) + + if event_type in (EventType.ON_STARTUP, EventType.STARTUP): + self.on_startup.append(func) + else: + self.on_shutdown.append(func) + + def on_event(self, event_type: str) -> Callable: + def wrapper(func: Callable) -> Callable: + self.add_event_handler(event_type, func) + return func + + return wrapper + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] == ScopeType.LIFESPAN: + await self.lifespan(scope, receive, send) + return + await self.middleware_stack(scope, receive, send) + + +class Router(BaseRouter): + """ + Contains all the needed base routing system and adds the ability to use decorators. + """ + + def get( + self, + path: str, + name: str | None = None, + middleware: Sequence[DefineMiddleware] | None = None, + permissions: Sequence[DefinePermission] | None = None, + exception_handlers: Mapping[Any, ExceptionHandler] | None = None, + include_in_schema: bool = True, + ) -> Callable[..., Any]: + """ + Decorator for defining a GET route. + + Args: + path (str): The URL path pattern for the route. + methods (list[str] | None, optional): The HTTP methods allowed for the route. Defaults to None. + name (str | None, optional): The name of the route. Defaults to None. + middleware (Sequence[DefineMiddleware] | None, optional): The middleware functions to apply to the route. Defaults to None. + permissions (Sequence[DefinePermission] | None, optional): The permissions required for the route. Defaults to None. + exception_handlers (Mapping[Any, ExceptionHandler] | None, optional): The exception handlers for the route. Defaults to None. + include_in_schema (bool, optional): Whether to include the route in the API schema. Defaults to True. + + Returns: + Callable[..., Any]: The decorated function. + """ + + def wrapper(func: Callable) -> Callable: + self.add_route( + path=path, + handler=func, + methods=["GET"], + name=name, + middleware=middleware, + permissions=permissions, + exception_handlers=exception_handlers, + include_in_schema=include_in_schema, + ) + return func + + return wrapper + + def head( + self, + path: str, + name: str | None = None, + middleware: Sequence[DefineMiddleware] | None = None, + permissions: Sequence[DefinePermission] | None = None, + exception_handlers: Mapping[Any, ExceptionHandler] | None = None, + include_in_schema: bool = True, + ) -> Callable[..., Any]: + """ + Decorator for defining a HEAD route. + + Args: + path (str): The URL path pattern for the route. + methods (list[str] | None, optional): The HTTP methods allowed for the route. Defaults to None. + name (str | None, optional): The name of the route. Defaults to None. + middleware (Sequence[DefineMiddleware] | None, optional): The middleware functions to apply to the route. Defaults to None. + permissions (Sequence[DefinePermission] | None, optional): The permissions required for the route. Defaults to None. + exception_handlers (Mapping[Any, ExceptionHandler] | None, optional): The exception handlers for the route. Defaults to None. + include_in_schema (bool, optional): Whether to include the route in the API schema. Defaults to True. + + Returns: + Callable[..., Any]: The decorated function. + """ + + def wrapper(func: Callable) -> Callable: + self.add_route( + path=path, + handler=func, + methods=["HEAD"], + name=name, + middleware=middleware, + permissions=permissions, + exception_handlers=exception_handlers, + include_in_schema=include_in_schema, + ) + return func + + return wrapper + + def post( + self, + path: str, + name: str | None = None, + middleware: Sequence[DefineMiddleware] | None = None, + permissions: Sequence[DefinePermission] | None = None, + exception_handlers: Mapping[Any, ExceptionHandler] | None = None, + include_in_schema: bool = True, + ) -> Callable[..., Any]: + """ + Decorator for defining a POST route. + + Args: + path (str): The URL path pattern for the route. + methods (list[str] | None, optional): The HTTP methods allowed for the route. Defaults to None. + name (str | None, optional): The name of the route. Defaults to None. + middleware (Sequence[DefineMiddleware] | None, optional): The middleware functions to apply to the route. Defaults to None. + permissions (Sequence[DefinePermission] | None, optional): The permissions required for the route. Defaults to None. + exception_handlers (Mapping[Any, ExceptionHandler] | None, optional): The exception handlers for the route. Defaults to None. + include_in_schema (bool, optional): Whether to include the route in the API schema. Defaults to True. + + Returns: + Callable[..., Any]: The decorated function. + """ + + def wrapper(func: Callable) -> Callable: + self.add_route( + path=path, + handler=func, + methods=["POST"], + name=name, + middleware=middleware, + permissions=permissions, + exception_handlers=exception_handlers, + include_in_schema=include_in_schema, + ) + return func + + return wrapper + + def put( self, path: str, - handler: Callable[[WebSocket], Awaitable[None]], name: str | None = None, middleware: Sequence[DefineMiddleware] | None = None, permissions: Sequence[DefinePermission] | None = None, exception_handlers: Mapping[Any, ExceptionHandler] | None = None, - ) -> None: - """ - Manually creates a `WebSocketPath` from a given handler. + include_in_schema: bool = True, + ) -> Callable[..., Any]: """ - route = WebSocketPath( - path, - handler=handler, - middleware=middleware, - permissions=permissions, - name=name, - exception_handlers=exception_handlers, - ) - self.routes.append(route) + Decorator for defining a PUT route. - def add_event_handler( - self, event_type: str, func: Callable[[], Any] - ) -> None: # pragma: no cover - assert event_type in ( - EventType.ON_STARTUP, - EventType.ON_SHUTDOWN, - EventType.STARTUP, - EventType.SHUTDOWN, - ) + Args: + path (str): The URL path pattern for the route. + methods (list[str] | None, optional): The HTTP methods allowed for the route. Defaults to None. + name (str | None, optional): The name of the route. Defaults to None. + middleware (Sequence[DefineMiddleware] | None, optional): The middleware functions to apply to the route. Defaults to None. + permissions (Sequence[DefinePermission] | None, optional): The permissions required for the route. Defaults to None. + exception_handlers (Mapping[Any, ExceptionHandler] | None, optional): The exception handlers for the route. Defaults to None. + include_in_schema (bool, optional): Whether to include the route in the API schema. Defaults to True. - if event_type in (EventType.ON_STARTUP, EventType.STARTUP): - self.on_startup.append(func) - else: - self.on_shutdown.append(func) + Returns: + Callable[..., Any]: The decorated function. + """ - def on_event(self, event_type: str) -> Callable: def wrapper(func: Callable) -> Callable: - self.add_event_handler(event_type, func) + self.add_route( + path=path, + handler=func, + methods=["PUT"], + name=name, + middleware=middleware, + permissions=permissions, + exception_handlers=exception_handlers, + include_in_schema=include_in_schema, + ) return func return wrapper - async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] == ScopeType.LIFESPAN: - await self.lifespan(scope, receive, send) - return - await self.middleware_stack(scope, receive, send) - - -class Router(BaseRouter): - """ - Contains all the needed base routing system and adds the ability to use decorators. - """ - - def get( + def patch( self, path: str, name: str | None = None, @@ -1655,7 +1541,7 @@ def get( include_in_schema: bool = True, ) -> Callable[..., Any]: """ - Decorator for defining a GET route. + Decorator for defining a PATCH route. Args: path (str): The URL path pattern for the route. @@ -1674,7 +1560,7 @@ def wrapper(func: Callable) -> Callable: self.add_route( path=path, handler=func, - methods=["GET"], + methods=["PATCH"], name=name, middleware=middleware, permissions=permissions, @@ -1685,7 +1571,7 @@ def wrapper(func: Callable) -> Callable: return wrapper - def head( + def delete( self, path: str, name: str | None = None, @@ -1695,7 +1581,7 @@ def head( include_in_schema: bool = True, ) -> Callable[..., Any]: """ - Decorator for defining a HEAD route. + Decorator for defining a DELETE route. Args: path (str): The URL path pattern for the route. @@ -1714,7 +1600,7 @@ def wrapper(func: Callable) -> Callable: self.add_route( path=path, handler=func, - methods=["HEAD"], + methods=["DELETE"], name=name, middleware=middleware, permissions=permissions, @@ -1725,7 +1611,7 @@ def wrapper(func: Callable) -> Callable: return wrapper - def post( + def trace( self, path: str, name: str | None = None, @@ -1735,7 +1621,7 @@ def post( include_in_schema: bool = True, ) -> Callable[..., Any]: """ - Decorator for defining a POST route. + Decorator for defining a TRACE route. Args: path (str): The URL path pattern for the route. @@ -1754,7 +1640,7 @@ def wrapper(func: Callable) -> Callable: self.add_route( path=path, handler=func, - methods=["POST"], + methods=["TRACE"], name=name, middleware=middleware, permissions=permissions, @@ -1765,7 +1651,7 @@ def wrapper(func: Callable) -> Callable: return wrapper - def put( + def options( self, path: str, name: str | None = None, @@ -1775,7 +1661,7 @@ def put( include_in_schema: bool = True, ) -> Callable[..., Any]: """ - Decorator for defining a PUT route. + Decorator for defining a OPTIONS route. Args: path (str): The URL path pattern for the route. @@ -1794,7 +1680,7 @@ def wrapper(func: Callable) -> Callable: self.add_route( path=path, handler=func, - methods=["PUT"], + methods=["OPTIONS"], name=name, middleware=middleware, permissions=permissions, @@ -1805,9 +1691,10 @@ def wrapper(func: Callable) -> Callable: return wrapper - def patch( + def route( self, path: str, + methods: list[str], name: str | None = None, middleware: Sequence[DefineMiddleware] | None = None, permissions: Sequence[DefinePermission] | None = None, @@ -1815,7 +1702,7 @@ def patch( include_in_schema: bool = True, ) -> Callable[..., Any]: """ - Decorator for defining a PATCH route. + Decorator for defining a generic route. Args: path (str): The URL path pattern for the route. @@ -1834,7 +1721,7 @@ def wrapper(func: Callable) -> Callable: self.add_route( path=path, handler=func, - methods=["PATCH"], + methods=methods, name=name, middleware=middleware, permissions=permissions, @@ -1845,203 +1732,318 @@ def wrapper(func: Callable) -> Callable: return wrapper - def delete( + def websocket( self, path: str, name: str | None = None, middleware: Sequence[DefineMiddleware] | None = None, permissions: Sequence[DefinePermission] | None = None, exception_handlers: Mapping[Any, ExceptionHandler] | None = None, - include_in_schema: bool = True, ) -> Callable[..., Any]: """ - Decorator for defining a DELETE route. + Decorator for defining a WebSocket route. + + Args: + path (str): The URL path for the WebSocket route. + name (str, optional): The name of the route. Defaults to None. + middleware (Sequence[DefineMiddleware], optional): The middleware to apply to the route. Defaults to None. + permissions (Sequence[DefinePermission], optional): The permissions required for the route. Defaults to None. + exception_handlers (Mapping[Any, ExceptionHandler], optional): The exception handlers for the route. Defaults to None. + + Returns: + Callable[..., Any]: The decorated function. + + """ + + def wrapper(func: Callable) -> Callable: + self.add_websocket_route( + path=path, + handler=func, + name=name, + middleware=middleware, + permissions=permissions, + exception_handlers=exception_handlers, + ) + return func + + return wrapper + + +# Declare an alias for the Path +RoutePath = Path + + +class Include(BasePath): + router_class: ClassVar[type[BaseRouter]] = Router + + __slots__ = ( + "path", + "app", + "namespace", + "pattern", + "name", + "exception_handlers", + "middleware", + "permissions", + "exception_handlers", + "deprecated", + ) + + def __init__( + self, + path: str, + app: ASGIApp | str | None = None, + routes: Sequence[BasePath] | None = None, + namespace: str | None = None, + pattern: str | None = None, + name: str | None = None, + *, + middleware: Sequence[DefineMiddleware] | None = None, + permissions: Sequence[DefinePermission] | None = None, + exception_handlers: Mapping[Any, ExceptionHandler] | None = None, + include_in_schema: bool = True, + deprecated: bool = False, + redirect_slashes: bool = True, + ) -> None: + """ + Initialize the router with specified parameters. + + Args: + path (str): The path associated with the router. + app (Union[ASGIApp, str, None]): The ASGI app. + routes (Union[Sequence[BasePath], None]): The routes. + namespace (Union[str, None]): The namespace. + pattern (Union[str, None]): The pattern. + name (Union[str, None]): The name. + middleware (Union[Sequence[DefineMiddleware], None]): The middleware. + permissions (Union[Sequence[DefinePermission], None]): The permissions. + include_in_schema (bool): Flag to include in the schema. + redirect_slashes (bool): (Only namespace or routes) Redirect slashes on mismatch. + + Returns: + None + """ + assert path == "" or path.startswith("/"), "Routed paths must start with '/'" + assert ( + app is not None or routes is not None or namespace is not None + ), "Either 'app=...', or 'routes=...', or 'namespace=...' must be specified" + self.path = clean_path(path) + + assert ( + namespace is None or routes is None + ), "Either 'namespace=...' or 'routes=', not both." + + if namespace and not isinstance(namespace, str): + raise ImproperlyConfigured("Namespace must be a string. Example: 'myapp.routes'.") + + if pattern and not isinstance(pattern, str): + raise ImproperlyConfigured("Pattern must be a string. Example: 'route_patterns'.") + + if pattern and routes: + raise ImproperlyConfigured("Pattern must be used only with namespace.") + + if namespace is not None: + routes = include(namespace, pattern) + + self.__base_app__: ASGIApp | Router + if isinstance(app, str): + self.__base_app__ = import_string(app) + self.handle_not_found = self.handle_not_found_fallthrough # type: ignore + elif app is not None: + self.handle_not_found = self.handle_not_found_fallthrough # type: ignore + self.__base_app__ = app + else: + self.__base_app__ = self.router_class( + routes=routes, + default=self.handle_not_found_fallthrough, + is_sub_router=True, + redirect_slashes=redirect_slashes, + ) + + self.app = self.__base_app__ + + self.middleware = middleware if middleware is not None else [] + self.permissions = permissions if permissions is not None else [] + self.exception_handlers = {} if exception_handlers is None else dict(exception_handlers) + + self._apply_middleware(middleware) + self._apply_permissions(permissions) + + self.name = name + self.include_in_schema = include_in_schema + self.deprecated = deprecated + + self.path_regex, self.path_format, self.param_convertors, self.path_start = compile_path( + clean_path(self.path + "/{path:path}") + ) + + def _apply_middleware(self, middleware: Sequence[DefineMiddleware] | None) -> None: + """ + Apply middleware to the app. + + Args: + middleware (Union[Sequence[DefineMiddleware], None]): The middleware. + + Returns: + None + """ + if middleware is not None: + for cls, args, options in reversed(middleware): + self.app = cls(app=self.app, *args, **options) + + def _apply_permissions(self, permissions: Sequence[DefinePermission] | None) -> None: + """ + Apply permissions to the app. Args: - path (str): The URL path pattern for the route. - methods (list[str] | None, optional): The HTTP methods allowed for the route. Defaults to None. - name (str | None, optional): The name of the route. Defaults to None. - middleware (Sequence[DefineMiddleware] | None, optional): The middleware functions to apply to the route. Defaults to None. - permissions (Sequence[DefinePermission] | None, optional): The permissions required for the route. Defaults to None. - exception_handlers (Mapping[Any, ExceptionHandler] | None, optional): The exception handlers for the route. Defaults to None. - include_in_schema (bool, optional): Whether to include the route in the API schema. Defaults to True. + permissions (Union[Sequence[DefinePermission], None]): The permissions. Returns: - Callable[..., Any]: The decorated function. + None """ + if permissions is not None: + for cls, args, options in reversed(permissions): + self.app = cls(app=self.app, *args, **options) - def wrapper(func: Callable) -> Callable: - self.add_route( - path=path, - handler=func, - methods=["DELETE"], - name=name, - middleware=middleware, - permissions=permissions, - exception_handlers=exception_handlers, - include_in_schema=include_in_schema, - ) - return func - - return wrapper + @property + def routes(self) -> list[BasePath]: + """ + Returns a list of declared path objects. + """ + return getattr(self.__base_app__, "routes", []) - def trace( - self, - path: str, - name: str | None = None, - middleware: Sequence[DefineMiddleware] | None = None, - permissions: Sequence[DefinePermission] | None = None, - exception_handlers: Mapping[Any, ExceptionHandler] | None = None, - include_in_schema: bool = True, - ) -> Callable[..., Any]: + def search(self, scope: Scope) -> tuple[Match, Scope]: """ - Decorator for defining a TRACE route. + Searches within the route patterns and matches against the regex. + + If found, then dispatches the request to the handler of the object. Args: - path (str): The URL path pattern for the route. - methods (list[str] | None, optional): The HTTP methods allowed for the route. Defaults to None. - name (str | None, optional): The name of the route. Defaults to None. - middleware (Sequence[DefineMiddleware] | None, optional): The middleware functions to apply to the route. Defaults to None. - permissions (Sequence[DefinePermission] | None, optional): The permissions required for the route. Defaults to None. - exception_handlers (Mapping[Any, ExceptionHandler] | None, optional): The exception handlers for the route. Defaults to None. - include_in_schema (bool, optional): Whether to include the route in the API schema. Defaults to True. + scope (Scope): The request scope. Returns: - Callable[..., Any]: The decorated function. + Tuple[Match, Scope]: The match result and child scope. """ + if scope["type"] in {ScopeType.HTTP, ScopeType.WEBSOCKET}: + root_path = scope.get("root_path", "") + route_path = get_route_path(scope) + match = self.path_regex.match(route_path) - def wrapper(func: Callable) -> Callable: - self.add_route( - path=path, - handler=func, - methods=["TRACE"], - name=name, - middleware=middleware, - permissions=permissions, - exception_handlers=exception_handlers, - include_in_schema=include_in_schema, - ) - return func + if match: + return self.handle_match(scope, match, route_path, root_path) - return wrapper + return Match.NONE, {} - def options( - self, - path: str, - name: str | None = None, - middleware: Sequence[DefineMiddleware] | None = None, - permissions: Sequence[DefinePermission] | None = None, - exception_handlers: Mapping[Any, ExceptionHandler] | None = None, - include_in_schema: bool = True, - ) -> Callable[..., Any]: + def handle_match( + self, scope: Scope, match: re.Match, route_path: str, root_path: str + ) -> tuple[Match, Scope]: """ - Decorator for defining a OPTIONS route. + Handles the case when a match is found in the route patterns. Args: - path (str): The URL path pattern for the route. - methods (list[str] | None, optional): The HTTP methods allowed for the route. Defaults to None. - name (str | None, optional): The name of the route. Defaults to None. - middleware (Sequence[DefineMiddleware] | None, optional): The middleware functions to apply to the route. Defaults to None. - permissions (Sequence[DefinePermission] | None, optional): The permissions required for the route. Defaults to None. - exception_handlers (Mapping[Any, ExceptionHandler] | None, optional): The exception handlers for the route. Defaults to None. - include_in_schema (bool, optional): Whether to include the route in the API schema. Defaults to True. + scope (Scope): The request scope. + match: The match object from the regex. Returns: - Callable[..., Any]: The decorated function. + Tuple[Match, Scope]: The match result and child scope. """ + matched_params = { + key: self.param_convertors[key].transform(value) + for key, value in match.groupdict().items() + } + remaining_path = f'/{matched_params.pop("path", "")}' + matched_path = route_path[: -len(remaining_path)] - def wrapper(func: Callable) -> Callable: - self.add_route( - path=path, - handler=func, - methods=["OPTIONS"], - name=name, - middleware=middleware, - permissions=permissions, - exception_handlers=exception_handlers, - include_in_schema=include_in_schema, - ) - return func + path_params = {**scope.get("path_params", {}), **matched_params} + child_scope = { + "path_params": path_params, + "app_root_path": scope.get("app_root_path", root_path), + "root_path": root_path + matched_path, + "handler": self.app, + } + return Match.FULL, child_scope - return wrapper + async def handle_dispatch(self, scope: Scope, receive: Receive, send: Send) -> None: + """ + Handles the dispatch of the request to the appropriate handler. - def route( - self, - path: str, - methods: list[str], - name: str | None = None, - middleware: Sequence[DefineMiddleware] | None = None, - permissions: Sequence[DefinePermission] | None = None, - exception_handlers: Mapping[Any, ExceptionHandler] | None = None, - include_in_schema: bool = True, - ) -> Callable[..., Any]: + Args: + scope (Scope): The request scope. + receive (Receive): The receive channel. + send (Send): The send channel. + + Returns: + None """ - Decorator for defining a generic route. + try: + await self.app(scope, receive, send) + except Exception as ex: + await self.handle_exception_handlers(scope, receive, send, ex) + + def path_for(self, name: str, /, **path_params: Any) -> URLPath: + """ + Generate a URLPath for a given route name and path parameters. Args: - path (str): The URL path pattern for the route. - methods (list[str] | None, optional): The HTTP methods allowed for the route. Defaults to None. - name (str | None, optional): The name of the route. Defaults to None. - middleware (Sequence[DefineMiddleware] | None, optional): The middleware functions to apply to the route. Defaults to None. - permissions (Sequence[DefinePermission] | None, optional): The permissions required for the route. Defaults to None. - exception_handlers (Mapping[Any, ExceptionHandler] | None, optional): The exception handlers for the route. Defaults to None. - include_in_schema (bool, optional): Whether to include the route in the API schema. Defaults to True. + name (str): The name of the route. + path_params: Path parameters for route substitution. Returns: - Callable[..., Any]: The decorated function. + URLPath: The generated URLPath. + + Raises: + NoMatchFound: If no matching route is found for the given name and parameters. """ - def wrapper(func: Callable) -> Callable: - self.add_route( - path=path, - handler=func, - methods=methods, - name=name, - middleware=middleware, - permissions=permissions, - exception_handlers=exception_handlers, - include_in_schema=include_in_schema, + if self.name is not None and name == self.name and "path" in path_params: + path_params["path"] = path_params["path"].lstrip("/") + path, remaining_params = replace_params( + self.path_format, self.param_convertors, path_params ) - return func + if not remaining_params: + return URLPath(path=path) + elif self.name is None or name.startswith(self.name + ":"): + return self._path_for_without_name(name, path_params) - return wrapper + raise NoMatchFound(name, path_params) - def websocket( - self, - path: str, - name: str | None = None, - middleware: Sequence[DefineMiddleware] | None = None, - permissions: Sequence[DefinePermission] | None = None, - exception_handlers: Mapping[Any, ExceptionHandler] | None = None, - ) -> Callable[..., Any]: + def _path_for_without_name(self, name: str, path_params: dict) -> URLPath: """ - Decorator for defining a WebSocket route. + Generate a URLPath for a route without a specific name and with path parameters. Args: - path (str): The URL path for the WebSocket route. - name (str, optional): The name of the route. Defaults to None. - middleware (Sequence[DefineMiddleware], optional): The middleware to apply to the route. Defaults to None. - permissions (Sequence[DefinePermission], optional): The permissions required for the route. Defaults to None. - exception_handlers (Mapping[Any, ExceptionHandler], optional): The exception handlers for the route. Defaults to None. + name (str): The name of the route. + path_params: Path parameters for route substitution. Returns: - Callable[..., Any]: The decorated function. + URLPath: The generated URLPath. + Raises: + NoMatchFound: If no matching route is found for the given name and parameters. """ + if self.name is None: + remaining_name = name + else: + remaining_name = name[len(self.name) + 1 :] - def wrapper(func: Callable) -> Callable: - self.add_websocket_route( - path=path, - handler=func, - name=name, - middleware=middleware, - permissions=permissions, - exception_handlers=exception_handlers, - ) - return func + path_kwarg = path_params.get("path") + path_params["path"] = "" + path_prefix, remaining_params = replace_params( + self.path_format, self.param_convertors, path_params + ) - return wrapper + if path_kwarg is not None: + remaining_params["path"] = path_kwarg + + for route in self.routes or []: + try: + url = route.path_for(remaining_name, **remaining_params) + return URLPath(path=path_prefix.rstrip("/") + str(url), protocol=url.protocol) + except NoMatchFound: + pass + raise NoMatchFound(name, path_params) -# Declare an alias for the Path -RoutePath = Path + def __repr__(self) -> str: + name = self.name or "" + return f"{self.__class__.__name__}(path={self.path!r}, name={name!r}, app={self.app!r})"