Coverage for src / lexigram / admin / middleware / nav_push.py: 25%

1"""HX-Push-Url middleware for full-page body swaps. 

2 

3htmx only updates the browser URL when the server responds with an 

4``HX-Push-Url`` header (or the element carries ``hx-push-url``). Fragment 

5responses (list data zones) already set this header themselves; this 

6middleware covers full-page responses targeted at ``body`` so that 

7client-side navigation via ``htmx.ajax(..., {target: "body"})`` keeps the 

8address bar in sync and htmx history (back/forward) works. 

9""" 

10 

11from __future__ import annotations 

12 

13from starlette.types import ASGIApp, Message, Receive, Scope, Send 

14 

15 

16class AdminNavPushMiddleware: 

17 """Add ``HX-Push-Url`` to HTML responses for body-targeted htmx GETs. 

18 

19 Any htmx request without an ``HX-Target`` (or with target ``body``) 

20 that receives a 2xx HTML response gets an ``HX-Push-Url`` header set 

21 to the request URL, so the browser address bar matches the page the 

22 server rendered. 

23 """ 

24 

25 def __init__(self, app: ASGIApp) -> None: 

26 """Initialise the middleware. 

27 

28 Args: 

29 app: The next ASGI application in the stack. 

30 """ 

31 self._app = app 

32 

33 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

34 """Process an incoming HTTP request. 

35 

36 Args: 

37 scope: ASGI connection scope. 

38 receive: ASGI receive callable. 

39 send: ASGI send callable. 

40 """ 

41 if scope["type"] != "http": 

42 await self._app(scope, receive, send) 

43 return 

44 

45 headers = scope.get("headers") or [] 

46 if not self._is_body_targeted_get(headers, scope["method"]): 

47 await self._app(scope, receive, send) 

48 return 

49 

50 push_url = self._build_push_url(scope) 

51 push_url_bytes = push_url.encode("utf-8") 

52 

53 async def send_with_push(message: Message) -> None: 

54 if message["type"] == "http.response.start": 

55 status: int = message.get("status", 0) 

56 response_headers = message.get("headers") or [] 

57 if 200 <= status < 300 and self._is_html(response_headers): 

58 message["headers"] = [ 

59 *response_headers, 

60 (b"hx-push-url", push_url_bytes), 

61 ] 

62 await send(message) 

63 

64 await self._app(scope, receive, send_with_push) 

65 

66 # ------------------------------------------------------------------ 

67 # Helpers 

68 # ------------------------------------------------------------------ 

69 

70 @classmethod 

71 def _is_body_targeted_get( 

72 cls, headers: list[tuple[bytes, bytes]], method: str 

73 ) -> bool: 

74 """Return True for htmx GET/HEAD requests targeting the body. 

75 

76 Args: 

77 headers: Request headers from the ASGI scope. 

78 method: Request method. 

79 

80 Returns: 

81 True when the request is a GET/HEAD htmx request without a 

82 specific swap target (or with target ``body``). 

83 """ 

84 if method not in ("GET", "HEAD"): 

85 return False 

86 if cls._get_header(headers, b"hx-request") != b"true": 

87 return False 

88 target = cls._get_header(headers, b"hx-target") 

89 return target is None or target == b"body" 

90 

91 @staticmethod 

92 def _get_header(headers: list[tuple[bytes, bytes]], name: bytes) -> bytes | None: 

93 """Return a header value, case-insensitively. 

94 

95 Args: 

96 headers: Header tuples from an ASGI scope or message. 

97 name: Lower-case header name to look up. 

98 

99 Returns: 

100 The header value, or None when absent. 

101 """ 

102 for key, value in headers: 

103 if key.lower() == name: 

104 return value 

105 return None 

106 

107 @classmethod 

108 def _is_html(cls, headers: list[tuple[bytes, bytes]]) -> bool: 

109 """Return True when the response content type is HTML. 

110 

111 Args: 

112 headers: Response header tuples from the start message. 

113 

114 Returns: 

115 True when the Content-Type contains ``text/html``. 

116 """ 

117 content_type = cls._get_header(headers, b"content-type") 

118 return content_type is not None and b"text/html" in content_type 

119 

120 @staticmethod 

121 def _build_push_url(scope: Scope) -> str: 

122 """Build the pushable URL (path + query) for the request. 

123 

124 ``raw_path`` is preferred because Starlette's ``Mount`` rewrites 

125 ``scope["path"]`` to the sub-path while ``root_path`` carries the 

126 prefix, and combining them double-counts the mount prefix. 

127 

128 Args: 

129 scope: ASGI connection scope. 

130 

131 Returns: 

132 The full URL path including any query string. 

133 """ 

134 raw_path = scope.get("raw_path") 

135 if raw_path: 

136 return raw_path.decode("latin-1") 

137 root_path = scope.get("root_path", "") 

138 path: str = scope["path"] 

139 query: bytes = scope.get("query_string", b"") 

140 query_string = query.decode("latin-1") if query else "" 

141 return f"{root_path}{path}{'?' + query_string if query_string else ''}"