dnrops
/
datastar
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add ruff config and format all files

This commit is contained in:
Chase Sterling 2025-04-17 20:37:21 -04:00
parent b474020d22
commit 03e9524819
9 changed files with 97 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
sdk/python/pyproject.toml
View File

@ -83,3 +83,24 @@ commit_args = ""
setup_hooks = []
pre_commit_hooks = []
post_commit_hooks = []
[tool.ruff]
exclude = [
"src/datastar_py/consts.py",
]
[tool.ruff.lint]
select = [
# pycodestyle
"E",
# Pyflakes
"F",
# pyupgrade
"UP",
# flake8-bugbear
"B",
# flake8-simplify
"SIM",
# isort
"I",
]
fixable = ["ALL"]

2
sdk/python/src/datastar_py/__init__.py
View File

@ -1,3 +1,3 @@
from .sse import SSE_HEADERS, ServerSentEventGenerator
__all__ = ["ServerSentEventGenerator", "SSE_HEADERS"]
__all__ = ["SSE_HEADERS", "ServerSentEventGenerator"]

15
sdk/python/src/datastar_py/django.py
View File

@ -2,10 +2,18 @@ import json
from functools import wraps
from typing import Any
from django.http import StreamingHttpResponse as _StreamingHttpResponse, HttpRequest
from django.http import HttpRequest
from django.http import StreamingHttpResponse as _StreamingHttpResponse
from .sse import SSE_HEADERS, ServerSentEventGenerator
__all__ = [
"SSE_HEADERS",
"DatastarStreamingHttpResponse",
"ServerSentEventGenerator",
"read_signals",
]
class DatastarStreamingHttpResponse(_StreamingHttpResponse, ServerSentEventGenerator):
@wraps(_StreamingHttpResponse.__init__)
@ -13,8 +21,9 @@ class DatastarStreamingHttpResponse(_StreamingHttpResponse, ServerSentEventGener
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"))
return json.loads(request.GET["datastar"])
else:
return json.loads(request.body)
return json.loads(request.body)

12
sdk/python/src/datastar_py/fastapi.py
View File

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

9
sdk/python/src/datastar_py/fasthtml.py
View File

@ -1,2 +1,9 @@
from .sse import ServerSentEventGenerator
from .sse import SSE_HEADERS, ServerSentEventGenerator
from .starlette import DatastarStreamingResponse, read_signals
__all__ = [
"SSE_HEADERS",
"DatastarStreamingResponse",
"ServerSentEventGenerator",
"read_signals",
]

14
sdk/python/src/datastar_py/quart.py
View File

@ -1,9 +1,17 @@
import json
from typing import Any
from quart import make_response as _make_response, request
from quart import make_response as _make_response
from quart import request
from .sse import ServerSentEventGenerator, SSE_HEADERS
from .sse import SSE_HEADERS, ServerSentEventGenerator
__all__ = [
"SSE_HEADERS",
"ServerSentEventGenerator",
"make_datastar_response",
"read_signals",
]
async def make_datastar_response(async_generator):
@ -14,6 +22,6 @@ async def make_datastar_response(async_generator):
async def read_signals() -> dict[str, Any]:
if request.method == "GET":
return json.loads(request.args.get("datastar"))
return json.loads(request.args["datastar"])
else:
return await request.get_json()

12
sdk/python/src/datastar_py/sanic.py
View File

@ -4,15 +4,23 @@ from typing import TYPE_CHECKING, Any
from .sse import SSE_HEADERS, ServerSentEventGenerator
if TYPE_CHECKING:
from sanic import Request, HTTPResponse
from sanic import HTTPResponse, Request
__all__ = [
"SSE_HEADERS",
"ServerSentEventGenerator",
"datastar_respond",
"read_signals",
]
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"))
return json.loads(request.args["datastar"])
else:
return await request.json()

20
sdk/python/src/datastar_py/sse.py
View File

@ -60,13 +60,19 @@ class ServerSentEventGenerator:
fragments = fragments.__html__()
data_lines = []
if merge_mode:
data_lines.append(f"data: {consts.MERGE_MODE_DATALINE_LITERAL} {merge_mode}")
data_lines.append(
f"data: {consts.MERGE_MODE_DATALINE_LITERAL} {merge_mode}"
)
if selector:
data_lines.append(f"data: {consts.SELECTOR_DATALINE_LITERAL} {selector}")
if use_view_transition:
data_lines.append(f"data: {consts.USE_VIEW_TRANSITION_DATALINE_LITERAL} true")
data_lines.append(
f"data: {consts.USE_VIEW_TRANSITION_DATALINE_LITERAL} true"
)
else:
data_lines.append(f"data: {consts.USE_VIEW_TRANSITION_DATALINE_LITERAL} false")
data_lines.append(
f"data: {consts.USE_VIEW_TRANSITION_DATALINE_LITERAL} false"
)
data_lines.extend(
f"data: {consts.FRAGMENTS_DATALINE_LITERAL} {x}"
@ -92,9 +98,13 @@ class ServerSentEventGenerator:
if selector:
data_lines.append(f"data: {consts.SELECTOR_DATALINE_LITERAL} {selector}")
if use_view_transition:
data_lines.append(f"data: {consts.USE_VIEW_TRANSITION_DATALINE_LITERAL} true")
data_lines.append(
f"data: {consts.USE_VIEW_TRANSITION_DATALINE_LITERAL} true"
)
else:
data_lines.append(f"data: {consts.USE_VIEW_TRANSITION_DATALINE_LITERAL} false")
data_lines.append(
f"data: {consts.USE_VIEW_TRANSITION_DATALINE_LITERAL} false"
)
return ServerSentEventGenerator._send(
consts.EventType.REMOVE_FRAGMENTS,

9
sdk/python/src/datastar_py/starlette.py
View File

@ -7,6 +7,13 @@ from starlette.responses import StreamingResponse as _StreamingResponse
from .sse import SSE_HEADERS, ServerSentEventGenerator
__all__ = [
"SSE_HEADERS",
"DatastarStreamingResponse",
"ServerSentEventGenerator",
"read_signals",
]
class DatastarStreamingResponse(_StreamingResponse, ServerSentEventGenerator):
@wraps(_StreamingResponse.__init__)
@ -17,6 +24,6 @@ class DatastarStreamingResponse(_StreamingResponse, ServerSentEventGenerator):
async def read_signals(request: Request) -> dict[str, Any]:
if request.method == "GET":
return json.loads(request.query_params.get("datastar"))
return json.loads(request.query_params["datastar"])
else:
return await request.json()