summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/contrib/htmx
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:10:44 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:10:44 -0400
commit6d7ba58f880be618ade07f8ea080fe8c4bf8a896 (patch)
treeb1c931051ffcebd2bd9d61d98d6233ffa289bbce /venv/lib/python3.11/site-packages/litestar/contrib/htmx
parent4f884c9abc32990b4061a1bb6997b4b37e58ea0b (diff)
venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/contrib/htmx')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/__init__.py0
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/__init__.cpython-311.pycbin0 -> 205 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/_utils.cpython-311.pycbin0 -> 6781 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/request.cpython-311.pycbin0 -> 7932 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/response.cpython-311.pycbin0 -> 11070 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/types.cpython-311.pycbin0 -> 2382 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/_utils.py148
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/request.py113
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/response.py200
-rw-r--r--venv/lib/python3.11/site-packages/litestar/contrib/htmx/types.py54
10 files changed, 515 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__init__.py b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__init__.py
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/__init__.cpython-311.pyc
new file mode 100644
index 0000000..31d4982
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/__init__.cpython-311.pyc
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/_utils.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/_utils.cpython-311.pyc
new file mode 100644
index 0000000..d860774
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/_utils.cpython-311.pyc
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/request.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/request.cpython-311.pyc
new file mode 100644
index 0000000..65b99d9
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/request.cpython-311.pyc
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/response.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/response.cpython-311.pyc
new file mode 100644
index 0000000..0bb64b8
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/response.cpython-311.pyc
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/types.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/types.cpython-311.pyc
new file mode 100644
index 0000000..0af7128
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/__pycache__/types.cpython-311.pyc
Binary files differ
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
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/request.py b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/request.py
new file mode 100644
index 0000000..b4fad18
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/request.py
@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from contextlib import suppress
+from functools import cached_property
+from typing import TYPE_CHECKING, Any
+from urllib.parse import unquote, urlsplit, urlunsplit
+
+from litestar import Request
+from litestar.connection.base import empty_receive, empty_send
+from litestar.contrib.htmx._utils import HTMXHeaders
+from litestar.exceptions import SerializationException
+from litestar.serialization import decode_json
+
+__all__ = ("HTMXDetails", "HTMXRequest")
+
+
+if TYPE_CHECKING:
+ from litestar.types import Receive, Scope, Send
+
+
+class HTMXDetails:
+ """HTMXDetails holds all the values sent by HTMX client in headers and provide convenient properties."""
+
+ def __init__(self, request: Request) -> None:
+ """Initialize :class:`HTMXDetails`"""
+ self.request = request
+
+ def _get_header_value(self, name: HTMXHeaders) -> str | None:
+ """Parse request header
+
+ Check for uri encoded header and unquotes it in readable format.
+ """
+
+ if value := self.request.headers.get(name.value.lower()):
+ is_uri_encoded = self.request.headers.get(f"{name.value.lower()}-uri-autoencoded") == "true"
+ return unquote(value) if is_uri_encoded else value
+ return None
+
+ def __bool__(self) -> bool:
+ """Check if request is sent by an HTMX client."""
+ return self._get_header_value(HTMXHeaders.REQUEST) == "true"
+
+ @cached_property
+ def boosted(self) -> bool:
+ """Check if request is boosted."""
+ return self._get_header_value(HTMXHeaders.BOOSTED) == "true"
+
+ @cached_property
+ def current_url(self) -> str | None:
+ """Current url value sent by HTMX client."""
+ return self._get_header_value(HTMXHeaders.CURRENT_URL)
+
+ @cached_property
+ def current_url_abs_path(self) -> str | None:
+ """Current url abs path value, to get query and path parameter sent by HTMX client."""
+ if self.current_url:
+ split = urlsplit(self.current_url)
+ if split.scheme == self.request.scope["scheme"] and split.netloc == self.request.headers.get("host"):
+ return str(urlunsplit(split._replace(scheme="", netloc="")))
+ return None
+ return self.current_url
+
+ @cached_property
+ def history_restore_request(self) -> bool:
+ """If True then, request is for history restoration after a miss in the local history cache."""
+ return self._get_header_value(HTMXHeaders.HISTORY_RESTORE_REQUEST) == "true"
+
+ @cached_property
+ def prompt(self) -> str | None:
+ """User Response to prompt.
+
+ .. code-block:: html
+
+ <button hx-delete="/account" hx-prompt="Enter your account name to confirm deletion">Delete My Account</button>
+ """
+ return self._get_header_value(HTMXHeaders.PROMPT)
+
+ @cached_property
+ def target(self) -> str | None:
+ """ID of the target element if provided on the element."""
+ return self._get_header_value(HTMXHeaders.TARGET)
+
+ @cached_property
+ def trigger(self) -> str | None:
+ """ID of the triggered element if provided on the element."""
+ return self._get_header_value(HTMXHeaders.TRIGGER_ID)
+
+ @cached_property
+ def trigger_name(self) -> str | None:
+ """Name of the triggered element if provided on the element."""
+ return self._get_header_value(HTMXHeaders.TRIGGER_NAME)
+
+ @cached_property
+ def triggering_event(self) -> Any:
+ """Name of the triggered event.
+
+ This value is added by ``event-header`` extension of HTMX to the ``Triggering-Event`` header to requests.
+ """
+ if value := self._get_header_value(HTMXHeaders.TRIGGERING_EVENT):
+ with suppress(SerializationException):
+ return decode_json(value=value, type_decoders=self.request.route_handler.resolve_type_decoders())
+ return None
+
+
+class HTMXRequest(Request):
+ """HTMX Request class to work with HTMX client."""
+
+ __slots__ = ("htmx",)
+
+ def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send) -> None:
+ """Initialize :class:`HTMXRequest`"""
+ super().__init__(scope=scope, receive=receive, send=send)
+ self.htmx = HTMXDetails(self)
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/response.py b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/response.py
new file mode 100644
index 0000000..0a56e1f
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/response.py
@@ -0,0 +1,200 @@
+from __future__ import annotations
+
+from typing import Any, Generic, TypeVar
+from urllib.parse import quote
+
+from litestar import Response
+from litestar.contrib.htmx._utils import HTMX_STOP_POLLING, get_headers
+from litestar.contrib.htmx.types import (
+ EventAfterType,
+ HtmxHeaderType,
+ LocationType,
+ PushUrlType,
+ ReSwapMethod,
+ TriggerEventType,
+)
+from litestar.response import Template
+from litestar.status_codes import HTTP_200_OK
+
+__all__ = (
+ "ClientRedirect",
+ "ClientRefresh",
+ "HTMXTemplate",
+ "HXLocation",
+ "HXStopPolling",
+ "PushUrl",
+ "ReplaceUrl",
+ "Reswap",
+ "Retarget",
+ "TriggerEvent",
+)
+
+
+# HTMX defined HTTP status code.
+# Response carrying this status code will ask client to stop Polling.
+T = TypeVar("T")
+
+
+class HXStopPolling(Response):
+ """Stop HTMX client from Polling."""
+
+ def __init__(self) -> None:
+ """Initialize"""
+ super().__init__(content=None)
+ self.status_code = HTMX_STOP_POLLING
+
+
+class ClientRedirect(Response):
+ """HTMX Response class to support client side redirect."""
+
+ def __init__(self, redirect_to: str) -> None:
+ """Set status code to 200 (required by HTMX), and pass redirect url."""
+ super().__init__(content=None, headers=get_headers(hx_headers=HtmxHeaderType(redirect=redirect_to)))
+ del self.headers["Location"]
+
+
+class ClientRefresh(Response):
+ """Response to support HTMX client page refresh"""
+
+ def __init__(self) -> None:
+ """Set Status code to 200 and set headers."""
+ super().__init__(content=None, headers=get_headers(hx_headers=HtmxHeaderType(refresh=True)))
+
+
+class PushUrl(Generic[T], Response[T]):
+ """Response to push new url into the history stack."""
+
+ def __init__(self, content: T, push_url: PushUrlType, **kwargs: Any) -> None:
+ """Initialize PushUrl."""
+ super().__init__(
+ content=content,
+ status_code=HTTP_200_OK,
+ headers=get_headers(hx_headers=HtmxHeaderType(push_url=push_url)),
+ **kwargs,
+ )
+
+
+class ReplaceUrl(Generic[T], Response[T]):
+ """Response to replace url in the Browser Location bar."""
+
+ def __init__(self, content: T, replace_url: PushUrlType, **kwargs: Any) -> None:
+ """Initialize ReplaceUrl."""
+ super().__init__(
+ content=content,
+ status_code=HTTP_200_OK,
+ headers=get_headers(hx_headers=HtmxHeaderType(replace_url=replace_url)),
+ **kwargs,
+ )
+
+
+class Reswap(Generic[T], Response[T]):
+ """Response to specify how the response will be swapped."""
+
+ def __init__(
+ self,
+ content: T,
+ method: ReSwapMethod,
+ **kwargs: Any,
+ ) -> None:
+ """Initialize Reswap."""
+ super().__init__(content=content, headers=get_headers(hx_headers=HtmxHeaderType(re_swap=method)), **kwargs)
+
+
+class Retarget(Generic[T], Response[T]):
+ """Response to target different element on the page."""
+
+ def __init__(self, content: T, target: str, **kwargs: Any) -> None:
+ """Initialize Retarget."""
+ super().__init__(content=content, headers=get_headers(hx_headers=HtmxHeaderType(re_target=target)), **kwargs)
+
+
+class TriggerEvent(Generic[T], Response[T]):
+ """Trigger Client side event."""
+
+ def __init__(
+ self,
+ content: T,
+ name: str,
+ after: EventAfterType,
+ params: dict[str, Any] | None = None,
+ **kwargs: Any,
+ ) -> None:
+ """Initialize TriggerEvent."""
+ event = TriggerEventType(name=name, params=params, after=after)
+ headers = get_headers(hx_headers=HtmxHeaderType(trigger_event=event))
+ super().__init__(content=content, headers=headers, **kwargs)
+
+
+class HXLocation(Response):
+ """Client side redirect without full page reload."""
+
+ def __init__(
+ self,
+ redirect_to: str,
+ source: str | None = None,
+ event: str | None = None,
+ target: str | None = None,
+ swap: ReSwapMethod | None = None,
+ hx_headers: dict[str, Any] | None = None,
+ values: dict[str, str] | None = None,
+ **kwargs: Any,
+ ) -> None:
+ """Initialize HXLocation, Set status code to 200 (required by HTMX),
+ and pass redirect url.
+ """
+ super().__init__(
+ content=None,
+ headers={"Location": quote(redirect_to, safe="/#%[]=:;$&()+,!?*@'~")},
+ **kwargs,
+ )
+ spec: dict[str, Any] = get_headers(
+ hx_headers=HtmxHeaderType(
+ location=LocationType(
+ path=str(self.headers.get("Location")),
+ source=source,
+ event=event,
+ target=target,
+ swap=swap,
+ values=values,
+ hx_headers=hx_headers,
+ )
+ )
+ )
+ del self.headers["Location"]
+ self.headers.update(spec)
+
+
+class HTMXTemplate(Template):
+ """HTMX template wrapper"""
+
+ def __init__(
+ self,
+ push_url: PushUrlType | None = None,
+ re_swap: ReSwapMethod | None = None,
+ re_target: str | None = None,
+ trigger_event: str | None = None,
+ params: dict[str, Any] | None = None,
+ after: EventAfterType | None = None,
+ **kwargs: Any,
+ ) -> None:
+ """Create HTMXTemplate response.
+
+ Args:
+ push_url: Either a string value specifying a URL to push to browser history or ``False`` to prevent HTMX client from
+ pushing a url to browser history.
+ re_swap: Method value to instruct HTMX which swapping method to use.
+ re_target: Value for 'id of target element' to apply changes to.
+ trigger_event: Event name to trigger.
+ params: Dictionary of parameters if any required with trigger event parameter.
+ after: Changes to apply after ``receive``, ``settle`` or ``swap`` event.
+ **kwargs: Additional arguments to pass to ``Template``.
+ """
+ super().__init__(**kwargs)
+
+ event: TriggerEventType | None = None
+ if trigger_event:
+ event = TriggerEventType(name=str(trigger_event), params=params, after=after)
+
+ self.headers.update(
+ get_headers(HtmxHeaderType(push_url=push_url, re_swap=re_swap, re_target=re_target, trigger_event=event))
+ )
diff --git a/venv/lib/python3.11/site-packages/litestar/contrib/htmx/types.py b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/types.py
new file mode 100644
index 0000000..aa8f9cd
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/contrib/htmx/types.py
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Literal, TypedDict, Union
+
+__all__ = (
+ "HtmxHeaderType",
+ "LocationType",
+ "TriggerEventType",
+)
+
+if TYPE_CHECKING:
+ from typing_extensions import Required
+
+
+EventAfterType = Literal["receive", "settle", "swap", None]
+
+PushUrlType = Union[str, bool]
+
+ReSwapMethod = Literal[
+ "innerHTML", "outerHTML", "beforebegin", "afterbegin", "beforeend", "afterend", "delete", "none", None
+]
+
+
+class LocationType(TypedDict):
+ """Type for HX-Location header."""
+
+ path: Required[str]
+ source: str | None
+ event: str | None
+ target: str | None
+ swap: ReSwapMethod | None
+ values: dict[str, str] | None
+ hx_headers: dict[str, Any] | None
+
+
+class TriggerEventType(TypedDict):
+ """Type for HX-Trigger header."""
+
+ name: Required[str]
+ params: dict[str, Any] | None
+ after: EventAfterType | None
+
+
+class HtmxHeaderType(TypedDict, total=False):
+ """Type for hx_headers parameter in get_headers()."""
+
+ location: LocationType | None
+ redirect: str | None
+ refresh: bool
+ push_url: PushUrlType | None
+ replace_url: PushUrlType | None
+ re_swap: ReSwapMethod | None
+ re_target: str | None
+ trigger_event: TriggerEventType | None