Add read_signals helpers to the python SDK

Author: gazpachoking

sdk/python/src/datastar_py/django.py

@@ -1,6 +1,8 @@
+import json
 from functools import wraps
+from typing import Any
 
-from django.http import StreamingHttpResponse as _StreamingHttpResponse
+from django.http import StreamingHttpResponse as _StreamingHttpResponse, HttpRequest
 
 from .sse import SSE_HEADERS, ServerSentEventGenerator
 
@@ -10,3 +12,9 @@ class DatastarStreamingHttpResponse(_StreamingHttpResponse, ServerSentEventGener
     def __init__(self, *args, **kwargs):
         kwargs["headers"] = {**SSE_HEADERS, **kwargs.get("headers", {})}
         super().__init__(*args, **kwargs)
+
+def read_signals(request: HttpRequest) -> dict[str, Any]:
+    if request.method == "GET":
+        return json.loads(request.GET.get("datastar"))
+    else:
+        return json.loads(request.body)

sdk/python/src/datastar_py/fastapi.py

@@ -1 +1,7 @@
-from .starlette import DatastarStreamingResponse, ServerSentEventGenerator
+from typing import Annotated, Any
+
+from fastapi import Depends
+from .starlette import DatastarStreamingResponse, ServerSentEventGenerator, read_signals
+
+
+SignalsDep = Annotated[dict[str, Any], Depends(read_signals)]

sdk/python/src/datastar_py/fasthtml.py

@@ -1,2 +1,2 @@
 from .sse import ServerSentEventGenerator
-from .starlette import DatastarStreamingResponse
+from .starlette import DatastarStreamingResponse, read_signals

sdk/python/src/datastar_py/quart.py

@@ -1,4 +1,7 @@
-from quart import make_response as _make_response
+import json
+from typing import Any
+
+from quart import make_response as _make_response, request
 
 from .sse import ServerSentEventGenerator, SSE_HEADERS
 
@@ -7,3 +10,10 @@ async def make_datastar_response(async_generator):
     response = await _make_response(async_generator, SSE_HEADERS)
     response.timeout = None
     return response
+
+
+async def read_signals() -> dict[str, Any]:
+    if request.method == "GET":
+        return json.loads(request.args.get("datastar"))
+    else:
+        return await request.get_json()

sdk/python/src/datastar_py/sanic.py

@@ -1,4 +1,5 @@
-from typing import TYPE_CHECKING
+import json
+from typing import TYPE_CHECKING, Any
 
 from .sse import SSE_HEADERS, ServerSentEventGenerator
 
@@ -9,3 +10,9 @@
 async def datastar_respond(request: "Request") -> "HTTPResponse":
     response = await request.respond(headers=SSE_HEADERS)
     return response
+
+async def read_signals(request: "Request") -> dict[str, Any]:
+    if request.method == "GET":
+        return json.loads(request.args.get("datastar"))
+    else:
+        return await request.json()

sdk/python/src/datastar_py/starlette.py

@@ -1,5 +1,8 @@
+import json
 from functools import wraps
+from typing import Any
 
+from starlette.requests import Request
 from starlette.responses import StreamingResponse as _StreamingResponse
 
 from .sse import SSE_HEADERS, ServerSentEventGenerator
@@ -10,3 +13,10 @@ class DatastarStreamingResponse(_StreamingResponse, ServerSentEventGenerator):
     def __init__(self, *args, **kwargs):
         kwargs["headers"] = {**SSE_HEADERS, **kwargs.get("headers", {})}
         super().__init__(*args, **kwargs)
+
+
+async def read_signals(request: Request) -> dict[str, Any]:
+    if request.method == "GET":
+        return json.loads(request.query_params.get("datastar"))
+    else:
+        return await request.json()