Coverage for src/lexigram/web/middleware/timing.py: 32%

1"""Request timing middleware.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from starlette.types import ASGIApp, Receive, Scope, Send 

8 

9from lexigram.primitives import clock as ambient_clock 

10 

11 

12class TimingMiddleware: 

13 """Pure ASGI middleware that adds request timing headers. 

14 

15 10x faster than BaseHTTPMiddleware - no task creation overhead. 

16 """ 

17 

18 def __init__( 

19 self, 

20 app: ASGIApp, 

21 header_name: str = "X-Process-Time", 

22 ): 

23 self.app = app 

24 self.header_name = header_name 

25 

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

27 """Pure ASGI middleware implementation.""" 

28 

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

30 # Pass through non-HTTP requests 

31 await self.app(scope, receive, send) 

32 return 

33 

34 # Start timing 

35 start_time = ambient_clock.timestamp() 

36 

37 # Wrap send to add timing header to response 

38 async def send_with_timing(message: Any) -> None: 

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

40 # Calculate duration and add header 

41 duration_ms = (ambient_clock.timestamp() - start_time) * 1000 

42 timing_header = f"{duration_ms:.2f}" 

43 

44 headers = list(message.get("headers", [])) 

45 # Add timing header 

46 headers.append([self.header_name.encode(), timing_header.encode()]) 

47 message["headers"] = headers 

48 

49 await send(message) 

50 

51 # Process request 

52 await self.app(scope, receive, send_with_timing)