Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,6 +111,33 @@
]


class JSExpression(str):
MARKER = "_!EXPR!_"
REPLACEMENT = ""

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


class JSRegex(str):
MARKER = "_!REGEX!_"
REPLACEMENT = "/"

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


def _js_object(obj: dict) -> str:
"""Create a JS object where the values can be JS expressions or regex."""
result = json.dumps(obj)
return (
result.replace(f'"{JSExpression.MARKER}', JSExpression.REPLACEMENT)
.replace(f'{JSExpression.MARKER}"', JSExpression.REPLACEMENT)
.replace(f'"{JSRegex.MARKER}', JSRegex.REPLACEMENT)
.replace(f'{JSRegex.MARKER}"', JSRegex.REPLACEMENT)
)


class AttributeGenerator:
def __init__(self, alias: str = "data-") -> None:
"""A helper which can generate all the Datastar attributes.
Expand All@@ -134,7 +161,11 @@ def signals(
rather than literals.
"""
signals = {**(signals_dict if signals_dict else {}), **signals}
val = _js_object(signals) if expressions_ else json.dumps(signals)
val = (
_js_object({k: JSExpression(v) for k, v in signals.items()})
if expressions_
else json.dumps(signals)
)
return SignalsAttr(value=val, alias=self._alias)

def computed(self, computed_dict: Mapping | None = None, /, **computed: str) -> BaseAttr:
Expand All@@ -159,7 +190,11 @@ def ignore(self) -> IgnoreAttr:
def attr(self, attr_dict: Mapping | None = None, /, **attrs: str) -> BaseAttr:
"""Set the value of any HTML attributes to expressions, and keep them in sync."""
attrs = {**(attr_dict if attr_dict else {}), **attrs}
return BaseAttr("attr", value=_js_object(attrs), alias=self._alias)
return BaseAttr(
"attr",
value=_js_object({k: JSExpression(v) for k, v in attrs.items()}),
alias=self._alias,
)

def bind(self, signal_name: str) -> BaseAttr:
"""Set up two-way data binding between a signal and an element's value."""
Expand All@@ -168,7 +203,11 @@ def bind(self, signal_name: str) -> BaseAttr:
def class_(self, class_dict: Mapping | None = None, /, **classes: str) -> BaseAttr:
"""Add or removes classes to or from an element based on expressions."""
classes = {**(class_dict if class_dict else {}), **classes}
return BaseAttr("class", value=_js_object(classes), alias=self._alias)
return BaseAttr(
"class",
value=_js_object({k: JSExpression(v) for k, v in classes.items()}),
alias=self._alias,
)

@overload
def on(self, event: Literal["interval"], expression: str) -> OnIntervalAttr: ...
Expand DownExpand Up@@ -259,7 +298,11 @@ def show(self, expression: str) -> BaseAttr:
def style(self, style_dict: Mapping | None = None, /, **styles: str) -> BaseAttr:
"""Sets the value of inline CSS styles on an element based on an expression, and keeps them in sync."""
styles = {**(style_dict if style_dict else {}), **styles}
return BaseAttr("style", value=_js_object(styles), alias=self._alias)
return BaseAttr(
"style",
value=_js_object({k: JSExpression(v) for k, v in styles.items()}),
alias=self._alias,
)

def text(self, expression: str) -> BaseAttr:
"""Bind the text content of an element to an expression."""
Expand DownExpand Up@@ -728,16 +771,4 @@ def _escape(s: str) -> str:
)


def _js_object(obj: dict) -> str:
"""Create a JS object where the values are expressions rather than strings."""
return (
"{"
+ ", ".join(
f"{json.dumps(k)}: {_js_object(v) if isinstance(v, dict) else v}"
for k, v in obj.items()
)
+ "}"
)


attribute_generator = AttributeGenerator()
125 changes: 125 additions & 0 deletions src/datastar_py/attributes/actions.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
from __future__ import annotations

from typing import Literal, TypedDict, Unpack

from datastar_py.attributes import JSExpression, JSRegex, SignalValue, _js_object


class _FetchOptions(TypedDict, total=False):
content_type: Literal["json", "form"]
include_signals: str
exclude_signals: str
selector: str
headers: dict[str, str]
open_when_hidden: bool
retry_interval: int
retry_scalar: float
retry_max_wait_ms: int
retry_max_count: int
request_cancellation: Literal["auto", "disabled"] | str


def _fetch(
method: Literal["get", "post", "put", "patch", "delete"],
url: str,
**options: Unpack[_FetchOptions],
) -> str:
result = f"@{method}('{url}'"
if options:
mapped_options = {}
if "content_type" in options:
mapped_options["contentType"] = options["content_type"]
if "include_signals" in options or "exclude_signals" in options:
filter_signals = {}
if "include_signals" in options:
filter_signals["include"] = JSRegex(options["include_signals"])
if "exclude_signals" in options:
filter_signals["exclude"] = JSRegex(options["exclude_signals"])
mapped_options["filterSignals"] = filter_signals
if "selector" in options:
mapped_options["selector"] = options["selector"]
if "headers" in options:
mapped_options["headers"] = _js_object(options["headers"])
if "open_when_hidden" in options:
mapped_options["openWhenHidden"] = options["open_when_hidden"]
if "retry_interval" in options:
mapped_options["retryInterval"] = options["retry_interval"]
if "retry_scalar" in options:
mapped_options["retryScalar"] = options["retry_scalar"]
if "retry_max_wait_ms" in options:
mapped_options["retryMaxWaitMs"] = options["retry_max_wait_ms"]
if "request_cancellation" in options:
if options["request_cancellation"] in ("auto", "disabled"):
mapped_options["requestCancellation"] = options["request_cancellation"]
else:
mapped_options["requestCancellation"] = JSExpression(
options["request_cancellation"]
)
result += f", {_js_object(mapped_options)}"
result += ")"
return result


