Source code for django_htmx.middleware
from __future__ import annotations
import json
from typing import Any
from typing import Awaitable
from typing import Callable
from urllib.parse import unquote
from urllib.parse import urlsplit
from urllib.parse import urlunsplit
from asgiref.sync import iscoroutinefunction
from asgiref.sync import markcoroutinefunction
from django.http import HttpRequest
from django.http.response import HttpResponseBase
from django.utils.functional import cached_property
[docs]
class HtmxMiddleware:
sync_capable = True
async_capable = True
def __init__(
self,
get_response: (
Callable[[HttpRequest], HttpResponseBase]
| Callable[[HttpRequest], Awaitable[HttpResponseBase]]
),
) -> None:
self.get_response = get_response
self.async_mode = iscoroutinefunction(self.get_response)
if self.async_mode:
# Mark the class as async-capable, but do the actual switch
# inside __call__ to avoid swapping out dunder methods
markcoroutinefunction(self)
def __call__(
self, request: HttpRequest
) -> HttpResponseBase | Awaitable[HttpResponseBase]:
if self.async_mode:
return self.__acall__(request)
request.htmx = HtmxDetails(request) # type: ignore [attr-defined]
return self.get_response(request)
async def __acall__(self, request: HttpRequest) -> HttpResponseBase:
request.htmx = HtmxDetails(request) # type: ignore [attr-defined]
return await self.get_response(request) # type: ignore [no-any-return, misc]
[docs]
class HtmxDetails:
def __init__(self, request: HttpRequest) -> None:
self.request = request
def _get_header_value(self, name: str) -> str | None:
value = self.request.headers.get(name) or None
if value:
if self.request.headers.get(f"{name}-URI-AutoEncoded") == "true":
value = unquote(value)
return value
@cached_property
def boosted(self) -> bool:
return self._get_header_value("HX-Boosted") == "true"
@cached_property
def current_url(self) -> str | None:
return self._get_header_value("HX-Current-URL")
@cached_property
def current_url_abs_path(self) -> str | None:
url = self.current_url
if url is not None:
split = urlsplit(url)
if (
split.scheme == self.request.scheme
and split.netloc == self.request.get_host()
):
url = urlunsplit(split._replace(scheme="", netloc=""))
else:
url = None
return url
@cached_property
def history_restore_request(self) -> bool:
return self._get_header_value("HX-History-Restore-Request") == "true"
@cached_property
def prompt(self) -> str | None:
return self._get_header_value("HX-Prompt")
@cached_property
def target(self) -> str | None:
return self._get_header_value("HX-Target")
@cached_property
def trigger(self) -> str | None:
return self._get_header_value("HX-Trigger")
@cached_property
def trigger_name(self) -> str | None:
return self._get_header_value("HX-Trigger-Name")
@cached_property
def triggering_event(self) -> Any:
value = self._get_header_value("Triggering-Event")
if value is not None:
try:
value = json.loads(value)
except json.JSONDecodeError:
value = None
return value