rework as location object

Author: rmorshea

src/idom/server/default.py

@@ -5,7 +5,7 @@
 
 from idom.types import RootComponentConstructor
 
-from .types import ServerImplementation
+from .types import Location, ServerImplementation
 from .utils import all_implementations
 
 
@@ -36,9 +36,15 @@ async def serve_development_app(
 
 
 def use_scope() -> Any:
+    """Return the current ASGI/WSGI scope"""
     return _default_implementation().use_scope()
 
 
+def use_location() -> Location:
+    """Return the current route as a string"""
+    return _default_implementation().use_location()
+
+
 _DEFAULT_IMPLEMENTATION: ServerImplementation[Any] | None = None
 
 

src/idom/server/fastapi.py

@@ -8,6 +8,10 @@
 serve_development_app = starlette.serve_development_app
 """Alias for :func:`idom.server.starlette.serve_development_app`"""
 
+# see: https://github.com/idom-team/flake8-idom-hooks/issues/12
+use_location = starlette.use_location  # noqa: ROH101
+"""Alias for :func:`idom.server.starlette.use_location`"""
+
 # see: https://github.com/idom-team/flake8-idom-hooks/issues/12
 use_scope = starlette.use_scope  # noqa: ROH101
 """Alias for :func:`idom.server.starlette.use_scope`"""

src/idom/server/flask.py

@@ -29,6 +29,7 @@
 from idom.core.layout import LayoutEvent, LayoutUpdate
 from idom.core.serve import serve_json_patch
 from idom.core.types import ComponentType, RootComponentConstructor
+from idom.server.types import Location
 from idom.utils import Ref
 
 from .utils import safe_client_build_dir_path, safe_web_modules_dir_path
@@ -110,6 +111,23 @@ def run_server() -> None:
             raise RuntimeError("Failed to shutdown server.")
 
 
+def use_location() -> Location:
+    """Get the current route as a string"""
+    conn = use_connection()
+    search = conn.request.query_string.decode()
+    return Location(pathname="/" + conn.path, search="?" + search if search else "")
+
+
+def use_scope() -> dict[str, Any]:
+    """Get the current WSGI environment"""
+    return use_request().environ
+
+
+def use_request() -> Request:
+    """Get the current ``Request``"""
+    return use_connection().request
+
+
 def use_connection() -> Connection:
     """Get the current :class:`Connection`"""
     connection = use_context(ConnectionContext)
@@ -120,14 +138,18 @@ def use_connection() -> Connection:
     return connection
 
 
-def use_request() -> Request:
-    """Get the current ``Request``"""
-    return use_connection().request
+@dataclass
+class Connection:
+    """A simple wrapper for holding connection information"""
 
+    request: Request
+    """The current request object"""
 
-def use_scope() -> dict[str, Any]:
-    """Get the current WSGI environment"""
-    return use_request().environ
+    websocket: WebSocket
+    """A handle to the current websocket"""
+
+    path: str
+    """The current path being served"""
 
 
 @dataclass
@@ -181,13 +203,13 @@ def send(value: Any) -> None:
         def recv() -> LayoutEvent:
             return LayoutEvent(**json.loads(ws.receive()))
 
-        dispatch_in_thread(ws, path, constructor(), send, recv)
+        _dispatch_in_thread(ws, path, constructor(), send, recv)
 
     sock.route("/_api/stream", endpoint="without_path")(model_stream)
     sock.route("/<path:path>/_api/stream", endpoint="with_path")(model_stream)
 
 
-def dispatch_in_thread(
+def _dispatch_in_thread(
     websocket: WebSocket,
     path: str,
     component: ComponentType,
@@ -260,20 +282,6 @@ def run_send() -> None:
         )
 
 
-@dataclass
-class Connection:
-    """A simple wrapper for holding connection information"""
-
-    request: Request
-    """The current request object"""
-
-    websocket: WebSocket
-    """A handle to the current websocket"""
-
-    path: str
-    """The current path being served"""
-
-
 class _DispatcherThreadInfo(NamedTuple):
     dispatch_loop: asyncio.AbstractEventLoop
     dispatch_future: "asyncio.Future[Any]"

src/idom/server/sanic.py

@@ -24,6 +24,7 @@
     serve_json_patch,
 )
 from idom.core.types import RootComponentConstructor
+from idom.server.types import Location
 
 from ._asgi import serve_development_asgi
 from .utils import safe_client_build_dir_path, safe_web_modules_dir_path
@@ -66,6 +67,28 @@ async def serve_development_app(
     await serve_development_asgi(app, host, port, started)
 
 
+def use_location() -> Location:
+    """Get the current route as a string"""
+    conn = use_connection()
+    search = conn.request.query_string
+    return Location(pathname="/" + conn.path, search="?" + search if search else "")
+
+
+def use_scope() -> ASGIScope:
+    """Get the current ASGI scope"""
+    app = use_request().app
+    try:
+        asgi_app = app._asgi_app
+    except AttributeError:  # pragma: no cover
+        raise RuntimeError("No scope. Sanic may not be running with an ASGI server")
+    return asgi_app.transport.scope
+
+
+def use_request() -> request.Request:
+    """Get the current ``Request``"""
+    return use_connection().request
+
+
 def use_connection() -> Connection:
     """Get the current :class:`Connection`"""
     connection = use_context(ConnectionContext)
@@ -76,19 +99,18 @@ def use_connection() -> Connection:
     return connection
 
 
-def use_request() -> request.Request:
-    """Get the current ``Request``"""
-    return use_connection().request
+@dataclass
+class Connection:
+    """A simple wrapper for holding connection information"""
 
+    request: request.Request
+    """The current request object"""
 
-def use_scope() -> ASGIScope:
-    """Get the current ASGI scope"""
-    app = use_request().app
-    try:
-        asgi_app = app._asgi_app
-    except AttributeError:  # pragma: no cover
-        raise RuntimeError("No scope. Sanic may not be running with an ASGI server")
-    return asgi_app.transport.scope
+    websocket: WebSocketCommonProtocol
+    """A handle to the current websocket"""
+
+    path: str
+    """The current path being served"""
 
 
 @dataclass
@@ -156,20 +178,6 @@ async def model_stream(
     blueprint.add_websocket_route(model_stream, "/<path:path>/_api/stream")
 
 
-@dataclass
-class Connection:
-    """A simple wrapper for holding connection information"""
-
-    request: request.Request
-    """The current request object"""
-
-    websocket: WebSocketCommonProtocol
-    """A handle to the current websocket"""
-
-    path: str
-    """The current path being served"""
-
-
 def _make_send_recv_callbacks(
     socket: WebSocketCommonProtocol,
 ) -> Tuple[SendCoroutine, RecvCoroutine]:

src/idom/server/starlette.py

@@ -22,6 +22,7 @@
     serve_json_patch,
 )
 from idom.core.types import RootComponentConstructor
+from idom.server.types import Location
 
 from ._asgi import serve_development_asgi
 from .utils import CLIENT_BUILD_DIR, safe_client_build_dir_path
@@ -71,6 +72,19 @@ async def serve_development_app(
     await serve_development_asgi(app, host, port, started)
 
 
+def use_location() -> Location:
+    """Get the current route as a string"""
+    scope = use_scope()
+    pathname = "/" + scope["path_params"].get("path", "")
+    search = scope["query_string"].decode()
+    return Location(pathname, "?" + search if search else "")
+
+
+def use_scope() -> Scope:
+    """Get the current ASGI scope dictionary"""
+    return use_websocket().scope
+
+
 def use_websocket() -> WebSocket:
     """Get the current WebSocket object"""
     websocket = use_context(WebSocketContext)
@@ -81,11 +95,6 @@ def use_websocket() -> WebSocket:
     return websocket
 
 
-def use_scope() -> Scope:
-    """Get the current ASGI scope dictionary"""
-    return use_websocket().scope
-
-
 @dataclass
 class Options:
     """Optionsuration options for :class:`StarletteRenderServer`"""
@@ -147,9 +156,6 @@ def _setup_single_view_dispatcher_route(
     @app.websocket_route(options.url_prefix + "/_api/stream")
     @app.websocket_route(options.url_prefix + "/{path:path}/_api/stream")
     async def model_stream(socket: WebSocket) -> None:
-        path_params: dict[str, str] = socket.scope["path_params"]
-        path_params["path"] = "/" + path_params.get("path", "")
-
         await socket.accept()
         send, recv = _make_send_recv_callbacks(socket)
         try:

src/idom/server/tornado.py

@@ -21,12 +21,13 @@
 from idom.core.layout import Layout, LayoutEvent
 from idom.core.serve import VdomJsonPatch, serve_json_patch
 from idom.core.types import ComponentConstructor
+from idom.server.types import Location
 
 from .utils import CLIENT_BUILD_DIR, safe_client_build_dir_path
 
 
-RequestContext: type[Context[HTTPServerRequest | None]] = create_context(
-    None, "RequestContext"
+ConnectionContext: type[Context[HTTPServerRequest | None]] = create_context(
+    None, "ConnectionContext"
 )
 
 
@@ -88,19 +89,41 @@ async def serve_development_app(
         await server.close_all_connections()
 
 
+def use_location() -> Location:
+    """Get the current route as a string"""
+    conn = use_connection()
+    search = conn.request.query
+    return Location(pathname="/" + conn.path, search="?" + search if search else "")
+
+
+def use_scope() -> dict[str, Any]:
+    """Get the current WSGI environment dictionary"""
+    return WSGIContainer.environ(use_request())
+
+
 def use_request() -> HTTPServerRequest:
     """Get the current ``HTTPServerRequest``"""
-    request = use_context(RequestContext)
-    if request is None:
+    return use_connection().request
+
+
+def use_connection() -> Connection:
+    connection = use_context(ConnectionContext)
+    if connection is None:
         raise RuntimeError(  # pragma: no cover
-            "No request. Are you running with a Tornado server?"
+            "No connection. Are you running with a Tornado server?"
         )
-    return request
+    return connection
 
 
-def use_scope() -> dict[str, Any]:
-    """Get the current WSGI environment dictionary"""
-    return WSGIContainer.environ(use_request())
+@dataclass
+class Connection:
+    """A simple wrapper for holding connection information"""
+
+    request: HTTPServerRequest
+    """The current request object"""
+
+    path: str
+    """The current path being served"""
 
 
 @dataclass
@@ -179,7 +202,7 @@ class ModelStreamHandler(WebSocketHandler):
     def initialize(self, component_constructor: ComponentConstructor) -> None:
         self._component_constructor = component_constructor
 
-    async def open(self, *args: Any, **kwargs: Any) -> None:
+    async def open(self, path: str) -> None:
         message_queue: "AsyncQueue[str]" = AsyncQueue()
 
         async def send(value: VdomJsonPatch) -> None:
@@ -192,7 +215,10 @@ async def recv() -> LayoutEvent:
         self._dispatch_future = asyncio.ensure_future(
             serve_json_patch(
                 Layout(
-                    RequestContext(self._component_constructor(), value=self.request)
+                    ConnectionContext(
+                        self._component_constructor(),
+                        value=Connection(self.request, path),
+                    )
                 ),
                 send,
                 recv,

src/idom/server/types.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+from dataclasses import dataclass
 from typing import Any, MutableMapping, TypeVar
 
 from typing_extensions import Protocol, runtime_checkable
@@ -37,3 +38,21 @@ async def serve_development_app(
 
     def use_scope(self) -> MutableMapping[str, Any]:
         """Get an ASGI scope or WSGI environment dictionary"""
+
+    def use_location(self) -> Location:
+        """Get the current location (URL)"""
+
+
+@dataclass
+class Location:
+    """Represents the current location (URL)
+
+    Analogous to, but not necessarily identical to, the client-side
+    ``document.location`` object.
+    """
+
+    pathname: str
+    """the path of the URL for the location"""
+
+    search: str = ""
+    """A search or query string - a '?' followed by the parameters of the URL."""

src/idom/server/utils.py

@@ -61,7 +61,7 @@ def run(
 
 def safe_client_build_dir_path(path: str) -> Path:
     """Prevent path traversal out of :data:`CLIENT_BUILD_DIR`"""
-    start, _, end = path.rpartition("/")
+    start, _, end = (path[:-1] if path.endswith("/") else path).rpartition("/")
     file = end or start
     final_path = traversal_safe_path(CLIENT_BUILD_DIR, file)
     return final_path if final_path.is_file() else (CLIENT_BUILD_DIR / "index.html")