expose use_scope instead of use_connection

Author: rmorshea

src/idom/server/_conn.py

@@ -28,8 +28,8 @@ async def serve_development_app(
     )
 
 
-def use_connection() -> Any:
-    return _default_implementation().use_connection()
+def use_scope() -> Any:
+    return _default_implementation().use_scope()
 
 
 def _default_implementation() -> ServerImplementation[Any]:

src/idom/server/default.py

@@ -11,15 +11,15 @@
     _setup_options,
     _setup_single_view_dispatcher_route,
     serve_development_app,
-    use_connection,
+    use_scope,
 )
 
 
 __all__ = (
     "configure",
     "serve_development_app",
     "create_development_app",
-    "use_connection",
+    "use_scope",
 )
 
 

src/idom/server/fastapi.py

@@ -9,6 +9,7 @@
 from threading import Thread
 from typing import Any, Callable, Dict, NamedTuple, Optional, Union, cast
 from urllib.parse import parse_qs as parse_query_string
+from wsgiref.types import WSGIEnvironment
 
 from flask import (
     Blueprint,
@@ -29,17 +30,18 @@
 
 import idom
 from idom.config import IDOM_WEB_MODULES_DIR
-from idom.core.hooks import use_context
+from idom.core.hooks import Context, create_context, use_context
 from idom.core.layout import LayoutEvent, LayoutUpdate
 from idom.core.serve import serve_json_patch
 from idom.core.types import ComponentType, RootComponentConstructor
 
-from ._conn import Connection
 from .utils import CLIENT_BUILD_DIR
 
 
 logger = logging.getLogger(__name__)
 
+RequestContext: type[Context[Request | None]] = create_context(None, "RequestContext")
+
 
 def configure(
     app: Flask, component: RootComponentConstructor, options: Options | None = None
@@ -105,12 +107,17 @@ def run_server():
             raise RuntimeError("Failed to shutdown server.")
 
 
-def use_connection() -> Request:
+def use_request() -> Request:
     """Get the current ``Request``"""
-    value = use_context(Connection)
-    if value is None:
-        raise RuntimeError("No established connection.")
-    return value
+    request = use_context(RequestContext)
+    if request is None:
+        raise RuntimeError("No request. Are you running with a Flask server?")
+    return request
+
+
+def use_scope() -> WSGIEnvironment:
+    """Get the current WSGI environment"""
+    return use_request().environ
 
 
 class Options(TypedDict, total=False):
@@ -221,7 +228,9 @@ async def recv_coro() -> Any:
 
         async def main() -> None:
             await serve_json_patch(
-                idom.Layout(Connection(component, value=request)), send_coro, recv_coro
+                idom.Layout(RequestContext(component, value=request)),
+                send_coro,
+                recv_coro,
             )
 
         main_future = asyncio.ensure_future(main())

src/idom/server/flask.py

@@ -9,10 +9,12 @@
 
 from mypy_extensions import TypedDict
 from sanic import Blueprint, Sanic, request, response
+from sanic.models.asgi import ASGIScope
 from sanic_cors import CORS
 from websockets.legacy.protocol import WebSocketCommonProtocol
 
 from idom.config import IDOM_WEB_MODULES_DIR
+from idom.core.hooks import Context, create_context, use_context
 from idom.core.layout import Layout, LayoutEvent
 from idom.core.serve import (
     RecvCoroutine,
@@ -22,12 +24,15 @@
 )
 from idom.core.types import RootComponentConstructor
 
-from ._conn import Connection
 from .utils import CLIENT_BUILD_DIR
 
 
 logger = logging.getLogger(__name__)
 
+RequestContext: type[Context[request.Request | None]] = create_context(
+    None, "RequestContext"
+)
+
 
 def configure(
     app: Sanic, component: RootComponentConstructor, options: Options | None = None
@@ -64,12 +69,17 @@ async def serve_development_app(
         app.stop()
 
 
-def use_connection() -> request.Request:
+def use_request() -> request.Request:
     """Get the current ``Request``"""
-    value = use_connection(Connection)
-    if value is None:
-        raise RuntimeError("No established connection.")
-    return value
+    request = use_context(RequestContext)
+    if request is None:
+        raise RuntimeError("No request. Are you running with a Sanic server?")
+    return request
+
+
+def use_scope() -> ASGIScope:
+    """Get the current ASGI scope"""
+    return use_request().app._asgi_app.transport.scope
 
 
 class Options(TypedDict, total=False):
@@ -132,7 +142,7 @@ async def model_stream(
         send, recv = _make_send_recv_callbacks(socket)
         component_params = {k: request.args.get(k) for k in request.args}
         await serve_json_patch(
-            Layout(Connection(constructor(**component_params), value=request)),
+            Layout(RequestContext(constructor(**component_params), value=request)),
             send,
             recv,
         )

src/idom/server/sanic.py

@@ -3,7 +3,6 @@
 import asyncio
 import json
 import logging
-from asyncio import FIRST_EXCEPTION, CancelledError
 from typing import Any, Dict, Tuple, Union
 
 from mypy_extensions import TypedDict
@@ -12,12 +11,11 @@
 from starlette.requests import Request
 from starlette.responses import RedirectResponse
 from starlette.staticfiles import StaticFiles
+from starlette.types import Scope
 from starlette.websockets import WebSocket, WebSocketDisconnect
-from uvicorn.config import Config as UvicornConfig
-from uvicorn.server import Server as UvicornServer
 
 from idom.config import IDOM_WEB_MODULES_DIR
-from idom.core.hooks import use_context
+from idom.core.hooks import Context, create_context, use_context
 from idom.core.layout import Layout, LayoutEvent
 from idom.core.serve import (
     RecvCoroutine,
@@ -28,12 +26,15 @@
 from idom.core.types import RootComponentConstructor
 
 from ._asgi import serve_development_asgi
-from ._conn import Connection
 from .utils import CLIENT_BUILD_DIR
 
 
 logger = logging.getLogger(__name__)
 
+WebSocketContext: type[Context[WebSocket | None]] = create_context(
+    None, "WebSocketContext"
+)
+
 
 def configure(
     app: Starlette,
@@ -69,12 +70,17 @@ async def serve_development_app(
     await serve_development_asgi(app, host, port, started)
 
 
-def use_connection() -> WebSocket:
-    """Get the current ``WebSocket`` connection"""
-    value = use_context(Connection)
-    if value is None:
-        raise RuntimeError("No established connection.")
-    return value
+def use_websocket() -> WebSocket:
+    """Get the current WebSocket object"""
+    websocket = use_context(WebSocketContext)
+    if websocket is None:
+        raise RuntimeError("No websocket. Are you running with a Starllette server?")
+    return websocket
+
+
+def use_scope() -> Scope:
+    """Get the current ASGI scope dictionary"""
+    return use_websocket().scope
 
 
 class Options(TypedDict, total=False):
@@ -155,7 +161,7 @@ async def model_stream(socket: WebSocket) -> None:
         send, recv = _make_send_recv_callbacks(socket)
         try:
             await serve_json_patch(
-                Layout(Connection(constructor(), value=socket)),
+                Layout(WebSocketContext(constructor(), value=socket)),
                 send,
                 recv,
             )

src/idom/server/starlette.py

@@ -12,18 +12,23 @@
 from tornado.platform.asyncio import AsyncIOMainLoop
 from tornado.web import Application, RedirectHandler, RequestHandler, StaticFileHandler
 from tornado.websocket import WebSocketHandler
+from tornado.wsgi import WSGIContainer
 from typing_extensions import TypedDict
 
 from idom.config import IDOM_WEB_MODULES_DIR
-from idom.core.hooks import use_context
+from idom.core.hooks import Context, create_context, use_context
 from idom.core.layout import Layout, LayoutEvent
 from idom.core.serve import VdomJsonPatch, serve_json_patch
 from idom.core.types import ComponentConstructor
 
-from ._conn import Connection
 from .utils import CLIENT_BUILD_DIR
 
 
+RequestContext: type[Context[HTTPServerRequest | None]] = create_context(
+    None, "RequestContext"
+)
+
+
 def configure(
     app: Application,
     component: ComponentConstructor,
@@ -72,12 +77,17 @@ async def serve_development_app(
         await server.close_all_connections()
 
 
-def use_connection() -> HTTPServerRequest:
+def use_request() -> HTTPServerRequest:
     """Get the current ``HTTPServerRequest``"""
-    value = use_context(Connection)
-    if value is None:
-        raise RuntimeError("No established connection.")
-    return value
+    request = use_context(RequestContext)
+    if request is None:
+        raise RuntimeError("No request. Are you running with a Tornado server?")
+    return request
+
+
+def use_scope() -> dict[str, Any]:
+    """Get the current WSGI environment dictionary"""
+    return WSGIContainer.environ(use_request())
 
 
 class Options(TypedDict, total=False):
@@ -160,7 +170,6 @@ def initialize(self, component_constructor: ComponentConstructor) -> None:
 
     async def open(self, *args: str, **kwargs: str) -> None:
         message_queue: "AsyncQueue[str]" = AsyncQueue()
-        query_params = {k: v[0].decode() for k, v in self.request.arguments.items()}
 
         async def send(value: VdomJsonPatch) -> None:
             await self.write_message(json.dumps(value))
@@ -171,7 +180,9 @@ async def recv() -> LayoutEvent:
         self._message_queue = message_queue
         self._dispatch_future = asyncio.ensure_future(
             serve_json_patch(
-                Layout(Connection(self._component_constructor(), value=self.request)),
+                Layout(
+                    RequestContext(self._component_constructor(), value=self.request)
+                ),
                 send,
                 recv,
             )

src/idom/server/tornado.py

@@ -1,9 +1,11 @@
+from __future__ import annotations
+
 import asyncio
-from typing import Any, TypeVar
+from typing import Any, MutableMapping, TypeVar
 
 from typing_extensions import Protocol, runtime_checkable
 
-from idom.core.types import RootComponentConstructor
+from idom.types import RootComponentConstructor
 
 
 _App = TypeVar("_App")
@@ -24,5 +26,5 @@ async def serve_development_app(
     ) -> None:
         """Run an application using a development server"""
 
-    def use_connection() -> Any:
-        """Get information about a currently active request to the server"""
+    def use_scope() -> MutableMapping[str, Any]:
+        """Get an ASGI scope or WSGI environment dictionary"""