Coverage for fastblocks / htmx.py: 91%

1"""FastBlocks Native HTMX Support. 

2 

3This module consolidates and enhances HTMX functionality for FastBlocks, 

4originally based on the asgi-htmx library. 

5 

6Original asgi-htmx library: 

7- Author: Marcelo Trylesinski 

8- Repository: https://github.com/marcelotrylesisnki/asgi-htmx 

9- License: MIT 

10 

11The FastBlocks implementation extends the original with: 

12- ACB (Asynchronous Component Base) integration 

13- FastBlocks-specific debugging and logging 

14- Enhanced template integration 

15- Response helpers optimized for FastBlocks 

16- Event-driven HTMX updates via ACB Events system 

17""" 

18 

19import asyncio 

20import json 

21import typing as t 

22import warnings 

23from contextlib import suppress 

24from typing import Any 

25from urllib.parse import unquote 

26 

27try: 

28 from acb.debug import debug 

29except ImportError: 

30 # Fallback debug function if acb.debug is not available 

31 import logging 

32 

33 debug = logging.debug 

34from starlette.responses import HTMLResponse 

35 

36if t.TYPE_CHECKING: 

37 from starlette.types import Scope 

38else: 

39 Scope = dict 

40 

41try: 

42 from starlette.requests import Request as StarletteRequest 

43 

44 _starlette_available = True 

45except ImportError: 

46 _starlette_available = False 

47 StarletteRequest: t.Any = t.Any # type: ignore[misc,no-redef] # Fallback when Starlette unavailable 

48 

49STARLETTE_AVAILABLE = _starlette_available 

50 

51 

52class HtmxDetails: 

53 def __init__(self, scope: "Scope") -> None: 

54 self._scope = scope 

55 debug( 

56 f"HtmxDetails: Processing HTMX headers for {scope.get('path', 'unknown')}" 

57 ) 

58 

59 def _get_header(self, name: bytes) -> str | None: 

60 value = _get_header(self._scope, name) 

61 if value and debug: 

62 debug(f"HtmxDetails: {name.decode()}: {value}") 

63 return value 

64 

65 def __bool__(self) -> bool: 

66 is_htmx = self._get_header(b"HX-Request") == "true" 

67 debug(f"HtmxDetails: Is HTMX request: {is_htmx}") 

68 return is_htmx 

69 

70 @property 

71 def boosted(self) -> bool: 

72 return self._get_header(b"HX-Boosted") == "true" 

73 

74 @property 

75 def current_url(self) -> str | None: 

76 return self._get_header(b"HX-Current-URL") 

77 

78 @property 

79 def history_restore_request(self) -> bool: 

80 return self._get_header(b"HX-History-Restore-Request") == "true" 

81 

82 @property 

83 def prompt(self) -> str | None: 

84 return self._get_header(b"HX-Prompt") 

85 

86 @property 

87 def target(self) -> str | None: 

88 return self._get_header(b"HX-Target") 

89 

90 @property 

91 def trigger(self) -> str | None: 

92 return self._get_header(b"HX-Trigger") 

93 

94 @property 

95 def trigger_name(self) -> str | None: 

96 return self._get_header(b"HX-Trigger-Name") 

97 

98 @property 

99 def triggering_event(self) -> t.Any: 

100 value = self._get_header(b"Triggering-Event") 

101 if value is None: 

102 return None 

103 try: 

104 event_data = json.loads(value) 

105 debug(f"HtmxDetails: Parsed triggering event: {event_data}") 

106 return event_data 

107 except json.JSONDecodeError as e: 

108 debug(f"HtmxDetails: Failed to parse triggering event JSON: {e}") 

109 return None 

110 

111 def get_all_headers(self) -> dict[str, str | None]: 

112 headers = { 

113 "HX-Request": self._get_header(b"HX-Request"), 

114 "HX-Boosted": self._get_header(b"HX-Boosted"), 

115 "HX-Current-URL": self.current_url, 

116 "HX-History-Restore-Request": self._get_header( 

117 b"HX-History-Restore-Request" 

118 ), 

119 "HX-Prompt": self.prompt, 

120 "HX-Target": self.target, 

121 "HX-Trigger": self.trigger, 

122 "HX-Trigger-Name": self.trigger_name, 

123 "Triggering-Event": self._get_header(b"Triggering-Event"), 

124 } 

125 

126 return {k: v for k, v in headers.items() if v is not None} 

127 

128 

129def _get_header(scope: "Scope", key: bytes) -> str | None: 

130 key_lower = key.lower() 

131 value: str | None = None 

132 should_unquote = False 

133 

134 # Extract header value and autoencoding flag 

135 try: 

136 for k, v in scope["headers"]: 

