Allow user defined routes in ReactPy()

pyproject.toml

@@ -256,7 +256,6 @@ exclude_also = [
 ]
 
 [tool.ruff]
-target-version = "py39"
 line-length = 88
 lint.select = [
   "A",
@@ -328,13 +327,6 @@ lint.unfixable = [
 [tool.ruff.lint.isort]
 known-first-party = ["reactpy"]
 
-[tool.ruff.lint.flake8-tidy-imports]
-ban-relative-imports = "all"
-
-[tool.flake8]
-select = ["RPY"]                                                  # only need to check with reactpy-flake8
-exclude = ["**/node_modules/*", ".eggs/*", ".tox/*", "**/venv/*"]
-
 [tool.ruff.lint.per-file-ignores]
 # Tests can use magic values, assertions, and relative imports
 "**/tests/**/*" = ["PLR2004", "S101", "TID252"]
@@ -350,7 +342,3 @@ exclude = ["**/node_modules/*", ".eggs/*", ".tox/*", "**/venv/*"]
   # Allow print
   "T201",
 ]
-
-[tool.black]
-target-version = ["py39"]
-line-length = 88

src/reactpy/asgi/middleware.py

@@ -21,13 +21,21 @@
 from reactpy.core.hooks import ConnectionContext
 from reactpy.core.layout import Layout
 from reactpy.core.serve import serve_layout
-from reactpy.types import Connection, Location, ReactPyConfig, RootComponentConstructor
+from reactpy.types import (
+    AsgiApp,
+    AsgiHttpApp,
+    AsgiLifespanApp,
+    AsgiWebsocketApp,
+    Connection,
+    Location,
+    ReactPyConfig,
+    RootComponentConstructor,
+)
 
 _logger = logging.getLogger(__name__)
 
 
 class ReactPyMiddleware:
-    _asgi_single_callable: bool = True
     root_component: RootComponentConstructor | None = None
     root_components: dict[str, RootComponentConstructor]
     multiple_root_components: bool = True
@@ -73,8 +81,13 @@ def __init__(
         self.js_modules_pattern = re.compile(f"^{self.web_modules_path}.*")
         self.static_pattern = re.compile(f"^{self.static_path}.*")
 
+        # User defined ASGI apps
+        self.extra_http_routes: dict[str, AsgiHttpApp] = {}
+        self.extra_ws_routes: dict[str, AsgiWebsocketApp] = {}
+        self.extra_lifespan_app: AsgiLifespanApp | None = None
+
         # Component attributes
-        self.user_app: asgi_types.ASGI3Application = guarantee_single_callable(app)  # type: ignore
+        self.asgi_app: asgi_types.ASGI3Application = guarantee_single_callable(app)  # type: ignore
         self.root_components = import_components(root_components)
 
         # Directory attributes
@@ -106,8 +119,13 @@ async def __call__(
         if scope["type"] == "http" and self.match_web_modules_path(scope):
             return await self.web_modules_app(scope, receive, send)
 
+        # URL routing for user-defined routes
+        matched_app = self.match_extra_paths(scope)
+        if matched_app:
+            return await matched_app(scope, receive, send)  # type: ignore
+
         # Serve the user's application
-        await self.user_app(scope, receive, send)
+        await self.asgi_app(scope, receive, send)
 
     def match_dispatch_path(self, scope: asgi_types.WebSocketScope) -> bool:
         return bool(re.match(self.dispatcher_pattern, scope["path"]))
@@ -118,6 +136,11 @@ def match_static_path(self, scope: asgi_types.HTTPScope) -> bool:
     def match_web_modules_path(self, scope: asgi_types.HTTPScope) -> bool:
         return bool(re.match(self.js_modules_pattern, scope["path"]))
 
+    def match_extra_paths(self, scope: asgi_types.Scope) -> AsgiApp | None:
+        # Custom defined routes are unused within middleware to encourage users to handle
+        # routing within their root ASGI application.
+        return None
+
 
 @dataclass
 class ComponentDispatchApp:
@@ -223,7 +246,7 @@ async def __call__(
         """ASGI app for ReactPy static files."""
         if not self._static_file_server:
             self._static_file_server = ServeStaticASGI(
-                self.parent.user_app,
+                self.parent.asgi_app,
                 root=self.parent.static_dir,
                 prefix=self.parent.static_path,
             )
@@ -245,7 +268,7 @@ async def __call__(
         """ASGI app for ReactPy web modules."""
         if not self._static_file_server:
             self._static_file_server = ServeStaticASGI(
-                self.parent.user_app,
+                self.parent.asgi_app,
                 root=self.parent.web_modules_dir,
                 prefix=self.parent.web_modules_path,
                 autorefresh=True,

src/reactpy/asgi/standalone.py

@@ -6,14 +6,28 @@
 from datetime import datetime, timezone
 from email.utils import formatdate
 from logging import getLogger
+from typing import Callable, Literal, cast, overload
 
 from asgiref import typing as asgi_types
 from typing_extensions import Unpack
 
 from reactpy import html
 from reactpy.asgi.middleware import ReactPyMiddleware
-from reactpy.asgi.utils import dict_to_byte_list, http_response, vdom_head_to_html
-from reactpy.types import ReactPyConfig, RootComponentConstructor, VdomDict
+from reactpy.asgi.utils import (
+    dict_to_byte_list,
+    http_response,
+    import_dotted_path,
+    vdom_head_to_html,
+)
+from reactpy.types import (
+    AsgiApp,
+    AsgiHttpApp,
+    AsgiLifespanApp,
+    AsgiWebsocketApp,
+    ReactPyConfig,
+    RootComponentConstructor,
+    VdomDict,
+)
 from reactpy.utils import render_mount_template
 
 _logger = getLogger(__name__)
@@ -34,7 +48,7 @@ def __init__(
         """ReactPy's standalone ASGI application.
 
         Parameters:
-            root_component: The root component to render. This component is assumed to be a single page application.
+            root_component: The root component to render. This app is typically a single page application.
             http_headers: Additional headers to include in the HTTP response for the base HTML document.
             html_head: Additional head elements to include in the HTML response.
             html_lang: The language of the HTML document.
@@ -51,6 +65,89 @@ def match_dispatch_path(self, scope: asgi_types.WebSocketScope) -> bool:
         """Method override to remove `dotted_path` from the dispatcher URL."""
         return str(scope["path"]) == self.dispatcher_path
 
+    def match_extra_paths(self, scope: asgi_types.Scope) -> AsgiApp | None:
+        """Method override to match user-provided HTTP/Websocket routes."""
+        if scope["type"] == "lifespan":
+            return self.extra_lifespan_app
+
+        if scope["type"] == "http":
+            routing_dictionary = self.extra_http_routes.items()
+
+        if scope["type"] == "websocket":
+            routing_dictionary = self.extra_ws_routes.items()  # type: ignore
+
+        return next(
+            (
+                app
+                for route, app in routing_dictionary
+                if re.match(route, scope["path"])
+            ),
+            None,
+        )
+
+    @overload
+    def route(
+        self,
+        path: str,
+        type: Literal["http"] = "http",
+    ) -> Callable[[AsgiHttpApp | str], AsgiApp]: ...
+
+    @overload
+    def route(
+        self,
+        path: str,
+        type: Literal["websocket"],
+    ) -> Callable[[AsgiWebsocketApp | str], AsgiApp]: ...
+
+    def route(
+        self,
+        path: str,
+        type: Literal["http", "websocket"] = "http",
+    ) -> (
+        Callable[[AsgiHttpApp | str], AsgiApp]
+        | Callable[[AsgiWebsocketApp | str], AsgiApp]
+    ):
+        """Interface that allows user to define their own HTTP/Websocket routes
+        within the current ReactPy application.
+
+        Parameters:
+            path: The URL route to match, using regex format.
+            type: The protocol to route for. Can be 'http' or 'websocket'.
+        """
+
+        def decorator(
+            app: AsgiApp | str,
+        ) -> AsgiApp:
+            re_path = path
+            if not re_path.startswith("^"):
+                re_path = f"^{re_path}"
+            if not re_path.endswith("$"):
+                re_path = f"{re_path}$"
+
+            asgi_app: AsgiApp = import_dotted_path(app) if isinstance(app, str) else app
+            if type == "http":
+                self.extra_http_routes[re_path] = cast(AsgiHttpApp, asgi_app)
+            elif type == "websocket":
+                self.extra_ws_routes[re_path] = cast(AsgiWebsocketApp, asgi_app)
+
+            return asgi_app
+
+        return decorator
+
+    def lifespan(self, app: AsgiLifespanApp | str) -> None:
+        """Interface that allows user to define their own lifespan app
+        within the current ReactPy application.
+
+        Parameters:
+            app: The ASGI application to route to.
+        """
+        if self.extra_lifespan_app:
+            raise ValueError("Only one lifespan app can be defined.")
+
+        self.extra_lifespan_app = (
+            import_dotted_path(app) if isinstance(app, str) else app
+        )
+
 
 @dataclass
 class ReactPyApp:

src/reactpy/testing/backend.py

@@ -140,7 +140,7 @@ async def __aexit__(
             raise LogAssertionError(msg) from logged_errors[0]
 
         await asyncio.wait_for(
-            self.webserver.shutdown(), timeout=60 if GITHUB_ACTIONS else 5
+            self.webserver.shutdown(), timeout=90 if GITHUB_ACTIONS else 5
         )
 
     async def restart(self) -> None:

src/reactpy/types.py

@@ -2,7 +2,7 @@
 
 import sys
 from collections import namedtuple
-from collections.abc import Mapping, Sequence
+from collections.abc import Awaitable, Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from types import TracebackType
@@ -15,6 +15,7 @@
     NamedTuple,
     Protocol,
     TypeVar,
+    Union,
     overload,
     runtime_checkable,
 )
@@ -296,3 +297,73 @@ class ReactPyConfig(TypedDict, total=False):
     async_rendering: bool
     debug: bool
     tests_default_timeout: int
+
+
+AsgiHttpReceive = Callable[
+    [],
+    Awaitable[asgi_types.HTTPRequestEvent | asgi_types.HTTPDisconnectEvent],
+]
+
+AsgiHttpSend = Callable[
+    [
+        asgi_types.HTTPResponseStartEvent
+        | asgi_types.HTTPResponseBodyEvent
+        | asgi_types.HTTPResponseTrailersEvent
+        | asgi_types.HTTPServerPushEvent
+        | asgi_types.HTTPDisconnectEvent
+    ],
+    Awaitable[None],
+]
+
+AsgiWebsocketReceive = Callable[
+    [],
+    Awaitable[
+        asgi_types.WebSocketConnectEvent
+        | asgi_types.WebSocketDisconnectEvent
+        | asgi_types.WebSocketReceiveEvent
+    ],
+]
+
+AsgiWebsocketSend = Callable[
+    [
+        asgi_types.WebSocketAcceptEvent
+        | asgi_types.WebSocketSendEvent
+        | asgi_types.WebSocketResponseStartEvent
+        | asgi_types.WebSocketResponseBodyEvent
+        | asgi_types.WebSocketCloseEvent
+    ],
+    Awaitable[None],
+]
+
+AsgiLifespanReceive = Callable[
+    [],
+    Awaitable[asgi_types.LifespanStartupEvent | asgi_types.LifespanShutdownEvent],
+]
+
+AsgiLifespanSend = Callable[
+    [
+        asgi_types.LifespanStartupCompleteEvent
+        | asgi_types.LifespanStartupFailedEvent
+        | asgi_types.LifespanShutdownCompleteEvent
+        | asgi_types.LifespanShutdownFailedEvent
+    ],
+    Awaitable[None],
+]
+
+AsgiHttpApp = Callable[
+    [asgi_types.HTTPScope, AsgiHttpReceive, AsgiHttpSend],
+    Awaitable[None],
+]
+
+AsgiWebsocketApp = Callable[
+    [asgi_types.WebSocketScope, AsgiWebsocketReceive, AsgiWebsocketSend],
+    Awaitable[None],
+]
+
+AsgiLifespanApp = Callable[
+    [asgi_types.LifespanScope, AsgiLifespanReceive, AsgiLifespanSend],
+    Awaitable[None],
+]
+
+
+AsgiApp = Union[AsgiHttpApp, AsgiWebsocketApp, AsgiLifespanApp]