Coverage for src/lexigram/web/middleware/adapter.py: 0%

1"""Middleware Adapter - Bridge Lexigram to Starlette""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, cast 

6 

7from starlette.requests import Request as StarletteRequest 

8from starlette.responses import ( 

9 HTMLResponse as StarletteHTMLResponse, 

10) 

11from starlette.responses import ( 

12 Response as StarletteResponse, 

13) 

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

15 

16from lexigram.web.transport.responses import Response as WebResponse 

17 

18 

19class _LexigramMiddlewareAdapter: 

20 """Adapter that bridges Lexigram's Middleware protocol to Starlette's ASGI pipeline""" 

21 

22 def __init__(self, app: ASGIApp, lexigram_mw: Any): 

23 self.app = app 

24 self.lexigram_mw = lexigram_mw 

25 

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

27 """Pure ASGI implementation to support HTTP and WebSocket (pass-through)""" 

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

29 # Pass through non-HTTP requests (WebSocket, lifespan, etc.) 

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

31 return 

32 

33 # 1. Expose the Starlette request to Lexigram middleware 

34 request = StarletteRequest(scope, receive) 

35 

36 if callable(self.lexigram_mw): 

37 # Support for functional middleware: async def mw(request, call_next) 

38 response_started = [False] 

39 

40 async def lexigram_call_next(req: StarletteRequest) -> Any: 

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

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

43 response_started[0] = True 

44 await send(message) 

45 

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

47 return None 

48 

49 result = await cast("Any", self.lexigram_mw)(request, lexigram_call_next) 

50 # Only send result if the response hasn't been sent already. 

51 # If the inner app sent an error response and re-raised, a 

52 # functional middleware may catch and return a new response. 

53 # Sending it would cause a double-response RuntimeError. 

54 if result is not None and not response_started[0]: 

55 if hasattr(result, "__call__"): 

56 await result(scope, receive, send) 

57 else: 

58 response = self._convert_to_starlette_response(result) 

59 await response(scope, receive, send) 

60 return 

61 # 3. Lifecycle pattern: Execute Lexigram middleware pre-processing 

62 if hasattr(self.lexigram_mw, "process_request"): 

63 await self.lexigram_mw.process_request(request) 

64 

65 # 4. Call next in ASGI pipeline 

66 # We wrap 'send' to intercept for post-processing if needed 

67 if hasattr(self.lexigram_mw, "process_response"): 

68 

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

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

71 # We could potentially modify headers here 

72 pass 

73 await send(message) 

74 

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

76 

77 # Post-processing usually happens via Response object in Lexigram 

78 # For pure ASGI, we'd need to capture the body if we want to modify it. 

79 # Piccolina doesn't currently use body-modifying middleware via this adapter. 

80 else: 

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

82 

83 def _convert_to_starlette_response(self, response: Any) -> StarletteResponse: 

84 """Convert Lexigram Response back to Starlette Response (Legacy)""" 

85 # Kept for compatibility if needed, but pure ASGI uses send() 

86 if isinstance(response, StarletteResponse) and not isinstance( 

87 response, 

88 WebResponse, 

89 ): 

90 return response 

91 

92 # Already a StarletteResponse subclass? Check if it's Lexigram's wrapper 

93 lexigram_response = response 

94 

95 headers = dict(lexigram_response.headers or {}) 

96 content = getattr(lexigram_response, "body", None) 

97 if content is None: 

98 # Try to get from Starlette-style property if it's a wrapper 

99 content = getattr(lexigram_response, "_content", b"") 

100 

101 ct = headers.get("content-type") or headers.get("Content-Type", "") 

102 ct = (ct or "").lower() 

103 

104 # If content is already bytes, use raw StarletteResponse 

105 if isinstance(content, bytes): 

106 return StarletteResponse( 

107 content=content, 

108 status_code=lexigram_response.status_code, 

109 headers=headers, 

110 ) 

111 

112 if "text/html" in ct: 

113 return StarletteHTMLResponse( 

114 content=content, 

115 status_code=lexigram_response.status_code, 

116 headers=headers, 

117 ) 

118 if "application/json" in ct: 

119 from lexigram.web.transport.responses import JSONResponse as WebJSONResponse 

120 

121 return cast( 

122 "StarletteResponse", 

123 WebJSONResponse( 

124 content=content, 

125 status_code=lexigram_response.status_code, 

126 headers=headers, 

127 ), 

128 ) 

129 return StarletteResponse( 

130 content=content, 

131 status_code=lexigram_response.status_code, 

132 headers=headers, 

133 ) 

134 

135 

136class _SimpleASGIMiddlewareAdapter: 

137 """Adapter to convert transport-agnostic middleware to ASGI format. 

138 

139 Relocated from lexigram core. 

140 """ 

141 

142 def __init__(self, middleware: Any) -> None: 

143 self.middleware = middleware 

144 

145 async def __call__( 

146 self, 

147 scope: dict[str, Any], 

148 receive: Any, 

149 send: Any, 

150 ) -> None: 

151 """ASGI-compatible call.""" 

152 result: dict[str, Any] = {} 

153 

154 async def call_next() -> Any: 

155 async def asgi_handler() -> None: 

156 await send(result) 

157 

158 return await self._run_inner(asgi_handler) 

159 

160 try: 

161 await cast("Any", self.middleware)(scope, call_next) 

162 except Exception as e: # noqa: BLE001 — ASGI middleware adapter must capture any middleware error to record it before re-raising 

163 result["error"] = str(e) 

164 raise 

165 

166 async def _run_inner(self, handler: Any) -> None: 

167 await handler() 

168 

169 

170__all__ = ["_LexigramMiddlewareAdapter", "_SimpleASGIMiddlewareAdapter"]