137 if k.lower() == key_lower: 

138 value = v.decode("latin-1") 

139 if k.lower() == b"%s-uri-autoencoded" % key_lower and v == b"true": 

140 should_unquote = True 

141 except (KeyError, UnicodeDecodeError) as e: 

142 debug(f"HtmxDetails: Error processing header {key}: {e}") 

143 return None 

144 

145 # Return None if no value found 

146 if value is None: 

147 return None 

148 

149 # Handle URI autoencoding if needed 

150 try: 

151 return unquote(value) if should_unquote else value 

152 except Exception as e: 

153 debug(f"HtmxDetails: Error unquoting header value: {e}") 

154 return value 

155 

156 

157HtmxScope = dict[str, t.Any] 

158 

159if STARLETTE_AVAILABLE and StarletteRequest is not t.Any: 

160 

161 class HtmxRequest(StarletteRequest): # type: ignore[misc] 

162 scope: HtmxScope 

163 

164 @property 

165 def htmx(self) -> HtmxDetails: 

166 return t.cast(HtmxDetails, self.scope["htmx"]) 

167 

168 def is_htmx(self) -> bool: 

169 return bool(self.htmx) 

170 

171 def is_boosted(self) -> bool: 

172 return self.htmx.boosted 

173 

174 def get_htmx_headers(self) -> dict[str, str | None]: 

175 return self.htmx.get_all_headers() 

176else: 

177 

178 class HtmxRequest: # type: ignore[misc,no-redef] 

179 """Placeholder HtmxRequest when Starlette is not available.""" 

180 

181 def __init__(self, *args: t.Any, **kwargs: t.Any) -> None: 

182 raise ImportError( 

183 "Starlette is required for HtmxRequest. Install with: uv add starlette" 

184 ) 

185 

186 

187class HtmxResponse(HTMLResponse): 

188 def __init__( 

189 self, 

190 content: str = "", 

191 status_code: int = 200, 

192 headers: t.Mapping[str, str] | None = None, 

193 media_type: str | None = None, 

194 background: t.Any = None, 

195 trigger: str | None = None, 

196 trigger_after_settle: str | None = None, 

197 trigger_after_swap: str | None = None, 

198 retarget: str | None = None, 

199 reselect: str | None = None, 

200 reswap: str | None = None, 

201 push_url: str | bool | None = None, 

202 replace_url: str | bool | None = None, 

203 refresh: bool = False, 

204 redirect: str | None = None, 

205 location: dict[str, t.Any] | str | None = None, 

206 ) -> None: 

207 init_headers = dict(headers or {}) 

208 

209 # Set HTMX-specific headers 

210 self._set_htmx_headers( 

211 init_headers, 

212 trigger=trigger, 

213 trigger_after_settle=trigger_after_settle, 

214 trigger_after_swap=trigger_after_swap, 

215 retarget=retarget, 

216 reselect=reselect, 

217 reswap=reswap, 

218 push_url=push_url, 

219 replace_url=replace_url, 

220 refresh=refresh, 

221 redirect=redirect, 

222 location=location, 

223 ) 

224 

225 super().__init__( 

226 content=content, 

227 status_code=status_code, 

228 headers=init_headers, 

229 media_type=media_type, 

230 background=background, 

231 ) 

232 

233 def _set_htmx_headers( 

234 self, 

235 headers: dict[str, str], 

236 *, 

237 trigger: str | None = None, 

238 trigger_after_settle: str | None = None, 

239 trigger_after_swap: str | None = None, 

240 retarget: str | None = None, 

241 reselect: str | None = None, 

242 reswap: str | None = None, 

243 push_url: str | bool | None = None, 

244 replace_url: str | bool | None = None, 

245 refresh: bool = False, 

246 redirect: str | None = None, 

247 location: dict[str, t.Any] | str | None = None, 

248 ) -> None: 

249 """Set HTMX-specific headers in the response.""" 

250 if trigger: 

251 headers["HX-Trigger"] = trigger 

252 if trigger_after_settle: 

253 headers["HX-Trigger-After-Settle"] = trigger_after_settle 

254 if trigger_after_swap: 

255 headers["HX-Trigger-After-Swap"] = trigger_after_swap 

256 if retarget: 

257 headers["HX-Retarget"] = retarget 

258 if reselect: 

259 headers["HX-Reselect"] = reselect 

260 if reswap: 

261 headers["HX-Reswap"] = reswap 

262 if push_url is not None: 

263 headers["HX-Push-Url"] = str(push_url).lower() 

264 if replace_url is not None: 

265 headers["HX-Replace-Url"] = str(replace_url).lower() 

266 if refresh: 

267 headers["HX-Refresh"] = "true" 

