1 files changed, 148 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/_utils.py b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/_utils.py
new file mode 100644
index 0000000..894fd25
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/_utils.py
@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import TYPE_CHECKING, Any, Callable, cast
+from urllib.parse import quote
+
+from litestar.exceptions import ImproperlyConfiguredException
+from litestar.serialization import encode_json
+
+__all__ = (
+    "HTMXHeaders",
+    "get_headers",
+    "get_location_headers",
+    "get_push_url_header",
+    "get_redirect_header",
+    "get_refresh_header",
+    "get_replace_url_header",
+    "get_reswap_header",
+    "get_retarget_header",
+    "get_trigger_event_headers",
+)
+
+
+if TYPE_CHECKING:
+    from litestar.contrib.htmx.types import (
+        EventAfterType,
+        HtmxHeaderType,
+        LocationType,
+        PushUrlType,
+        ReSwapMethod,
+        TriggerEventType,
+    )
+
+HTMX_STOP_POLLING = 286
+
+
+class HTMXHeaders(str, Enum):
+    """Enum for HTMX Headers"""
+
+    REDIRECT = "HX-Redirect"
+    REFRESH = "HX-Refresh"
+    PUSH_URL = "HX-Push-Url"
+    REPLACE_URL = "HX-Replace-Url"
+    RE_SWAP = "HX-Reswap"
+    RE_TARGET = "HX-Retarget"
+    LOCATION = "HX-Location"
+
+    TRIGGER_EVENT = "HX-Trigger"
+    TRIGGER_AFTER_SETTLE = "HX-Trigger-After-Settle"
+    TRIGGER_AFTER_SWAP = "HX-Trigger-After-Swap"
+
+    REQUEST = "HX-Request"
+    BOOSTED = "HX-Boosted"
+    CURRENT_URL = "HX-Current-URL"
+    HISTORY_RESTORE_REQUEST = "HX-History-Restore-Request"
+    PROMPT = "HX-Prompt"
+    TARGET = "HX-Target"
+    TRIGGER_ID = "HX-Trigger"  # noqa: PIE796
+    TRIGGER_NAME = "HX-Trigger-Name"
+    TRIGGERING_EVENT = "Triggering-Event"
+
+
+def get_trigger_event_headers(trigger_event: TriggerEventType) -> dict[str, Any]:
+    """Return headers for trigger event response."""
+    after_params: dict[EventAfterType, str] = {
+        "receive": HTMXHeaders.TRIGGER_EVENT.value,
+        "settle": HTMXHeaders.TRIGGER_AFTER_SETTLE.value,
+        "swap": HTMXHeaders.TRIGGER_AFTER_SWAP.value,
+    }
+
+    if trigger_header := after_params.get(trigger_event["after"]):
+        return {trigger_header: encode_json({trigger_event["name"]: trigger_event["params"] or {}}).decode()}
+
+    raise ImproperlyConfiguredException(
+        "invalid value for 'after' param- allowed values are 'receive', 'settle' or 'swap'."
+    )
+
+
+def get_redirect_header(url: str) -> dict[str, Any]:
+    """Return headers for redirect response."""
+    return {HTMXHeaders.REDIRECT.value: quote(url, safe="/#%[]=:;$&()+,!?*@'~"), "Location": ""}
+
+
+def get_push_url_header(url: PushUrlType) -> dict[str, Any]:
+    """Return headers for push url to browser history response."""
+    if isinstance(url, str):
+        url = url if url != "False" else "false"
+    elif isinstance(url, bool):
+        url = "false"
+
+    return {HTMXHeaders.PUSH_URL.value: url}
+
+
+def get_replace_url_header(url: PushUrlType) -> dict[str, Any]:
+    """Return headers for replace url in browser tab response."""
+    url = (url if url != "False" else "false") if isinstance(url, str) else "false"
+    return {HTMXHeaders.REPLACE_URL: url}
+
+
+def get_refresh_header(refresh: bool) -> dict[str, Any]:
+    """Return headers for client refresh response."""
+    return {HTMXHeaders.REFRESH.value: "true" if refresh else ""}
+
+
+def get_reswap_header(method: ReSwapMethod) -> dict[str, Any]:
+    """Return headers for change swap method response."""
+    return {HTMXHeaders.RE_SWAP.value: method}
+
+
+def get_retarget_header(target: str) -> dict[str, Any]:
+    """Return headers for change target element response."""
+    return {HTMXHeaders.RE_TARGET.value: target}
+
+
+def get_location_headers(location: LocationType) -> dict[str, Any]:
+    """Return headers for redirect without page-reload response."""
+    if spec := {key: value for key, value in location.items() if value}:
+        return {HTMXHeaders.LOCATION.value: encode_json(spec).decode()}
+    raise ValueError("redirect_to is required parameter.")
+
+
+def get_headers(hx_headers: HtmxHeaderType) -> dict[str, Any]:
+    """Return headers for HTMX responses."""
+    if not hx_headers:
+        raise ValueError("Value for hx_headers cannot be None.")
+    htmx_headers_dict: dict[str, Callable] = {
+        "redirect": get_redirect_header,
+        "refresh": get_refresh_header,
+        "push_url": get_push_url_header,
+        "replace_url": get_replace_url_header,
+        "re_swap": get_reswap_header,
+        "re_target": get_retarget_header,
+        "trigger_event": get_trigger_event_headers,
+        "location": get_location_headers,
+    }
+
+    header: dict[str, Any] = {}
+    response: dict[str, Any]
+    key: str
+    value: Any
+
+    for key, value in hx_headers.items():
+        if key in ["redirect", "refresh", "location", "replace_url"]:
+            return cast("dict[str, Any]", htmx_headers_dict[key](value))
+        if value is not None:
+            response = htmx_headers_dict[key](value)
+            header.update(response)
+    return header