Coverage for src / lexigram / ui / performance / performance.py: 97%
121 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-10 04:11 +0800
1"""
2Performance utilities for Lexigram Admin UI.
4Provides response optimization, render caching, and lazy loading
5helpers for HTMX-powered components.
6"""
8from __future__ import annotations
10from dataclasses import dataclass, field
11from functools import wraps
12from hashlib import sha256
13from time import time
14from typing import TYPE_CHECKING, Any, TypeVar
16from lexigram.logging import get_logger
17from lexigram.serialization import dumps_str
18from lexigram.ui.core.base import render_to_string
19from lexigram.ui.core.zones import Zones
21if TYPE_CHECKING:
22 from collections.abc import Callable
24logger = get_logger(__name__)
26T = TypeVar("T")
29# === Response Optimization ===
32@dataclass
33class ResponseOptimizer:
34 """
35 Optimize HTMX responses with caching and conditional rendering.
37 Features:
38 - ETag-based caching for unchanged content
39 - Content hashing to detect changes
40 - Conditional response headers
41 """
43 def compute_etag(self, content: str) -> str:
44 """Compute ETag hash for content."""
45 return sha256(content.encode()).hexdigest()[:16]
47 def should_return_304(
48 self,
49 content: str,
50 request_etag: str | None,
51 ) -> bool:
52 """Check if we can return 304 Not Modified."""
53 if not request_etag:
54 return False
55 current_etag = self.compute_etag(content)
56 return request_etag in (current_etag, f'"{current_etag}"')
58 def optimize_response(
59 self,
60 content: str,
61 request_etag: str | None = None,
62 ) -> tuple[str, int, dict[str, str]]:
63 """
64 Optimize response with caching headers.
66 Args:
67 content: The HTML content to return
68 request_etag: The If-None-Match header value from request
70 Returns:
71 Tuple of (content, status_code, headers)
72 """
73 etag = self.compute_etag(content)
75 if self.should_return_304(content, request_etag):
76 return "", 304, {"ETag": f'"{etag}"'}
78 headers = {
79 "ETag": f'"{etag}"',
80 "Cache-Control": "private, no-cache",
81 "Vary": "HX-Request",
82 }
84 return content, 200, headers
87def optimize_htmx_response(
88 content: str,
89 request_etag: str | None = None,
90 optimizer: ResponseOptimizer | None = None,
91) -> tuple[str, int, dict[str, str]]:
92 """Convenience function for response optimization.
94 Args:
95 content: HTML content to optimize.
96 request_etag: ETag from the incoming request for cache validation.
97 optimizer: ResponseOptimizer instance. When None, returns the content unchanged.
98 """
99 if optimizer is None:
100 return content, 200, {}
101 return optimizer.optimize_response(content, request_etag)
104# === Render Caching ===
107@dataclass
108class RenderCache:
109 """
110 LRU cache for rendered component fragments.
112 Use for expensive-to-render components that don't change often.
113 """
115 _cache: dict[str, tuple[str, float]] = field(default_factory=dict)
116 max_size: int = 100
117 ttl_seconds: float = 60.0
119 def _make_key(self, component_name: str, **kwargs: Any) -> str:
120 """Create cache key from component name and params."""
121 sorted_items = sorted(kwargs.items())
122 params_str = dumps_str(sorted_items, sort_keys=True, default=str)
123 return f"{component_name}:{sha256(params_str.encode()).hexdigest()}"
125 def get(self, component_name: str, **kwargs: Any) -> str | None:
126 """Get cached render result if valid."""
127 key = self._make_key(component_name, **kwargs)
128 if key not in self._cache:
129 logger.debug("render_cache.miss", component=component_name)
130 return None
132 content, cached_at = self._cache[key]
133 if time() - cached_at > self.ttl_seconds:
134 del self._cache[key]
135 logger.debug("render_cache.expired", component=component_name)
136 return None
138 logger.debug("render_cache.hit", component=component_name)
139 return content
141 def set(self, component_name: str, content: str, **kwargs: Any) -> None:
142 """Cache a render result."""
143 # Evict old entries if at capacity
144 if len(self._cache) >= self.max_size:
145 # Remove oldest entry
146 oldest_key = min(self._cache, key=lambda k: self._cache[k][1])
147 del self._cache[oldest_key]
148 logger.debug("render_cache.evicted", component=component_name)
150 key = self._make_key(component_name, **kwargs)
151 self._cache[key] = (content, time())
152 logger.debug("render_cache.stored", component=component_name)
154 def invalidate(self, component_name: str | None = None) -> None:
155 """Invalidate cache entries."""
156 if component_name is None:
157 self._cache.clear()
158 else:
159 keys_to_remove = [
160 k for k in self._cache if k.startswith(f"{component_name}:")
161 ]
162 for key in keys_to_remove:
163 del self._cache[key]
165 def cached(
166 self,
167 component_name: str | None = None,
168 ) -> Callable[[Callable[..., str]], Callable[..., str]]:
169 """Decorator for caching render methods."""
171 def decorator(func: Callable[..., str]) -> Callable[..., str]:
172 name = component_name or func.__qualname__
174 @wraps(func)
175 def wrapper(*args: Any, **kwargs: Any) -> str:
176 # Try cache first
177 cached = self.get(name, **kwargs)
178 if cached is not None:
179 return cached
181 # Render and cache
182 result = func(*args, **kwargs)
183 self.set(name, result, **kwargs)
184 return result
186 return wrapper
188 return decorator
191def cached_render(
192 component_name: str | None = None,
193 cache: RenderCache | None = None,
194) -> Callable[[Callable[..., str]], Callable[..., str]]:
195 """Decorator for caching component renders.
197 Args:
198 component_name: Name for the cached component fragment.
199 cache: RenderCache instance to use. When None, returns an identity decorator.
200 """
201 if cache is None:
202 return lambda func: func
203 return cache.cached(component_name)
206# === Lazy Loading ===
209def lazy_load_placeholder(
210 url: str,
211 target_id: str,
212 trigger: str = "load",
213 placeholder: str | None = None,
214) -> str:
215 """
216 Create a lazy-load placeholder that fetches content on trigger.
218 Args:
219 url: URL to fetch content from
220 target_id: ID of the element to replace
221 trigger: HTMX trigger (load, revealed, intersect, etc.)
222 placeholder: Optional placeholder content (defaults to skeleton)
224 Returns:
225 HTML string for the placeholder
226 """
227 from lexigram.ui.atoms.skeleton import Skeleton
228 from lexigram.ui.core.base import el
230 if placeholder is None:
231 placeholder = render_to_string(Skeleton(variant="table", rows=5))
233 return render_to_string(
234 el(
235 "div",
236 placeholder,
237 id=target_id,
238 hx_get=url,
239 hx_trigger=trigger,
240 hx_swap="outerHTML",
241 ),
242 )
245def infinite_scroll_trigger(
246 url: str,
247 target: str | None = None,
248 swap: str = "beforeend",
249 threshold: str = "200px",
250) -> str:
251 """
252 Create an infinite scroll trigger element.
254 Args:
255 url: URL to fetch next page from
256 target: HTMX target selector (defaults to Zones.DATA)
257 swap: HTMX swap mode
258 threshold: Distance from bottom to trigger load
260 Returns:
261 HTML string for the trigger element
262 """
263 from lexigram.ui.atoms.spinner import Spinner
264 from lexigram.ui.core.base import el
266 target = target or Zones.DATA.selector
268 return render_to_string(
269 el(
270 "div",
271 Spinner(size="sm"),
272 class_="flex justify-center py-4",
273 hx_get=url,
274 hx_trigger=f"revealed threshold:{threshold}",
275 hx_target=target,
276 hx_swap=swap,
277 ),
278 )
281# === Debouncing ===
284from lexigram.ui.config import DebounceConfig
287def debounced_search_attrs(
288 url: str,
289 delay_ms: int = 300,
290 target: str | None = None,
291) -> dict[str, str]:
292 """
293 Generate HTMX attributes for a debounced search input.
295 Args:
296 url: URL to search endpoint
297 delay_ms: Debounce delay in milliseconds
298 target: HTMX target (defaults to Zones.DATA)
300 Returns:
301 Dictionary of HTMX attributes
302 """
303 target = target or Zones.DATA.selector
304 config = DebounceConfig(delay_ms=delay_ms)
306 return {
307 "hx-get": url,
308 "hx-trigger": config.to_trigger("input"),
309 "hx-target": target,
310 "hx-swap": Zones.DATA.swap_mode.value,
311 "hx-indicator": ".htmx-indicator",
312 }
315# === Request Coalescing ===
318@dataclass
319class RequestCoalescer:
320 """
321 Coalesce rapid-fire requests into single requests.
323 Useful for batch operations or rapid filter changes.
324 """
326 pending: dict[str, Any] = field(default_factory=dict)
327 window_ms: float = 100
329 def add(self, key: str, value: Any) -> None:
330 """Add a value to coalesce."""
331 self.pending[key] = value
333 def flush(self) -> dict[str, Any]:
334 """Get and clear all pending values."""
335 result = dict(self.pending)
336 self.pending.clear()
337 return result
340# === Performance Middleware Helpers ===
343def add_htmx_timing_header(
344 headers: dict[str, str],
345 render_time_ms: float,
346) -> dict[str, str]:
347 """Add Server-Timing header for HTMX requests."""
348 headers = dict(headers)
349 headers["Server-Timing"] = f"render;dur={render_time_ms:.2f}"
350 return headers
353def measure_render_time(func: Callable[..., str]) -> Callable[..., tuple[str, float]]:
354 """Decorator to measure render time."""
356 @wraps(func)
357 def wrapper(*args: Any, **kwargs: Any) -> tuple[str, float]:
358 start = time()
359 result = func(*args, **kwargs)
360 elapsed_ms = (time() - start) * 1000
361 return result, elapsed_ms
363 return wrapper