268 if redirect: 

269 headers["HX-Redirect"] = redirect 

270 if location: 

271 if isinstance(location, dict): 

272 headers["HX-Location"] = json.dumps(location) 

273 else: 

274 headers["HX-Location"] = str(location) 

275 

276 

277def htmx_trigger( 

278 trigger_events: str | dict[str, t.Any], 

279 content: str = "", 

280 status_code: int = 200, 

281 **kwargs: t.Any, 

282) -> HtmxResponse: 

283 trigger_data: dict[str, Any] 

284 if isinstance(trigger_events, dict): 

285 trigger_value = json.dumps(trigger_events) 

286 trigger_name = next(iter(trigger_events.keys()), "custom_trigger") 

287 trigger_data = trigger_events 

288 else: 

289 trigger_value = trigger_events 

290 trigger_name = trigger_events 

291 trigger_data = {} 

292 

293 # Schedule event publishing in background 

294 def _run_publish_event() -> None: 

295 async def _publish_event( 

296 trigger_name: str, trigger_data: dict[str, t.Any] 

297 ) -> None: 

298 from .adapters.templates._events_wrapper import publish_htmx_trigger 

299 

300 await publish_htmx_trigger( 

301 trigger_name=trigger_name, 

302 trigger_data=trigger_data, 

303 ) 

304 

305 # Create and schedule the task to run the async function 

306 

307 with warnings.catch_warnings(): 

308 warnings.simplefilter("ignore", RuntimeWarning) 

309 task = _publish_event(trigger_name, trigger_data) 

310 asyncio.create_task(task) 

311 

312 with suppress(Exception): 

313 _run_publish_event() 

314 

315 return HtmxResponse( 

316 content=content, 

317 status_code=status_code, 

318 trigger=trigger_value, 

319 **kwargs, 

320 ) 

321 

322 

323def htmx_redirect(url: str, **kwargs: t.Any) -> HtmxResponse: 

324 # Schedule event publishing in background 

325 def _run_publish_event() -> None: 

326 async def _publish_event(url: str) -> None: 

327 from ._events_integration import get_event_publisher 

328 

329 publisher = get_event_publisher() 

330 if publisher: 

331 await publisher.publish_htmx_update( 

332 update_type="redirect", 

333 target=url, 

334 ) 

335 

336 # Create and schedule the task to run the async function 

337 

338 with warnings.catch_warnings(): 

339 warnings.simplefilter("ignore", RuntimeWarning) 

340 task = _publish_event(url) 

341 asyncio.create_task(task) 

342 

343 with suppress(Exception): 

344 _run_publish_event() 

345 

346 return HtmxResponse(redirect=url, **kwargs) 

347 

348 

349def htmx_refresh(**kwargs: t.Any) -> HtmxResponse: 

350 # Get target from kwargs if provided 

351 target = kwargs.get("target", "#body") 

352 

353 # Schedule event publishing in background 

354 def _run_publish_event() -> None: 

355 async def _publish_event(target: str) -> None: 

356 from .adapters.templates._events_wrapper import publish_htmx_refresh 

357 

358 await publish_htmx_refresh(target=target) 

359 

360 # Create and schedule the task to run the async function 

361 

362 with warnings.catch_warnings(): 

363 warnings.simplefilter("ignore", RuntimeWarning) 

364 task = _publish_event(target) 

365 asyncio.create_task(task) 

366 

367 with suppress(Exception): 

368 _run_publish_event() 

369 

370 return HtmxResponse(refresh=True, **kwargs) 

371 

372 

373def htmx_push_url(url: str, content: str = "", **kwargs: t.Any) -> HtmxResponse: 

374 return HtmxResponse(content=content, push_url=url, **kwargs) 

375 

376 

377def htmx_retarget(target: str, content: str = "", **kwargs: t.Any) -> HtmxResponse: 

378 return HtmxResponse(content=content, retarget=target, **kwargs) 

379 

380 

381def is_htmx(scope_or_request: dict[str, t.Any] | t.Any) -> bool: 

382 if hasattr(scope_or_request, "headers"): 

383 headers = getattr(scope_or_request, "headers", {}) 

384 return headers.get("HX-Request") == "true" 

385 else: 

386 if isinstance(scope_or_request, dict): 

387 details = HtmxDetails(scope_or_request) 

388 return getattr(details, "is_htmx", False) 

389 return False 

390 

391 

392__all__ = [ 

393 "HtmxDetails", 

394 "HtmxRequest", 

395 "HtmxResponse", 

396 "htmx_trigger", 

397 "htmx_redirect", 

398 "htmx_refresh", 

399 "htmx_push_url", 

400 "htmx_retarget", 

401 "is_htmx", 

402]