Coverage for src / lexigram / admin / services / htmx_perf.py: 43%

1"""HTMX performance monitoring for Lexigram Admin. 

2 

3For pure HTMX attribute helpers (hx_swap_oob, hx_prefetch, etc.) 

4see lexigram.ui.htmx.helpers. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Any 

10 

11 

12class HTMXPerformanceMonitor: 

13 """Monitor and track HTMX request performance. 

14 

15 Helps identify slow endpoints and optimization opportunities. 

16 """ 

17 

18 def __init__(self) -> None: 

19 """Initialize with empty request log.""" 

20 self._requests: list[dict[str, Any]] = [] 

21 self._enabled = True 

22 

23 def enable(self) -> None: 

24 """Enable performance monitoring.""" 

25 self._enabled = True 

26 

27 def disable(self) -> None: 

28 """Disable performance monitoring.""" 

29 self._enabled = False 

30 

31 def record_request( 

32 self, 

33 url: str, 

34 method: str, 

35 duration_ms: float, 

36 size_bytes: int | None = None, 

37 ) -> None: 

38 """Record an HTMX request. 

39 

40 Args: 

41 url: Request URL 

42 method: HTTP method 

43 duration_ms: Request duration in milliseconds 

44 size_bytes: Response size in bytes 

45 """ 

46 if not self._enabled: 

47 return 

48 

49 self._requests.append( 

50 { 

51 "url": url, 

52 "method": method, 

53 "duration_ms": duration_ms, 

54 "size_bytes": size_bytes, 

55 }, 

56 ) 

57 

58 def get_stats(self) -> dict[str, Any]: 

59 """Get performance statistics. 

60 

61 Returns: 

62 Dictionary with performance metrics 

63 """ 

64 if not self._requests: 

65 return { 

66 "total_requests": 0, 

67 "avg_duration_ms": 0, 

68 "max_duration_ms": 0, 

69 "total_size_bytes": 0, 

70 } 

71 

72 durations = [r["duration_ms"] for r in self._requests] 

73 sizes = [ 

74 r["size_bytes"] for r in filter(lambda r: r["size_bytes"], self._requests) 

75 ] 

76 

77 return { 

78 "total_requests": len(self._requests), 

79 "avg_duration_ms": sum(durations) / len(durations), 

80 "max_duration_ms": max(durations), 

81 "total_size_bytes": sum(sizes) if sizes else 0, 

82 "slow_requests": [r for r in self._requests if r["duration_ms"] > 1000], 

83 } 

84 

85 def clear(self) -> None: 

86 """Clear recorded requests.""" 

87 self._requests.clear() 

88 

89 

90async def get_htmx_monitor(context: Any | None = None) -> HTMXPerformanceMonitor: 

91 """Get the global HTMX performance monitor instance.""" 

92 from lexigram.admin.lib.di import get_admin_resolver 

93 

94 resolver = get_admin_resolver(context) 

95 return await resolver.resolve(HTMXPerformanceMonitor) 

96 

97 

98def __getattr__(name: str) -> Any: 

99 if name == "htmx_monitor": 

100 raise AttributeError( 

101 "htmx_monitor is now async. Use: await get_htmx_monitor()", 

102 ) 

103 raise AttributeError(f"module {__name__} has no attribute {name}")