def get(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("get", url, **options)


def post(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("post", url, **options)


def put(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("put", url, **options)


def patch(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("patch", url, **options)


def delete(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("delete", url, **options)


def peek(expression: str) -> str:
"""Evaluate an expression containing signals without subscribing to changes in those signals."""
return f"@peek(() => {expression})"


def set_all(value: SignalValue, include: str | None = None, exclude: str | None = None) -> str:
"""Set the value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = f", {_js_object(filter_dict)}" if filter_dict else ""
return f"@setAll({value}{filter_string})"


def toggle_all(include: str | None = None, exclude: str | None = None) -> str:
"""Toggle the boolean value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = _js_object(filter_dict) if filter_dict else ""
return f"@toggleAll({filter_string})"


def clipboard(text: str, is_base_64: bool = False) -> str:
"""PRO: Copy text to the clipboard."""
return f"@clipboard({text}{', true' if is_base_64 else ''})"


def fit(
value: float | str,
old_min: float | str,
old_max: float | str,
new_min: float | str,
new_max: float | str,
should_clamp: bool = False,
should_round: bool = False,
) -> str:
"""PRO: Linearly interpolate a value from one range to another."""
return f"@fit({value}, {old_min}, {old_max}, {new_min}, {new_max}, {'true' if should_clamp else 'false'}{', true' if should_round else ''})"
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,6 +111,33 @@
]


class JSExpression(str):
MARKER = "_!EXPR!_"
REPLACEMENT = ""

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


class JSRegex(str):
MARKER = "_!REGEX!_"
REPLACEMENT = "/"

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


def _js_object(obj: dict) -> str:
"""Create a JS object where the values can be JS expressions or regex."""
result = json.dumps(obj)
return (
result.replace(f'"{JSExpression.MARKER}', JSExpression.REPLACEMENT)
.replace(f'{JSExpression.MARKER}"', JSExpression.REPLACEMENT)
.replace(f'"{JSRegex.MARKER}', JSRegex.REPLACEMENT)
.replace(f'{JSRegex.MARKER}"', JSRegex.REPLACEMENT)
)


class AttributeGenerator:
def __init__(self, alias: str = "data-") -> None:
"""A helper which can generate all the Datastar attributes.
Expand All@@ -134,7 +161,11 @@ def signals(
rather than literals.
"""
signals = {**(signals_dict if signals_dict else {}), **signals}
val = _js_object(signals) if expressions_ else json.dumps(signals)
val = (
_js_object({k: JSExpression(v) for k, v in signals.items()})
if expressions_
else json.dumps(signals)
)
return SignalsAttr(value=val, alias=self._alias)

def computed(self, computed_dict: Mapping | None = None, /, **computed: str) -> BaseAttr:
Expand All@@ -159,7 +190,11 @@ def ignore(self) -> IgnoreAttr:
def attr(self, attr_dict: Mapping | None = None, /, **attrs: str) -> BaseAttr:
"""Set the value of any HTML attributes to expressions, and keep them in sync."""
attrs = {**(attr_dict if attr_dict else {}), **attrs}
return BaseAttr("attr", value=_js_object(attrs), alias=self._alias)
return BaseAttr(
"attr",
value=_js_object({k: JSExpression(v) for k, v in attrs.items()}),
alias=self._alias,
)

def bind(self, signal_name: str) -> BaseAttr:
"""Set up two-way data binding between a signal and an element's value."""
Expand All@@ -168,7 +203,11 @@ def bind(self, signal_name: str) -> BaseAttr:
def class_(self, class_dict: Mapping | None = None, /, **classes: str) -> BaseAttr:
"""Add or removes classes to or from an element based on expressions."""
classes = {**(class_dict if class_dict else {}), **classes}
return BaseAttr("class", value=_js_object(classes), alias=self._alias)
return BaseAttr(
"class",
value=_js_object({k: JSExpression(v) for k, v in classes.items()}),
alias=self._alias,
)

@overload
def on(self, event: Literal["interval"], expression: str) -> OnIntervalAttr: ...
Expand DownExpand Up@@ -259,7 +298,11 @@ def show(self, expression: str) -> BaseAttr:
def style(self, style_dict: Mapping | None = None, /, **styles: str) -> BaseAttr:
"""Sets the value of inline CSS styles on an element based on an expression, and keeps them in sync."""
styles = {**(style_dict if style_dict else {}), **styles}
return BaseAttr("style", value=_js_object(styles), alias=self._alias)
return BaseAttr(
"style",
value=_js_object({k: JSExpression(v) for k, v in styles.items()}),
alias=self._alias,
)

def text(self, expression: str) -> BaseAttr:
"""Bind the text content of an element to an expression."""
Expand DownExpand Up@@ -728,16 +771,4 @@ def _escape(s: str) -> str:
)


def _js_object(obj: dict) -> str:
"""Create a JS object where the values are expressions rather than strings."""
return (
"{"
+ ", ".join(
f"{json.dumps(k)}: {_js_object(v) if isinstance(v, dict) else v}"
for k, v in obj.items()
)
+ "}"
)


attribute_generator = AttributeGenerator()
125 changes: 125 additions & 0 deletions src/datastar_py/attributes/actions.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
from __future__ import annotations

from typing import Literal, TypedDict, Unpack

from datastar_py.attributes import JSExpression, JSRegex, SignalValue, _js_object


class _FetchOptions(TypedDict, total=False):
content_type: Literal["json", "form"]
include_signals: str
exclude_signals: str
selector: str
headers: dict[str, str]
open_when_hidden: bool
retry_interval: int
retry_scalar: float
retry_max_wait_ms: int
retry_max_count: int
request_cancellation: Literal["auto", "disabled"] | str


def _fetch(
method: Literal["get", "post", "put", "patch", "delete"],
url: str,
**options: Unpack[_FetchOptions],
) -> str:
result = f"@{method}('{url}'"
if options:
mapped_options = {}
if "content_type" in options:
mapped_options["contentType"] = options["content_type"]
if "include_signals" in options or "exclude_signals" in options:
filter_signals = {}
if "include_signals" in options:
filter_signals["include"] = JSRegex(options["include_signals"])
if "exclude_signals" in options:
filter_signals["exclude"] = JSRegex(options["exclude_signals"])
mapped_options["filterSignals"] = filter_signals
if "selector" in options:
mapped_options["selector"] = options["selector"]
if "headers" in options:
mapped_options["headers"] = _js_object(options["headers"])
if "open_when_hidden" in options:
mapped_options["openWhenHidden"] = options["open_when_hidden"]
if "retry_interval" in options:
mapped_options["retryInterval"] = options["retry_interval"]
if "retry_scalar" in options:
mapped_options["retryScalar"] = options["retry_scalar"]
if "retry_max_wait_ms" in options:
mapped_options["retryMaxWaitMs"] = options["retry_max_wait_ms"]
if "request_cancellation" in options:
if options["request_cancellation"] in ("auto", "disabled"):
mapped_options["requestCancellation"] = options["request_cancellation"]
else:
mapped_options["requestCancellation"] = JSExpression(
options["request_cancellation"]
)
result += f", {_js_object(mapped_options)}"
result += ")"
return result


def get(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("get", url, **options)


def post(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("post", url, **options)


def put(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("put", url, **options)


def patch(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("patch", url, **options)


def delete(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("delete", url, **options)


def peek(expression: str) -> str:
"""Evaluate an expression containing signals without subscribing to changes in those signals."""
return f"@peek(() => {expression})"


def set_all(value: SignalValue, include: str | None = None, exclude: str | None = None) -> str:
"""Set the value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = f", {_js_object(filter_dict)}" if filter_dict else ""
return f"@setAll({value}{filter_string})"


def toggle_all(include: str | None = None, exclude: str | None = None) -> str:
"""Toggle the boolean value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = _js_object(filter_dict) if filter_dict else ""
return f"@toggleAll({filter_string})"


def clipboard(text: str, is_base_64: bool = False) -> str:
"""PRO: Copy text to the clipboard."""
return f"@clipboard({text}{', true' if is_base_64 else ''})"


def fit(
value: float | str,
old_min: float | str,
old_max: float | str,
new_min: float | str,
new_max: float | str,
should_clamp: bool = False,
should_round: bool = False,
) -> str:
"""PRO: Linearly interpolate a value from one range to another."""
return f"@fit({value}, {old_min}, {old_max}, {new_min}, {new_max}, {'true' if should_clamp else 'false'}{', true' if should_round else ''})"
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix"); } } catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,6 +111,33 @@
]


class JSExpression(str):
MARKER = "_!EXPR!_"
REPLACEMENT = ""

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


class JSRegex(str):
MARKER = "_!REGEX!_"
REPLACEMENT = "/"

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


def _js_object(obj: dict) -> str:
"""Create a JS object where the values can be JS expressions or regex."""
result = json.dumps(obj)
return (
result.replace(f'"{JSExpression.MARKER}', JSExpression.REPLACEMENT)
.replace(f'{JSExpression.MARKER}"', JSExpression.REPLACEMENT)
.replace(f'"{JSRegex.MARKER}', JSRegex.REPLACEMENT)
.replace(f'{JSRegex.MARKER}"', JSRegex.REPLACEMENT)
)


class AttributeGenerator:
def __init__(self, alias: str = "data-") -> None:
"""A helper which can generate all the Datastar attributes.
Expand All@@ -134,7 +161,11 @@ def signals(
rather than literals.
"""
signals = {**(signals_dict if signals_dict else {}), **signals}
val = _js_object(signals) if expressions_ else json.dumps(signals)
val = (
_js_object({k: JSExpression(v) for k, v in signals.items()})
if expressions_
else json.dumps(signals)
)
return SignalsAttr(value=val, alias=self._alias)

def computed(self, computed_dict: Mapping | None = None, /, **computed: str) -> BaseAttr:
Expand All@@ -159,7 +190,11 @@ def ignore(self) -> IgnoreAttr:
def attr(self, attr_dict: Mapping | None = None, /, **attrs: str) -> BaseAttr:
"""Set the value of any HTML attributes to expressions, and keep them in sync."""
attrs = {**(attr_dict if attr_dict else {}), **attrs}
return BaseAttr("attr", value=_js_object(attrs), alias=self._alias)
return BaseAttr(
"attr",
value=_js_object({k: JSExpression(v) for k, v in attrs.items()}),
alias=self._alias,
)

def bind(self, signal_name: str) -> BaseAttr:
"""Set up two-way data binding between a signal and an element's value."""
Expand All@@ -168,7 +203,11 @@ def bind(self, signal_name: str) -> BaseAttr:
def class_(self, class_dict: Mapping | None = None, /, **classes: str) -> BaseAttr:
"""Add or removes classes to or from an element based on expressions."""
classes = {**(class_dict if class_dict else {}), **classes}
return BaseAttr("class", value=_js_object(classes), alias=self._alias)
return BaseAttr(
"class",
value=_js_object({k: JSExpression(v) for k, v in classes.items()}),
alias=self._alias,
)

@overload
def on(self, event: Literal["interval"], expression: str) -> OnIntervalAttr: ...
Expand DownExpand Up@@ -259,7 +298,11 @@ def show(self, expression: str) -> BaseAttr:
def style(self, style_dict: Mapping | None = None, /, **styles: str) -> BaseAttr:
"""Sets the value of inline CSS styles on an element based on an expression, and keeps them in sync."""
styles = {**(style_dict if style_dict else {}), **styles}
return BaseAttr("style", value=_js_object(styles), alias=self._alias)
return BaseAttr(
"style",
value=_js_object({k: JSExpression(v) for k, v in styles.items()}),
alias=self._alias,
)

def text(self, expression: str) -> BaseAttr:
"""Bind the text content of an element to an expression."""
Expand DownExpand Up@@ -728,16 +771,4 @@ def _escape(s: str) -> str:
)


def _js_object(obj: dict) -> str:
"""Create a JS object where the values are expressions rather than strings."""
return (
"{"
+ ", ".join(
f"{json.dumps(k)}: {_js_object(v) if isinstance(v, dict) else v}"
for k, v in obj.items()
)
+ "}"
)


attribute_generator = AttributeGenerator()
125 changes: 125 additions & 0 deletions src/datastar_py/attributes/actions.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
from __future__ import annotations

from typing import Literal, TypedDict, Unpack

from datastar_py.attributes import JSExpression, JSRegex, SignalValue, _js_object


class _FetchOptions(TypedDict, total=False):
content_type: Literal["json", "form"]
include_signals: str
exclude_signals: str
selector: str
headers: dict[str, str]
open_when_hidden: bool
retry_interval: int
retry_scalar: float
retry_max_wait_ms: int
retry_max_count: int
request_cancellation: Literal["auto", "disabled"] | str


def _fetch(
method: Literal["get", "post", "put", "patch", "delete"],
url: str,
**options: Unpack[_FetchOptions],
) -> str:
result = f"@{method}('{url}'"
if options:
mapped_options = {}
if "content_type" in options:
mapped_options["contentType"] = options["content_type"]
if "include_signals" in options or "exclude_signals" in options:
filter_signals = {}
if "include_signals" in options:
filter_signals["include"] = JSRegex(options["include_signals"])
if "exclude_signals" in options:
filter_signals["exclude"] = JSRegex(options["exclude_signals"])
mapped_options["filterSignals"] = filter_signals
if "selector" in options:
mapped_options["selector"] = options["selector"]
if "headers" in options:
mapped_options["headers"] = _js_object(options["headers"])
if "open_when_hidden" in options:
mapped_options["openWhenHidden"] = options["open_when_hidden"]
if "retry_interval" in options:
mapped_options["retryInterval"] = options["retry_interval"]
if "retry_scalar" in options:
mapped_options["retryScalar"] = options["retry_scalar"]
if "retry_max_wait_ms" in options:
mapped_options["retryMaxWaitMs"] = options["retry_max_wait_ms"]
if "request_cancellation" in options:
if options["request_cancellation"] in ("auto", "disabled"):
mapped_options["requestCancellation"] = options["request_cancellation"]
else:
mapped_options["requestCancellation"] = JSExpression(
options["request_cancellation"]
)
result += f", {_js_object(mapped_options)}"
result += ")"
return result


def get(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("get", url, **options)


def post(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("post", url, **options)


def put(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("put", url, **options)


def patch(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("patch", url, **options)


def delete(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("delete", url, **options)


def peek(expression: str) -> str:
"""Evaluate an expression containing signals without subscribing to changes in those signals."""
return f"@peek(() => {expression})"


def set_all(value: SignalValue, include: str | None = None, exclude: str | None = None) -> str:
"""Set the value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = f", {_js_object(filter_dict)}" if filter_dict else ""
return f"@setAll({value}{filter_string})"


def toggle_all(include: str | None = None, exclude: str | None = None) -> str:
"""Toggle the boolean value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = _js_object(filter_dict) if filter_dict else ""
return f"@toggleAll({filter_string})"


def clipboard(text: str, is_base_64: bool = False) -> str:
"""PRO: Copy text to the clipboard."""
return f"@clipboard({text}{', true' if is_base_64 else ''})"


def fit(
value: float | str,
old_min: float | str,
old_max: float | str,
new_min: float | str,
new_max: float | str,
should_clamp: bool = False,
should_round: bool = False,
) -> str:
"""PRO: Linearly interpolate a value from one range to another."""
return f"@fit({value}, {old_min}, {old_max}, {new_min}, {new_max}, {'true' if should_clamp else 'false'}{', true' if should_round else ''})"
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms"); } } catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,6 +111,33 @@
]


class JSExpression(str):
MARKER = "_!EXPR!_"
REPLACEMENT = ""

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


class JSRegex(str):
MARKER = "_!REGEX!_"
REPLACEMENT = "/"

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


def _js_object(obj: dict) -> str:
"""Create a JS object where the values can be JS expressions or regex."""
result = json.dumps(obj)
return (
result.replace(f'"{JSExpression.MARKER}', JSExpression.REPLACEMENT)
.replace(f'{JSExpression.MARKER}"', JSExpression.REPLACEMENT)
.replace(f'"{JSRegex.MARKER}', JSRegex.REPLACEMENT)
.replace(f'{JSRegex.MARKER}"', JSRegex.REPLACEMENT)
)


class AttributeGenerator:
def __init__(self, alias: str = "data-") -> None:
"""A helper which can generate all the Datastar attributes.
Expand All@@ -134,7 +161,11 @@ def signals(
rather than literals.
"""
signals = {**(signals_dict if signals_dict else {}), **signals}
val = _js_object(signals) if expressions_ else json.dumps(signals)
val = (
_js_object({k: JSExpression(v) for k, v in signals.items()})
if expressions_
else json.dumps(signals)
)
return SignalsAttr(value=val, alias=self._alias)

def computed(self, computed_dict: Mapping | None = None, /, **computed: str) -> BaseAttr:
Expand All@@ -159,7 +190,11 @@ def ignore(self) -> IgnoreAttr:
def attr(self, attr_dict: Mapping | None = None, /, **attrs: str) -> BaseAttr:
"""Set the value of any HTML attributes to expressions, and keep them in sync."""
attrs = {**(attr_dict if attr_dict else {}), **attrs}
return BaseAttr("attr", value=_js_object(attrs), alias=self._alias)
return BaseAttr(
"attr",
value=_js_object({k: JSExpression(v) for k, v in attrs.items()}),
alias=self._alias,
)

def bind(self, signal_name: str) -> BaseAttr:
"""Set up two-way data binding between a signal and an element's value."""
Expand All@@ -168,7 +203,11 @@ def bind(self, signal_name: str) -> BaseAttr:
def class_(self, class_dict: Mapping | None = None, /, **classes: str) -> BaseAttr:
"""Add or removes classes to or from an element based on expressions."""
classes = {**(class_dict if class_dict else {}), **classes}
return BaseAttr("class", value=_js_object(classes), alias=self._alias)
return BaseAttr(
"class",
value=_js_object({k: JSExpression(v) for k, v in classes.items()}),
alias=self._alias,
)

@overload
def on(self, event: Literal["interval"], expression: str) -> OnIntervalAttr: ...
Expand DownExpand Up@@ -259,7 +298,11 @@ def show(self, expression: str) -> BaseAttr:
def style(self, style_dict: Mapping | None = None, /, **styles: str) -> BaseAttr:
"""Sets the value of inline CSS styles on an element based on an expression, and keeps them in sync."""
styles = {**(style_dict if style_dict else {}), **styles}
return BaseAttr("style", value=_js_object(styles), alias=self._alias)
return BaseAttr(
"style",
value=_js_object({k: JSExpression(v) for k, v in styles.items()}),
alias=self._alias,
)

def text(self, expression: str) -> BaseAttr:
"""Bind the text content of an element to an expression."""
Expand DownExpand Up@@ -728,16 +771,4 @@ def _escape(s: str) -> str:
)


def _js_object(obj: dict) -> str:
"""Create a JS object where the values are expressions rather than strings."""
return (
"{"
+ ", ".join(
f"{json.dumps(k)}: {_js_object(v) if isinstance(v, dict) else v}"
for k, v in obj.items()
)
+ "}"
)


attribute_generator = AttributeGenerator()
125 changes: 125 additions & 0 deletions src/datastar_py/attributes/actions.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
from __future__ import annotations

from typing import Literal, TypedDict, Unpack

from datastar_py.attributes import JSExpression, JSRegex, SignalValue, _js_object


class _FetchOptions(TypedDict, total=False):
content_type: Literal["json", "form"]
include_signals: str
exclude_signals: str
selector: str
headers: dict[str, str]
open_when_hidden: bool
retry_interval: int
retry_scalar: float
retry_max_wait_ms: int
retry_max_count: int
request_cancellation: Literal["auto", "disabled"] | str


def _fetch(
method: Literal["get", "post", "put", "patch", "delete"],
url: str,
**options: Unpack[_FetchOptions],
) -> str:
result = f"@{method}('{url}'"
if options:
mapped_options = {}
if "content_type" in options:
mapped_options["contentType"] = options["content_type"]
if "include_signals" in options or "exclude_signals" in options:
filter_signals = {}
if "include_signals" in options:
filter_signals["include"] = JSRegex(options["include_signals"])
if "exclude_signals" in options:
filter_signals["exclude"] = JSRegex(options["exclude_signals"])
mapped_options["filterSignals"] = filter_signals
if "selector" in options:
mapped_options["selector"] = options["selector"]
if "headers" in options:
mapped_options["headers"] = _js_object(options["headers"])
if "open_when_hidden" in options:
mapped_options["openWhenHidden"] = options["open_when_hidden"]
if "retry_interval" in options:
mapped_options["retryInterval"] = options["retry_interval"]
if "retry_scalar" in options:
mapped_options["retryScalar"] = options["retry_scalar"]
if "retry_max_wait_ms" in options:
mapped_options["retryMaxWaitMs"] = options["retry_max_wait_ms"]
if "request_cancellation" in options:
if options["request_cancellation"] in ("auto", "disabled"):
mapped_options["requestCancellation"] = options["request_cancellation"]
else:
mapped_options["requestCancellation"] = JSExpression(
options["request_cancellation"]
)
result += f", {_js_object(mapped_options)}"
result += ")"
return result


def get(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("get", url, **options)


def post(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("post", url, **options)


def put(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("put", url, **options)


def patch(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("patch", url, **options)


def delete(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("delete", url, **options)


def peek(expression: str) -> str:
"""Evaluate an expression containing signals without subscribing to changes in those signals."""
return f"@peek(() => {expression})"


def set_all(value: SignalValue, include: str | None = None, exclude: str | None = None) -> str:
"""Set the value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = f", {_js_object(filter_dict)}" if filter_dict else ""
return f"@setAll({value}{filter_string})"


def toggle_all(include: str | None = None, exclude: str | None = None) -> str:
"""Toggle the boolean value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = _js_object(filter_dict) if filter_dict else ""
return f"@toggleAll({filter_string})"


def clipboard(text: str, is_base_64: bool = False) -> str:
"""PRO: Copy text to the clipboard."""
return f"@clipboard({text}{', true' if is_base_64 else ''})"


def fit(
value: float | str,
old_min: float | str,
old_max: float | str,
new_min: float | str,
new_max: float | str,
should_clamp: bool = False,
should_round: bool = False,
) -> str:
"""PRO: Linearly interpolate a value from one range to another."""
return f"@fit({value}, {old_min}, {old_max}, {new_min}, {new_max}, {'true' if should_clamp else 'false'}{', true' if should_round else ''})"
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links"); } } catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); } })(); (function(){ try { var __m = "youtube.com"; var __re = new RegExp('^' + "youtube\\.com" + '
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,6 +111,33 @@
]


class JSExpression(str):
MARKER = "_!EXPR!_"
REPLACEMENT = ""

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


class JSRegex(str):
MARKER = "_!REGEX!_"
REPLACEMENT = "/"

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


def _js_object(obj: dict) -> str:
"""Create a JS object where the values can be JS expressions or regex."""
result = json.dumps(obj)
return (
result.replace(f'"{JSExpression.MARKER}', JSExpression.REPLACEMENT)
.replace(f'{JSExpression.MARKER}"', JSExpression.REPLACEMENT)
.replace(f'"{JSRegex.MARKER}', JSRegex.REPLACEMENT)
.replace(f'{JSRegex.MARKER}"', JSRegex.REPLACEMENT)
)


class AttributeGenerator:
def __init__(self, alias: str = "data-") -> None:
"""A helper which can generate all the Datastar attributes.
Expand All@@ -134,7 +161,11 @@ def signals(
rather than literals.
"""
signals = {**(signals_dict if signals_dict else {}), **signals}
val = _js_object(signals) if expressions_ else json.dumps(signals)
val = (
_js_object({k: JSExpression(v) for k, v in signals.items()})
if expressions_
else json.dumps(signals)
)
return SignalsAttr(value=val, alias=self._alias)

def computed(self, computed_dict: Mapping | None = None, /, **computed: str) -> BaseAttr:
Expand All@@ -159,7 +190,11 @@ def ignore(self) -> IgnoreAttr:
def attr(self, attr_dict: Mapping | None = None, /, **attrs: str) -> BaseAttr:
"""Set the value of any HTML attributes to expressions, and keep them in sync."""
attrs = {**(attr_dict if attr_dict else {}), **attrs}
return BaseAttr("attr", value=_js_object(attrs), alias=self._alias)
return BaseAttr(
"attr",
value=_js_object({k: JSExpression(v) for k, v in attrs.items()}),
alias=self._alias,
)

def bind(self, signal_name: str) -> BaseAttr:
"""Set up two-way data binding between a signal and an element's value."""
Expand All@@ -168,7 +203,11 @@ def bind(self, signal_name: str) -> BaseAttr:
def class_(self, class_dict: Mapping | None = None, /, **classes: str) -> BaseAttr:
"""Add or removes classes to or from an element based on expressions."""
classes = {**(class_dict if class_dict else {}), **classes}
return BaseAttr("class", value=_js_object(classes), alias=self._alias)
return BaseAttr(
"class",
value=_js_object({k: JSExpression(v) for k, v in classes.items()}),
alias=self._alias,
)

@overload
def on(self, event: Literal["interval"], expression: str) -> OnIntervalAttr: ...
Expand DownExpand Up@@ -259,7 +298,11 @@ def show(self, expression: str) -> BaseAttr:
def style(self, style_dict: Mapping | None = None, /, **styles: str) -> BaseAttr:
"""Sets the value of inline CSS styles on an element based on an expression, and keeps them in sync."""
styles = {**(style_dict if style_dict else {}), **styles}
return BaseAttr("style", value=_js_object(styles), alias=self._alias)
return BaseAttr(
"style",
value=_js_object({k: JSExpression(v) for k, v in styles.items()}),
alias=self._alias,
)

def text(self, expression: str) -> BaseAttr:
"""Bind the text content of an element to an expression."""
Expand DownExpand Up@@ -728,16 +771,4 @@ def _escape(s: str) -> str:
)


def _js_object(obj: dict) -> str:
"""Create a JS object where the values are expressions rather than strings."""
return (
"{"
+ ", ".join(
f"{json.dumps(k)}: {_js_object(v) if isinstance(v, dict) else v}"
for k, v in obj.items()
)
+ "}"
)


attribute_generator = AttributeGenerator()
125 changes: 125 additions & 0 deletions src/datastar_py/attributes/actions.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
from __future__ import annotations

from typing import Literal, TypedDict, Unpack

from datastar_py.attributes import JSExpression, JSRegex, SignalValue, _js_object


class _FetchOptions(TypedDict, total=False):
content_type: Literal["json", "form"]
include_signals: str
exclude_signals: str
selector: str
headers: dict[str, str]
open_when_hidden: bool
retry_interval: int
retry_scalar: float
retry_max_wait_ms: int
retry_max_count: int
request_cancellation: Literal["auto", "disabled"] | str


def _fetch(
method: Literal["get", "post", "put", "patch", "delete"],
url: str,
**options: Unpack[_FetchOptions],
) -> str:
result = f"@{method}('{url}'"
if options:
mapped_options = {}
if "content_type" in options:
mapped_options["contentType"] = options["content_type"]
if "include_signals" in options or "exclude_signals" in options:
filter_signals = {}
if "include_signals" in options:
filter_signals["include"] = JSRegex(options["include_signals"])
if "exclude_signals" in options:
filter_signals["exclude"] = JSRegex(options["exclude_signals"])
mapped_options["filterSignals"] = filter_signals
if "selector" in options:
mapped_options["selector"] = options["selector"]
if "headers" in options:
mapped_options["headers"] = _js_object(options["headers"])
if "open_when_hidden" in options:
mapped_options["openWhenHidden"] = options["open_when_hidden"]
if "retry_interval" in options:
mapped_options["retryInterval"] = options["retry_interval"]
if "retry_scalar" in options:
mapped_options["retryScalar"] = options["retry_scalar"]
if "retry_max_wait_ms" in options:
mapped_options["retryMaxWaitMs"] = options["retry_max_wait_ms"]
if "request_cancellation" in options:
if options["request_cancellation"] in ("auto", "disabled"):
mapped_options["requestCancellation"] = options["request_cancellation"]
else:
mapped_options["requestCancellation"] = JSExpression(
options["request_cancellation"]
)
result += f", {_js_object(mapped_options)}"
result += ")"
return result


def get(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("get", url, **options)


def post(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("post", url, **options)


def put(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("put", url, **options)


def patch(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("patch", url, **options)


def delete(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("delete", url, **options)


def peek(expression: str) -> str:
"""Evaluate an expression containing signals without subscribing to changes in those signals."""
return f"@peek(() => {expression})"


def set_all(value: SignalValue, include: str | None = None, exclude: str | None = None) -> str:
"""Set the value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = f", {_js_object(filter_dict)}" if filter_dict else ""
return f"@setAll({value}{filter_string})"


def toggle_all(include: str | None = None, exclude: str | None = None) -> str:
"""Toggle the boolean value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = _js_object(filter_dict) if filter_dict else ""
return f"@toggleAll({filter_string})"


def clipboard(text: str, is_base_64: bool = False) -> str:
"""PRO: Copy text to the clipboard."""
return f"@clipboard({text}{', true' if is_base_64 else ''})"


def fit(
value: float | str,
old_min: float | str,
old_max: float | str,
new_min: float | str,
new_max: float | str,
should_clamp: bool = False,
should_round: bool = False,
) -> str:
"""PRO: Linearly interpolate a value from one range to another."""
return f"@fit({value}, {old_min}, {old_max}, {new_min}, {new_max}, {'true' if should_clamp else 'false'}{', true' if should_round else ''})"
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default"); } } catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,6 +111,33 @@
]


class JSExpression(str):
MARKER = "_!EXPR!_"
REPLACEMENT = ""

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


class JSRegex(str):
MARKER = "_!REGEX!_"
REPLACEMENT = "/"

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


def _js_object(obj: dict) -> str:
"""Create a JS object where the values can be JS expressions or regex."""
result = json.dumps(obj)
return (
result.replace(f'"{JSExpression.MARKER}', JSExpression.REPLACEMENT)
.replace(f'{JSExpression.MARKER}"', JSExpression.REPLACEMENT)
.replace(f'"{JSRegex.MARKER}', JSRegex.REPLACEMENT)
.replace(f'{JSRegex.MARKER}"', JSRegex.REPLACEMENT)
)


class AttributeGenerator:
def __init__(self, alias: str = "data-") -> None:
"""A helper which can generate all the Datastar attributes.
Expand All@@ -134,7 +161,11 @@ def signals(
rather than literals.
"""
signals = {**(signals_dict if signals_dict else {}), **signals}
val = _js_object(signals) if expressions_ else json.dumps(signals)
val = (
_js_object({k: JSExpression(v) for k, v in signals.items()})
if expressions_
else json.dumps(signals)
)
return SignalsAttr(value=val, alias=self._alias)

def computed(self, computed_dict: Mapping | None = None, /, **computed: str) -> BaseAttr:
Expand All@@ -159,7 +190,11 @@ def ignore(self) -> IgnoreAttr:
def attr(self, attr_dict: Mapping | None = None, /, **attrs: str) -> BaseAttr:
"""Set the value of any HTML attributes to expressions, and keep them in sync."""
attrs = {**(attr_dict if attr_dict else {}), **attrs}
return BaseAttr("attr", value=_js_object(attrs), alias=self._alias)
return BaseAttr(
"attr",
value=_js_object({k: JSExpression(v) for k, v in attrs.items()}),
alias=self._alias,
)

def bind(self, signal_name: str) -> BaseAttr:
"""Set up two-way data binding between a signal and an element's value."""
Expand All@@ -168,7 +203,11 @@ def bind(self, signal_name: str) -> BaseAttr:
def class_(self, class_dict: Mapping | None = None, /, **classes: str) -> BaseAttr:
"""Add or removes classes to or from an element based on expressions."""
classes = {**(class_dict if class_dict else {}), **classes}
return BaseAttr("class", value=_js_object(classes), alias=self._alias)
return BaseAttr(
"class",
value=_js_object({k: JSExpression(v) for k, v in classes.items()}),
alias=self._alias,
)

@overload
def on(self, event: Literal["interval"], expression: str) -> OnIntervalAttr: ...
Expand DownExpand Up@@ -259,7 +298,11 @@ def show(self, expression: str) -> BaseAttr:
def style(self, style_dict: Mapping | None = None, /, **styles: str) -> BaseAttr:
"""Sets the value of inline CSS styles on an element based on an expression, and keeps them in sync."""
styles = {**(style_dict if style_dict else {}), **styles}
return BaseAttr("style", value=_js_object(styles), alias=self._alias)
return BaseAttr(
"style",
value=_js_object({k: JSExpression(v) for k, v in styles.items()}),
alias=self._alias,
)

def text(self, expression: str) -> BaseAttr:
"""Bind the text content of an element to an expression."""
Expand DownExpand Up@@ -728,16 +771,4 @@ def _escape(s: str) -> str:
)


def _js_object(obj: dict) -> str:
"""Create a JS object where the values are expressions rather than strings."""
return (
"{"
+ ", ".join(
f"{json.dumps(k)}: {_js_object(v) if isinstance(v, dict) else v}"
for k, v in obj.items()
)
+ "}"
)


attribute_generator = AttributeGenerator()
125 changes: 125 additions & 0 deletions src/datastar_py/attributes/actions.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
from __future__ import annotations

from typing import Literal, TypedDict, Unpack

from datastar_py.attributes import JSExpression, JSRegex, SignalValue, _js_object


class _FetchOptions(TypedDict, total=False):
content_type: Literal["json", "form"]
include_signals: str
exclude_signals: str
selector: str
headers: dict[str, str]
open_when_hidden: bool
retry_interval: int
retry_scalar: float
retry_max_wait_ms: int
retry_max_count: int
request_cancellation: Literal["auto", "disabled"] | str


def _fetch(
method: Literal["get", "post", "put", "patch", "delete"],
url: str,
**options: Unpack[_FetchOptions],
) -> str:
result = f"@{method}('{url}'"
if options:
mapped_options = {}
if "content_type" in options:
mapped_options["contentType"] = options["content_type"]
if "include_signals" in options or "exclude_signals" in options:
filter_signals = {}
if "include_signals" in options:
filter_signals["include"] = JSRegex(options["include_signals"])
if "exclude_signals" in options:
filter_signals["exclude"] = JSRegex(options["exclude_signals"])
mapped_options["filterSignals"] = filter_signals
if "selector" in options:
mapped_options["selector"] = options["selector"]
if "headers" in options:
mapped_options["headers"] = _js_object(options["headers"])
if "open_when_hidden" in options:
mapped_options["openWhenHidden"] = options["open_when_hidden"]
if "retry_interval" in options:
mapped_options["retryInterval"] = options["retry_interval"]
if "retry_scalar" in options:
mapped_options["retryScalar"] = options["retry_scalar"]
if "retry_max_wait_ms" in options:
mapped_options["retryMaxWaitMs"] = options["retry_max_wait_ms"]
if "request_cancellation" in options:
if options["request_cancellation"] in ("auto", "disabled"):
mapped_options["requestCancellation"] = options["request_cancellation"]
else:
mapped_options["requestCancellation"] = JSExpression(
options["request_cancellation"]
)
result += f", {_js_object(mapped_options)}"
result += ")"
return result


def get(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("get", url, **options)


def post(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("post", url, **options)


def put(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("put", url, **options)


def patch(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("patch", url, **options)


def delete(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("delete", url, **options)


def peek(expression: str) -> str:
"""Evaluate an expression containing signals without subscribing to changes in those signals."""
return f"@peek(() => {expression})"


def set_all(value: SignalValue, include: str | None = None, exclude: str | None = None) -> str:
"""Set the value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = f", {_js_object(filter_dict)}" if filter_dict else ""
return f"@setAll({value}{filter_string})"


def toggle_all(include: str | None = None, exclude: str | None = None) -> str:
"""Toggle the boolean value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = _js_object(filter_dict) if filter_dict else ""
return f"@toggleAll({filter_string})"


def clipboard(text: str, is_base_64: bool = False) -> str:
"""PRO: Copy text to the clipboard."""
return f"@clipboard({text}{', true' if is_base_64 else ''})"


def fit(
value: float | str,
old_min: float | str,
old_max: float | str,
new_min: float | str,
new_max: float | str,
should_clamp: bool = False,
should_round: bool = False,
) -> str:
"""PRO: Linearly interpolate a value from one range to another."""
return f"@fit({value}, {old_min}, {old_max}, {new_min}, {new_max}, {'true' if should_clamp else 'false'}{', true' if should_round else ''})"
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers"); } } catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,6 +111,33 @@
]


class JSExpression(str):
MARKER = "_!EXPR!_"
REPLACEMENT = ""

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


class JSRegex(str):
MARKER = "_!REGEX!_"
REPLACEMENT = "/"

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


def _js_object(obj: dict) -> str:
"""Create a JS object where the values can be JS expressions or regex."""
result = json.dumps(obj)
return (
result.replace(f'"{JSExpression.MARKER}', JSExpression.REPLACEMENT)
.replace(f'{JSExpression.MARKER}"', JSExpression.REPLACEMENT)
.replace(f'"{JSRegex.MARKER}', JSRegex.REPLACEMENT)
.replace(f'{JSRegex.MARKER}"', JSRegex.REPLACEMENT)
)


class AttributeGenerator:
def __init__(self, alias: str = "data-") -> None:
"""A helper which can generate all the Datastar attributes.
Expand All@@ -134,7 +161,11 @@ def signals(
rather than literals.
"""
signals = {**(signals_dict if signals_dict else {}), **signals}
val = _js_object(signals) if expressions_ else json.dumps(signals)
val = (
_js_object({k: JSExpression(v) for k, v in signals.items()})
if expressions_
else json.dumps(signals)
)
return SignalsAttr(value=val, alias=self._alias)

def computed(self, computed_dict: Mapping | None = None, /, **computed: str) -> BaseAttr:
Expand All@@ -159,7 +190,11 @@ def ignore(self) -> IgnoreAttr:
def attr(self, attr_dict: Mapping | None = None, /, **attrs: str) -> BaseAttr:
"""Set the value of any HTML attributes to expressions, and keep them in sync."""
attrs = {**(attr_dict if attr_dict else {}), **attrs}
return BaseAttr("attr", value=_js_object(attrs), alias=self._alias)
return BaseAttr(
"attr",
value=_js_object({k: JSExpression(v) for k, v in attrs.items()}),
alias=self._alias,
)

def bind(self, signal_name: str) -> BaseAttr:
"""Set up two-way data binding between a signal and an element's value."""
Expand All@@ -168,7 +203,11 @@ def bind(self, signal_name: str) -> BaseAttr:
def class_(self, class_dict: Mapping | None = None, /, **classes: str) -> BaseAttr:
"""Add or removes classes to or from an element based on expressions."""
classes = {**(class_dict if class_dict else {}), **classes}
return BaseAttr("class", value=_js_object(classes), alias=self._alias)
return BaseAttr(
"class",
value=_js_object({k: JSExpression(v) for k, v in classes.items()}),
alias=self._alias,
)

@overload
def on(self, event: Literal["interval"], expression: str) -> OnIntervalAttr: ...
Expand DownExpand Up@@ -259,7 +298,11 @@ def show(self, expression: str) -> BaseAttr:
def style(self, style_dict: Mapping | None = None, /, **styles: str) -> BaseAttr:
"""Sets the value of inline CSS styles on an element based on an expression, and keeps them in sync."""
styles = {**(style_dict if style_dict else {}), **styles}
return BaseAttr("style", value=_js_object(styles), alias=self._alias)
return BaseAttr(
"style",
value=_js_object({k: JSExpression(v) for k, v in styles.items()}),
alias=self._alias,
)

def text(self, expression: str) -> BaseAttr:
"""Bind the text content of an element to an expression."""
Expand DownExpand Up@@ -728,16 +771,4 @@ def _escape(s: str) -> str:
)


def _js_object(obj: dict) -> str:
"""Create a JS object where the values are expressions rather than strings."""
return (
"{"
+ ", ".join(
f"{json.dumps(k)}: {_js_object(v) if isinstance(v, dict) else v}"
for k, v in obj.items()
)
+ "}"
)


attribute_generator = AttributeGenerator()
125 changes: 125 additions & 0 deletions src/datastar_py/attributes/actions.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
from __future__ import annotations

from typing import Literal, TypedDict, Unpack

from datastar_py.attributes import JSExpression, JSRegex, SignalValue, _js_object


class _FetchOptions(TypedDict, total=False):
content_type: Literal["json", "form"]
include_signals: str
exclude_signals: str
selector: str
headers: dict[str, str]
open_when_hidden: bool
retry_interval: int
retry_scalar: float
retry_max_wait_ms: int
retry_max_count: int
request_cancellation: Literal["auto", "disabled"] | str


def _fetch(
method: Literal["get", "post", "put", "patch", "delete"],
url: str,
**options: Unpack[_FetchOptions],
) -> str:
result = f"@{method}('{url}'"
if options:
mapped_options = {}
if "content_type" in options:
mapped_options["contentType"] = options["content_type"]
if "include_signals" in options or "exclude_signals" in options:
filter_signals = {}
if "include_signals" in options:
filter_signals["include"] = JSRegex(options["include_signals"])
if "exclude_signals" in options:
filter_signals["exclude"] = JSRegex(options["exclude_signals"])
mapped_options["filterSignals"] = filter_signals
if "selector" in options:
mapped_options["selector"] = options["selector"]
if "headers" in options:
mapped_options["headers"] = _js_object(options["headers"])
if "open_when_hidden" in options:
mapped_options["openWhenHidden"] = options["open_when_hidden"]
if "retry_interval" in options:
mapped_options["retryInterval"] = options["retry_interval"]
if "retry_scalar" in options:
mapped_options["retryScalar"] = options["retry_scalar"]
if "retry_max_wait_ms" in options:
mapped_options["retryMaxWaitMs"] = options["retry_max_wait_ms"]
if "request_cancellation" in options:
if options["request_cancellation"] in ("auto", "disabled"):
mapped_options["requestCancellation"] = options["request_cancellation"]
else:
mapped_options["requestCancellation"] = JSExpression(
options["request_cancellation"]
)
result += f", {_js_object(mapped_options)}"
result += ")"
return result


def get(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("get", url, **options)


def post(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("post", url, **options)


def put(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("put", url, **options)


def patch(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("patch", url, **options)


def delete(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("delete", url, **options)


def peek(expression: str) -> str:
"""Evaluate an expression containing signals without subscribing to changes in those signals."""
return f"@peek(() => {expression})"


def set_all(value: SignalValue, include: str | None = None, exclude: str | None = None) -> str:
"""Set the value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = f", {_js_object(filter_dict)}" if filter_dict else ""
return f"@setAll({value}{filter_string})"


def toggle_all(include: str | None = None, exclude: str | None = None) -> str:
"""Toggle the boolean value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = _js_object(filter_dict) if filter_dict else ""
return f"@toggleAll({filter_string})"


def clipboard(text: str, is_base_64: bool = False) -> str:
"""PRO: Copy text to the clipboard."""
return f"@clipboard({text}{', true' if is_base_64 else ''})"


def fit(
value: float | str,
old_min: float | str,
old_max: float | str,
new_min: float | str,
new_max: float | str,
should_clamp: bool = False,
should_round: bool = False,
) -> str:
"""PRO: Linearly interpolate a value from one range to another."""
return f"@fit({value}, {old_min}, {old_max}, {new_min}, {new_max}, {'true' if should_clamp else 'false'}{', true' if should_round else ''})"
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode"); } } catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); } })(); })();
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,6 +111,33 @@
]


class JSExpression(str):
MARKER = "_!EXPR!_"
REPLACEMENT = ""

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


class JSRegex(str):
MARKER = "_!REGEX!_"
REPLACEMENT = "/"

def __new__(cls, value: str) -> Self:
return str.__new__(cls, f"{cls.MARKER}{value}{cls.MARKER}")


def _js_object(obj: dict) -> str:
"""Create a JS object where the values can be JS expressions or regex."""
result = json.dumps(obj)
return (
result.replace(f'"{JSExpression.MARKER}', JSExpression.REPLACEMENT)
.replace(f'{JSExpression.MARKER}"', JSExpression.REPLACEMENT)
.replace(f'"{JSRegex.MARKER}', JSRegex.REPLACEMENT)
.replace(f'{JSRegex.MARKER}"', JSRegex.REPLACEMENT)
)


class AttributeGenerator:
def __init__(self, alias: str = "data-") -> None:
"""A helper which can generate all the Datastar attributes.
Expand All@@ -134,7 +161,11 @@ def signals(
rather than literals.
"""
signals = {**(signals_dict if signals_dict else {}), **signals}
val = _js_object(signals) if expressions_ else json.dumps(signals)
val = (
_js_object({k: JSExpression(v) for k, v in signals.items()})
if expressions_
else json.dumps(signals)
)
return SignalsAttr(value=val, alias=self._alias)

def computed(self, computed_dict: Mapping | None = None, /, **computed: str) -> BaseAttr:
Expand All@@ -159,7 +190,11 @@ def ignore(self) -> IgnoreAttr:
def attr(self, attr_dict: Mapping | None = None, /, **attrs: str) -> BaseAttr:
"""Set the value of any HTML attributes to expressions, and keep them in sync."""
attrs = {**(attr_dict if attr_dict else {}), **attrs}
return BaseAttr("attr", value=_js_object(attrs), alias=self._alias)
return BaseAttr(
"attr",
value=_js_object({k: JSExpression(v) for k, v in attrs.items()}),
alias=self._alias,
)

def bind(self, signal_name: str) -> BaseAttr:
"""Set up two-way data binding between a signal and an element's value."""
Expand All@@ -168,7 +203,11 @@ def bind(self, signal_name: str) -> BaseAttr:
def class_(self, class_dict: Mapping | None = None, /, **classes: str) -> BaseAttr:
"""Add or removes classes to or from an element based on expressions."""
classes = {**(class_dict if class_dict else {}), **classes}
return BaseAttr("class", value=_js_object(classes), alias=self._alias)
return BaseAttr(
"class",
value=_js_object({k: JSExpression(v) for k, v in classes.items()}),
alias=self._alias,
)

@overload
def on(self, event: Literal["interval"], expression: str) -> OnIntervalAttr: ...
Expand DownExpand Up@@ -259,7 +298,11 @@ def show(self, expression: str) -> BaseAttr:
def style(self, style_dict: Mapping | None = None, /, **styles: str) -> BaseAttr:
"""Sets the value of inline CSS styles on an element based on an expression, and keeps them in sync."""
styles = {**(style_dict if style_dict else {}), **styles}
return BaseAttr("style", value=_js_object(styles), alias=self._alias)
return BaseAttr(
"style",
value=_js_object({k: JSExpression(v) for k, v in styles.items()}),
alias=self._alias,
)

def text(self, expression: str) -> BaseAttr:
"""Bind the text content of an element to an expression."""
Expand DownExpand Up@@ -728,16 +771,4 @@ def _escape(s: str) -> str:
)


def _js_object(obj: dict) -> str:
"""Create a JS object where the values are expressions rather than strings."""
return (
"{"
+ ", ".join(
f"{json.dumps(k)}: {_js_object(v) if isinstance(v, dict) else v}"
for k, v in obj.items()
)
+ "}"
)


attribute_generator = AttributeGenerator()
125 changes: 125 additions & 0 deletions src/datastar_py/attributes/actions.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
from __future__ import annotations

from typing import Literal, TypedDict, Unpack

from datastar_py.attributes import JSExpression, JSRegex, SignalValue, _js_object


class _FetchOptions(TypedDict, total=False):
content_type: Literal["json", "form"]
include_signals: str
exclude_signals: str
selector: str
headers: dict[str, str]
open_when_hidden: bool
retry_interval: int
retry_scalar: float
retry_max_wait_ms: int
retry_max_count: int
request_cancellation: Literal["auto", "disabled"] | str


def _fetch(
method: Literal["get", "post", "put", "patch", "delete"],
url: str,
**options: Unpack[_FetchOptions],
) -> str:
result = f"@{method}('{url}'"
if options:
mapped_options = {}
if "content_type" in options:
mapped_options["contentType"] = options["content_type"]
if "include_signals" in options or "exclude_signals" in options:
filter_signals = {}
if "include_signals" in options:
filter_signals["include"] = JSRegex(options["include_signals"])
if "exclude_signals" in options:
filter_signals["exclude"] = JSRegex(options["exclude_signals"])
mapped_options["filterSignals"] = filter_signals
if "selector" in options:
mapped_options["selector"] = options["selector"]
if "headers" in options:
mapped_options["headers"] = _js_object(options["headers"])
if "open_when_hidden" in options:
mapped_options["openWhenHidden"] = options["open_when_hidden"]
if "retry_interval" in options:
mapped_options["retryInterval"] = options["retry_interval"]
if "retry_scalar" in options:
mapped_options["retryScalar"] = options["retry_scalar"]
if "retry_max_wait_ms" in options:
mapped_options["retryMaxWaitMs"] = options["retry_max_wait_ms"]
if "request_cancellation" in options:
if options["request_cancellation"] in ("auto", "disabled"):
mapped_options["requestCancellation"] = options["request_cancellation"]
else:
mapped_options["requestCancellation"] = JSExpression(
options["request_cancellation"]
)
result += f", {_js_object(mapped_options)}"
result += ")"
return result


def get(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("get", url, **options)


def post(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("post", url, **options)


def put(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("put", url, **options)


def patch(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("patch", url, **options)


def delete(url: str, **options: Unpack[_FetchOptions]) -> str:
return _fetch("delete", url, **options)


def peek(expression: str) -> str:
"""Evaluate an expression containing signals without subscribing to changes in those signals."""
return f"@peek(() => {expression})"


def set_all(value: SignalValue, include: str | None = None, exclude: str | None = None) -> str:
"""Set the value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = f", {_js_object(filter_dict)}" if filter_dict else ""
return f"@setAll({value}{filter_string})"


def toggle_all(include: str | None = None, exclude: str | None = None) -> str:
"""Toggle the boolean value of all matching signals."""
filter_dict = {}
if include:
filter_dict["include"] = JSRegex(include)
if exclude:
filter_dict["exclude"] = JSRegex(exclude)
filter_string = _js_object(filter_dict) if filter_dict else ""
return f"@toggleAll({filter_string})"


def clipboard(text: str, is_base_64: bool = False) -> str:
"""PRO: Copy text to the clipboard."""
return f"@clipboard({text}{', true' if is_base_64 else ''})"


def fit(
value: float | str,
old_min: float | str,
old_max: float | str,
new_min: float | str,
new_max: float | str,
should_clamp: bool = False,
should_round: bool = False,
) -> str:
"""PRO: Linearly interpolate a value from one range to another."""
return f"@fit({value}, {old_min}, {old_max}, {new_min}, {new_max}, {'true' if should_clamp else 'false'}{', true' if should_round else ''})"
Loading