Coverage for src/lexigram/web/interceptors/builtin/timing.py: 35%

1"""Timing interceptor for performance measurement. 

2 

3Adds Server-Timing header with handler execution time. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any 

9 

10from lexigram.primitives import clock as ambient_clock 

11from lexigram.web.protocols import ( 

12 CallHandlerProtocol, 

13 ExecutionContextProtocol, 

14 WebInterceptorProtocol, 

15) 

16 

17 

18class HandlerTimingInterceptor(WebInterceptorProtocol): 

19 """Interceptor that measures and reports request handling time. 

20 

21 Adds Server-Timing header with the handler execution duration, 

22 useful for performance debugging and client-side profiling. 

23 

24 Example: 

25 ```python 

26 # Add to global interceptors 

27 router.add_interceptor(HandlerTimingInterceptor()) 

28 ``` 

29 

30 The Server-Timing header format: 

31 Server-Timing: handler;dur=12.5 

32 """ 

33 

34 def __init__( 

35 self, 

36 metric_name: str = "handler", 

37 precision: int = 1, 

38 ): 

39 """Initialize the timing interceptor. 

40 

41 Args: 

42 metric_name: Name for the timing metric in Server-Timing header. 

43 precision: Number of decimal places for timing (default 1). 

44 """ 

45 self._metric_name = metric_name 

46 self._precision = precision 

47 

48 async def intercept( 

49 self, 

50 context: ExecutionContextProtocol, 

51 next_handler: CallHandlerProtocol, 

52 ) -> Any: 

53 """Intercept and measure handler execution time. 

54 

55 Args: 

56 context: The execution context. 

57 next_handler: The next handler in the chain. 

58 

59 Returns: 

60 The response from the handler, with timing header added. 

61 """ 

62 start = ambient_clock.monotonic() 

63 

64 try: 

65 result = await next_handler.handle() 

66 finally: 

67 elapsed = ambient_clock.monotonic() - start 

68 duration_ms = elapsed * 1000 

69 

70 # Add Server-Timing header to response 

71 if result is not None: 

72 timing_value = ( 

73 f"{self._metric_name};dur={duration_ms:.{self._precision}f}" 

74 ) 

75 

76 # Add header to response if it has headers 

77 if hasattr(result, "headers"): 

78 # Use a setdefault to avoid overwriting existing headers 

79 if "server-timing" not in result.headers: 

80 result.headers["Server-Timing"] = timing_value 

81 elif hasattr(result, "add_header"): 

82 result.add_header("Server-Timing", timing_value) 

83 

84 return result 

85 

86 

87__all__ = ["HandlerTimingInterceptor"]