diff --git a/README.md b/README.md index 63d2653..9d9ac3c 100644 --- a/README.md +++ b/README.md @@ -201,3 +201,27 @@ data = AttributeGenerator(alias="data-star-") data = AttributeGenerator(alias="_data-") html.button("My Button", **data.on("click", "console.log('clicked')").debounce("1s").stop) ``` + +## Action Generation Helper + +datastar-py also includes helpers to generator Datastar actions and compose them with generated +attributes. + +```python +from datastar_py import attribute_generator as data +from datastar_py import action_generator as actions + +submit = data.on( + "submit", + actions.post( + "/contacts", + content_type="form", + retry="never", + request_cancellation="cleanup", + headers={"X-View": data.JSExpression("el.dataset.viewId")}, + ), +) +``` + +The fetch helpers are `get`, `post`, `put`, `patch`, and `delete` with keyword-only options for +signal filters, form selection, headers, payloads, retries, visibility, and request cancellation. diff --git a/src/datastar_py/__init__.py b/src/datastar_py/__init__.py index d83540a..c041c24 100644 --- a/src/datastar_py/__init__.py +++ b/src/datastar_py/__init__.py @@ -4,10 +4,10 @@ from collections.abc import Mapping from typing import Any -from .attributes import attribute_generator +from .attributes import action_generator, attribute_generator from .sse import SSE_HEADERS, ServerSentEventGenerator -__all__ = ["SSE_HEADERS", "ServerSentEventGenerator", "attribute_generator"] +__all__ = ["SSE_HEADERS", "ServerSentEventGenerator", "action_generator", "attribute_generator"] def _read_signals( diff --git a/src/datastar_py/attributes.py b/src/datastar_py/attributes.py index 49e081b..e531fb1 100644 --- a/src/datastar_py/attributes.py +++ b/src/datastar_py/attributes.py @@ -3,6 +3,7 @@ import collections.abc import dataclasses import json +import math import re from collections.abc import Iterable, Iterator, Mapping from itertools import chain @@ -162,7 +163,117 @@ def _as_javascript_expressions(value: object) -> object: return value +def _fetch(method: str, url: str | JSExpression, options: dict[str, object]) -> str: + _validate_fetch(url, options) + names = { + "content_type": "contentType", + "selector": "selector", + "headers": "headers", + "open_when_hidden": "openWhenHidden", + "payload": "payload", + "retry": "retry", + "retry_interval": "retryInterval", + "retry_scaler": "retryScaler", + "retry_max_wait": "retryMaxWait", + "retry_max_count": "retryMaxCount", + "request_cancellation": "requestCancellation", + } + mapped = { + names[key]: value for key, value in options.items() if key in names and value is not None + } + filters = _filters(options.get("include_signals"), options.get("exclude_signals")) + if filters: + mapped["filterSignals"] = filters + return f"@{method}({javascript(url)}{', ' + javascript(mapped) if mapped else ''})" + + +def _validate_fetch(url: str | JSExpression, options: dict[str, object]) -> None: + if not isinstance(url, JSExpression): + _require_nonblank_string("url", url) + _validate_choices(options) + _validate_content_options(options) + _validate_headers(options["headers"]) + _validate_retries(options) + value = options["open_when_hidden"] + if value is not None and not isinstance(value, bool): + raise TypeError("open_when_hidden must be a boolean") + + +def _validate_choices(options: dict[str, object]) -> None: + for name, choices in ( + ("content_type", ("json", "form")), + ("retry", ("auto", "always", "never", "error")), + ("request_cancellation", ("auto", "disabled", "cleanup")), + ): + value = options[name] + if value is None or (name == "request_cancellation" and isinstance(value, JSExpression)): + continue + if not isinstance(value, str): + suffix = " or JSExpression" if name == "request_cancellation" else "" + raise TypeError(f"{name} must be a string{suffix}") + if value not in choices: + raise ValueError(f"{name} must be one of: {', '.join(choices)}") + + +def _validate_content_options(options: dict[str, object]) -> None: + selector = options["selector"] + if selector is not None: + _require_nonblank_string("selector", selector) + if options["content_type"] != "form": + raise ValueError("selector requires content_type='form'") + if options["payload"] is not None and options["content_type"] == "form": + raise ValueError("payload requires JSON content_type") + + +def _validate_headers(headers: object) -> None: + if headers is None: + return + if not isinstance(headers, collections.abc.Mapping): + raise TypeError("headers must be a mapping") + for name, value in headers.items(): + _require_nonblank_string("header name", name) + if not isinstance(value, str | JSExpression): + raise TypeError("header values must be strings or JSExpression") + + +def _validate_retries(options: dict[str, object]) -> None: + for name in ("retry_interval", "retry_max_wait", "retry_max_count"): + value = options[name] + if value is None: + continue + if not isinstance(value, int) or isinstance(value, bool): + raise TypeError(f"{name} must be an integer") + if value < 0: + raise ValueError(f"{name} must be non-negative") + value = options["retry_scaler"] + if value is None: + return + if not isinstance(value, int | float) or isinstance(value, bool): + raise TypeError("retry_scaler must be a number") + try: + finite = math.isfinite(value) + except OverflowError: + finite = False + if not finite or value < 0: + raise ValueError("retry_scaler must be finite and non-negative") + + +def _filters(include: object, exclude: object) -> dict[str, JSExpression]: + for name, value in (("include", include), ("exclude", exclude)): + if value is not None and not isinstance(value, str | JSExpression): + raise TypeError(f"{name} filter must be a string or JSExpression") + return { + key: value + if isinstance(value, JSExpression) + else JSExpression(f"new RegExp({javascript(value)})") + for key, value in (("include", include), ("exclude", exclude)) + if value is not None + } + + class AttributeGenerator: + JSExpression = JSExpression + def __init__(self, alias: str = "data-") -> None: """A helper which can generate all the Datastar attributes. @@ -798,3 +909,228 @@ def _filter_dict(include: str | None = None, exclude: str | None = None) -> dict attribute_generator = AttributeGenerator() + + +class ActionGenerator: + """A namespace for generating Datastar action expressions.""" + + @staticmethod + def get( # noqa: PLR0913 + url: str | JSExpression, + *, + content_type: Literal["json", "form"] | None = None, + include_signals: str | JSExpression | None = None, + exclude_signals: str | JSExpression | None = None, + selector: str | None = None, + headers: collections.abc.Mapping[str, str | JSExpression] | None = None, + open_when_hidden: bool | None = None, + payload: object | None = None, + retry: Literal["auto", "always", "never", "error"] | None = None, + retry_interval: int | None = None, + retry_scaler: float | None = None, + retry_max_wait: int | None = None, + retry_max_count: int | None = None, + request_cancellation: Literal["auto", "disabled", "cleanup"] | JSExpression | None = None, + ) -> str: + """Build a @get expression; options set to None are omitted.""" + return _fetch( + "get", + url, + { + "content_type": content_type, + "include_signals": include_signals, + "exclude_signals": exclude_signals, + "selector": selector, + "headers": headers, + "open_when_hidden": open_when_hidden, + "payload": payload, + "retry": retry, + "retry_interval": retry_interval, + "retry_scaler": retry_scaler, + "retry_max_wait": retry_max_wait, + "retry_max_count": retry_max_count, + "request_cancellation": request_cancellation, + }, + ) + + @staticmethod + def post( # noqa: PLR0913 + url: str | JSExpression, + *, + content_type: Literal["json", "form"] | None = None, + include_signals: str | JSExpression | None = None, + exclude_signals: str | JSExpression | None = None, + selector: str | None = None, + headers: collections.abc.Mapping[str, str | JSExpression] | None = None, + open_when_hidden: bool | None = None, + payload: object | None = None, + retry: Literal["auto", "always", "never", "error"] | None = None, + retry_interval: int | None = None, + retry_scaler: float | None = None, + retry_max_wait: int | None = None, + retry_max_count: int | None = None, + request_cancellation: Literal["auto", "disabled", "cleanup"] | JSExpression | None = None, + ) -> str: + """Build a @post expression; options set to None are omitted.""" + return _fetch( + "post", + url, + { + "content_type": content_type, + "include_signals": include_signals, + "exclude_signals": exclude_signals, + "selector": selector, + "headers": headers, + "open_when_hidden": open_when_hidden, + "payload": payload, + "retry": retry, + "retry_interval": retry_interval, + "retry_scaler": retry_scaler, + "retry_max_wait": retry_max_wait, + "retry_max_count": retry_max_count, + "request_cancellation": request_cancellation, + }, + ) + + @staticmethod + def put( # noqa: PLR0913 + url: str | JSExpression, + *, + content_type: Literal["json", "form"] | None = None, + include_signals: str | JSExpression | None = None, + exclude_signals: str | JSExpression | None = None, + selector: str | None = None, + headers: collections.abc.Mapping[str, str | JSExpression] | None = None, + open_when_hidden: bool | None = None, + payload: object | None = None, + retry: Literal["auto", "always", "never", "error"] | None = None, + retry_interval: int | None = None, + retry_scaler: float | None = None, + retry_max_wait: int | None = None, + retry_max_count: int | None = None, + request_cancellation: Literal["auto", "disabled", "cleanup"] | JSExpression | None = None, + ) -> str: + """Build a @put expression; options set to None are omitted.""" + return _fetch( + "put", + url, + { + "content_type": content_type, + "include_signals": include_signals, + "exclude_signals": exclude_signals, + "selector": selector, + "headers": headers, + "open_when_hidden": open_when_hidden, + "payload": payload, + "retry": retry, + "retry_interval": retry_interval, + "retry_scaler": retry_scaler, + "retry_max_wait": retry_max_wait, + "retry_max_count": retry_max_count, + "request_cancellation": request_cancellation, + }, + ) + + @staticmethod + def patch( # noqa: PLR0913 + url: str | JSExpression, + *, + content_type: Literal["json", "form"] | None = None, + include_signals: str | JSExpression | None = None, + exclude_signals: str | JSExpression | None = None, + selector: str | None = None, + headers: collections.abc.Mapping[str, str | JSExpression] | None = None, + open_when_hidden: bool | None = None, + payload: object | None = None, + retry: Literal["auto", "always", "never", "error"] | None = None, + retry_interval: int | None = None, + retry_scaler: float | None = None, + retry_max_wait: int | None = None, + retry_max_count: int | None = None, + request_cancellation: Literal["auto", "disabled", "cleanup"] | JSExpression | None = None, + ) -> str: + """Build a @patch expression; options set to None are omitted.""" + return _fetch( + "patch", + url, + { + "content_type": content_type, + "include_signals": include_signals, + "exclude_signals": exclude_signals, + "selector": selector, + "headers": headers, + "open_when_hidden": open_when_hidden, + "payload": payload, + "retry": retry, + "retry_interval": retry_interval, + "retry_scaler": retry_scaler, + "retry_max_wait": retry_max_wait, + "retry_max_count": retry_max_count, + "request_cancellation": request_cancellation, + }, + ) + + @staticmethod + def delete( # noqa: PLR0913 + url: str | JSExpression, + *, + content_type: Literal["json", "form"] | None = None, + include_signals: str | JSExpression | None = None, + exclude_signals: str | JSExpression | None = None, + selector: str | None = None, + headers: collections.abc.Mapping[str, str | JSExpression] | None = None, + open_when_hidden: bool | None = None, + payload: object | None = None, + retry: Literal["auto", "always", "never", "error"] | None = None, + retry_interval: int | None = None, + retry_scaler: float | None = None, + retry_max_wait: int | None = None, + retry_max_count: int | None = None, + request_cancellation: Literal["auto", "disabled", "cleanup"] | JSExpression | None = None, + ) -> str: + """Build a @delete expression; options set to None are omitted.""" + return _fetch( + "delete", + url, + { + "content_type": content_type, + "include_signals": include_signals, + "exclude_signals": exclude_signals, + "selector": selector, + "headers": headers, + "open_when_hidden": open_when_hidden, + "payload": payload, + "retry": retry, + "retry_interval": retry_interval, + "retry_scaler": retry_scaler, + "retry_max_wait": retry_max_wait, + "retry_max_count": retry_max_count, + "request_cancellation": request_cancellation, + }, + ) + + @staticmethod + def peek(expression: JSExpression) -> str: + if not isinstance(expression, JSExpression): + raise TypeError("expression must be a JSExpression") + return f"@peek(() => ({expression.value}))" + + @staticmethod + def set_all( + value: object, + include: str | JSExpression | None = None, + exclude: str | JSExpression | None = None, + ) -> str: + filters = _filters(include, exclude) + return f"@setAll({javascript(value)}{', ' + javascript(filters) if filters else ''})" + + @staticmethod + def toggle_all( + include: str | JSExpression | None = None, + exclude: str | JSExpression | None = None, + ) -> str: + filters = _filters(include, exclude) + return f"@toggleAll({javascript(filters) if filters else ''})" + + +action_generator = ActionGenerator() diff --git a/tests/test_actions.py b/tests/test_actions.py new file mode 100644 index 0000000..ec131c0 --- /dev/null +++ b/tests/test_actions.py @@ -0,0 +1,136 @@ +import inspect +import math + +import pytest + +from datastar_py import action_generator as actions +from datastar_py import attribute_generator as ds + +FETCH_METHODS = [ + actions.get, + actions.post, + actions.put, + actions.patch, + actions.delete, +] + + +@pytest.mark.parametrize("method", FETCH_METHODS) +def test_fetch_methods_serialize_literal_urls(method) -> None: + url = "/a'\"\\\n/$token/雪/_!EXPR!_" + assert method(url) == f'@{method.__name__}("/a\'\\"\\\\\\n/$token/\\u96ea/_!EXPR!_")' + + +def test_fetch_options_serialize_with_javascript_types() -> None: + assert actions.post( + ds.JSExpression('"/expression"'), + content_type="json", + headers={ + "literal": 'quoted "text"', + "expression": ds.JSExpression('"a" + "b"'), + }, + open_when_hidden=False, + payload={"nested": [True, None, ds.JSExpression("1 + 2")]}, + retry="never", + retry_interval=0, + retry_scaler=1.5, + retry_max_wait=20, + retry_max_count=0, + request_cancellation=ds.JSExpression("controller"), + include_signals=r"^a/\w+\n", + exclude_signals=ds.JSExpression(r"/private/i"), + ) == ( + '@post(("/expression"), {"contentType": "json", ' + '"headers": {"literal": "quoted \\"text\\"", "expression": ("a" + "b")}, ' + '"openWhenHidden": false, "payload": {"nested": [true, null, (1 + 2)]}, ' + '"retry": "never", "retryInterval": 0, "retryScaler": 1.5, ' + '"retryMaxWait": 20, "retryMaxCount": 0, ' + '"requestCancellation": (controller), ' + '"filterSignals": {"include": (new RegExp("^a/\\\\w+\\\\n")), ' + '"exclude": (/private/i)}})' + ) + + +@pytest.mark.parametrize("method", FETCH_METHODS) +def test_none_omits_options_and_expressions_can_send_null(method) -> None: + options = {name: None for name in inspect.signature(method).parameters if name != "url"} + assert method("/", **options) == method("/") + assert method("/", payload=ds.JSExpression("null")) == ( + f'@{method.__name__}("/", {{"payload": (null)}})' + ) + + +def test_form_and_empty_options_are_preserved() -> None: + assert actions.post("/", content_type="form", selector="#form") == ( + '@post("/", {"contentType": "form", "selector": "#form"})' + ) + assert actions.get( + "/", open_when_hidden=False, retry_max_count=0, headers={}, include_signals="" + ) == ( + '@get("/", {"headers": {}, "openWhenHidden": false, "retryMaxCount": 0, ' + '"filterSignals": {"include": (new RegExp(""))}})' + ) + + +def test_signal_actions_serialize_data_expressions_and_filters() -> None: + assert actions.set_all( + {"literal": "$token", "expression": ds.JSExpression("1 + 1")}, + include="^public", + exclude="private", + ) == ( + '@setAll({"literal": "$token", "expression": (1 + 1)}, ' + '{"include": (new RegExp("^public")), ' + '"exclude": (new RegExp("private"))})' + ) + assert actions.toggle_all() == "@toggleAll()" + assert actions.peek(ds.JSExpression("{answer: 42}")) == ("@peek(() => ({answer: 42}))") + + +def test_fetch_signatures_expose_the_same_keyword_only_options() -> None: + expected = inspect.signature(actions.get) + for method in FETCH_METHODS: + assert inspect.signature(method) == expected + assert all( + parameter.kind == inspect.Parameter.KEYWORD_ONLY + for name, parameter in expected.parameters.items() + if name != "url" + ) + with pytest.raises(TypeError): + method("/", not_defined_kwarg=True) + + +@pytest.mark.parametrize( + ("options", "error", "message"), + ( + ({"content_type": "xml"}, ValueError, "content_type"), + ({"retry": "sometimes"}, ValueError, "retry"), + ({"request_cancellation": "controller"}, ValueError, "request_cancellation"), + ({"open_when_hidden": 0}, TypeError, "open_when_hidden"), + ({"headers": []}, TypeError, "headers"), + ({"headers": {1: "value"}}, TypeError, "header name"), + ({"headers": {"X-Value": False}}, TypeError, "header values"), + ({"include_signals": []}, TypeError, "include"), + ({"selector": ""}, ValueError, "selector"), + ({"selector": "#form"}, ValueError, "content_type='form'"), + ({"content_type": "form", "payload": {}}, ValueError, "payload"), + ({"retry_interval": True}, TypeError, "retry_interval"), + ({"retry_max_count": -1}, ValueError, "retry_max_count"), + ({"retry_scaler": "2"}, TypeError, "retry_scaler"), + ({"retry_scaler": math.inf}, ValueError, "retry_scaler"), + ), +) +def test_fetch_validation(options, error, message) -> None: + with pytest.raises(error, match=message): + actions.post("/", **options) + + +@pytest.mark.parametrize(("url", "error"), ((None, TypeError), ("", ValueError))) +def test_url_validation(url, error) -> None: + with pytest.raises(error, match="url"): + actions.get(url) + + +def test_action_wrapper_validation() -> None: + assert not hasattr(actions, "JSRegex") + with pytest.raises(TypeError, match="JSExpression"): + actions.peek("$token")