Coverage for src/lexigram/web/middleware/unified.py: 20%

1"""Unified middleware protocols and adapters.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7if TYPE_CHECKING: 

8 from collections.abc import Callable 

9 

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

11 

12# Re-export the canonical MiddlewareRegistry 

13 

14 

15class RequestResponseMiddlewareAdapter: 

16 """Adapter that wraps request/response middleware for ASGI. 

17 

18 This allows legacy middleware that uses: 

19 process_request(request) -> None 

20 process_response(response) -> Response 

21 

22 to work with the ASGI protocol. 

23 """ 

24 

25 def __init__( 

26 self, 

27 app: ASGIApp, 

28 process_request: Callable | None = None, 

29 process_response: Callable | None = None, 

30 ): 

31 self.app = app 

32 self.process_request = process_request 

33 self.process_response = process_response 

34 

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

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

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

38 return 

39 

40 from starlette.requests import Request 

41 

42 request = Request(scope, receive) 

43 

44 # Run request processing if available 

45 if self.process_request: 

46 await self.process_request(request) 

47 

48 # Create a response wrapper to capture the response 

49 response_wrapper = {"status": None, "headers": None, "body": b""} 

50 

51 async def send_wrapper(message: Any) -> None: 

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

53 # Store status and headers 

54 response_wrapper["status"] = message["status"] 

55 response_wrapper["headers"] = message.get("headers", []) 

56 # Forward the start message to client 

57 await send(message) 

58 elif message["type"] == "http.response.body": 

59 # Store body and forward to client 

60 body = message.get("body", b"") 

61 response_wrapper["body"] += body 

62 await send(message) 

63 

64 # Call the next middleware/app 

65 await self.app(scope, receive, send_wrapper) 

66 

67 # Run response processing if available 

68 if self.process_response and response_wrapper.get("status") is not None: 

69 # Create a simple response-like object for the processor 

70 class SimpleResponse: 

71 def __init__(self, status: Any, headers: Any, body: Any) -> None: 

72 self.status_code = status 

73 self.headers = headers 

74 self.body = body 

75 

76 simple_response = SimpleResponse( 

77 response_wrapper["status"], 

78 response_wrapper["headers"], 

79 response_wrapper["body"], 

80 ) 

81 processed = await self.process_response(simple_response) 

82 if processed: 

83 await processed(scope, receive, send) 

84 

85 

86class ASGIMiddlewareAdapter: 

87 """Adapter that provides ASGI interface for request/response middleware. 

88 

89 Usage: 

90 # Old style: 

91 class MyMiddleware: 

92 async def process_request(self, request): 

93 ... 

94 

95 # Wrap for ASGI: 

96 app = MyMiddleware(app) 

97 """ 

98 

99 def __init__(self, app: ASGIApp): 

100 self.app = app 

101 

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

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