Coverage for src/lexigram/web/middleware/base.py: 36%

1"""Middleware base class and registry. 

2 

3Canonical Middleware Pattern 

4---------------------------- 

5 

6There is **one** canonical way to write middleware in Lexigram: 

7 

8.. code-block:: python 

9 

10 class MyMiddleware: 

11 def __init__(self, app: ASGIApp) -> None: 

12 self.app = app 

13 

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

15 # before 

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

17 # after (e.g. modify headers via a send wrapper) 

18 

19Register via ``WebProvider`` ``middleware`` kwarg or 

20``provider.middleware_manager.add()``. 

21 

22For route-specific concerns, use guards and interceptors instead of middleware. 

23""" 

24 

25from __future__ import annotations 

26 

27from collections.abc import Callable 

28from enum import IntEnum 

29from typing import Any 

30 

31from starlette.types import ASGIApp 

32 

33from lexigram.logging import get_logger 

34 

35logger = get_logger(__name__) 

36 

37 

38class MiddlewarePriority(IntEnum): 

39 """Simple ordering for middleware registration. 

40 

41 Lower value => higher precedence (executed earlier / outermost). 

42 """ 

43 

44 EARLY = 0 

45 NORMAL = 1 

46 LATE = 2 

47 

48 

49class MiddlewareChain: 

50 """Chain of middleware for sequential execution""" 

51 

52 def __init__(self, middlewares: list[Any]): 

53 self.middlewares = middlewares 

54 

55 async def process_request(self, request: Any) -> Any | None: 

56 """Process request through middleware chain""" 

57 for middleware in self.middlewares: 

58 response = await middleware.process_request(request) 

59 if response is not None: 

60 # Short-circuit the chain 

61 return response 

62 return None 

63 

64 async def process_response(self, response: Any) -> Any: 

65 """Process response through middleware chain (in reverse order)""" 

66 for middleware in reversed(self.middlewares): 

67 response = await middleware.process_response(response) 

68 return response 

69 

70 

71class MiddlewareRegistry: 

72 """Registry to track and prevent duplicate middleware. 

73 

74 Ensures each middleware type is only registered once, 

75 preventing double-execution bugs. 

76 """ 

77 

78 def __init__(self) -> None: 

79 """Initialize middleware registry.""" 

80 self._registered_types: set[type[Any]] = set() 

81 self._registered_names: set[str] = set() 

82 self._middleware_order: list[str] = [] 

83 # Store tuples: (priority: int, middleware_class: Type, options: dict) 

84 self._middleware_stack: list[tuple[int, type[Any], dict]] = [] 

85 # Cache for sorted middleware 

86 self._sorted_cache: list[tuple[int, type[Any], dict]] | None = None 

87 

88 def register_middleware( 

89 self, 

90 middleware_class: type[Any], 

91 *, 

92 priority: MiddlewarePriority = MiddlewarePriority.NORMAL, 

93 **options: Any, 

94 ) -> None: 

95 """Register ASGI middleware class with duplicate checking and priority. 

96 

97 Args: 

98 middleware_class: ASGI Middleware class to register 

99 priority: MiddlewarePriority (EARLY, NORMAL, LATE) 

100 options: Additional middleware options 

101 

102 Raises: 

103 ValueError: If middleware already registered 

104 """ 

105 # Check if already registered 

106 if middleware_class in self._registered_types: 

107 logger.warning( 

108 "Middleware %s already registered, skipping", 

109 middleware_class.__name__, 

110 ) 

111 return 

112 

113 # Invalidate cache when new middleware is added 

114 self._sorted_cache = None 

115 

116 # Track registration 

117 self._registered_types.add(middleware_class) 

118 self._middleware_order.append(middleware_class.__name__) 

119 

120 # Store middleware entry with its priority for later composition 

121 self._middleware_stack.append((int(priority), middleware_class, options)) 

122 

123 logger.info( 

124 "Registered middleware: %s (order: %d, priority: %s)", 

125 middleware_class.__name__, 

126 len(self._middleware_order), 

127 priority.name, 

128 ) 

129 

130 def compose_app(self, app: ASGIApp) -> ASGIApp: 

131 """Compose the middleware stack around the app. 

132 

133 Uses cached sorted middleware for performance. 

134 

135 Args: 

136 app: The base ASGI app 

137 

138 Returns: 

139 App wrapped with all middleware 

140 """ 

141 # Use cached sorted middleware if available 

142 if self._sorted_cache is None: 

143 # Order entries by priority (EARLY -> NORMAL -> LATE) so that when we 

144 # apply them in reverse we get outermost EARLY -> ... -> innermost LATE. 

145 self._sorted_cache = sorted(self._middleware_stack, key=lambda t: t[0]) 

146 

147 # Apply middleware in reverse order (innermost first) 

148 for _, middleware_class, options in reversed(self._sorted_cache): 

149 app = middleware_class(app, **options) 

150 

151 return app 

152 

153 def register_function_middleware( 

154 self, 

155 app: Any, 

156 func: Callable[..., Any], 

157 name: str | None = None, 

158 ) -> None: 

159 """Register function-based middleware with duplicate checking. 

160 

161 Args: 

162 app: Starlette application 

163 func: Middleware function 

164 name: Unique name for this middleware (defaults to function name) 

165 

166 Raises: 

167 ValueError: If middleware with same name already registered 

168 """ 

169 middleware_name = name or func.__name__ 

170 

171 # Check if already registered 

172 if middleware_name in self._registered_names: 

173 logger.warning( 

174 "Function middleware '%s' already registered, skipping", 

175 middleware_name, 

176 ) 

177 return 

178 

179 # Register and track 

180 app.middleware("http")(func) 

181 self._registered_names.add(middleware_name) 

182 self._middleware_order.append(middleware_name) 

183 

184 logger.info( 

185 "Registered function middleware: %s (order: %d)", 

186 middleware_name, 

187 len(self._middleware_order), 

188 ) 

189 

190 def get_middleware_order(self) -> list[str]: 

191 """Get list of registered middleware in execution order. 

192 

193 Returns: 

194 List of middleware names in order they execute 

195 """ 

196 return self._middleware_order.copy() 

197 

198 def is_registered( 

199 self, 

200 middleware: type[Any] | str, 

201 ) -> bool: 

202 """Check if middleware is registered. 

203 

204 Args: 

205 middleware: Middleware class or name 

206 

207 Returns: 

208 True if middleware is registered 

209 """ 

210 if isinstance(middleware, str): 

211 return middleware in self._registered_names 

212 return middleware in self._registered_types