Coverage for src / lexigram / contracts / core / middleware.py: 0%

1"""Middleware protocols for Lexigram Framework. 

2 

3This module defines the universal middleware contract that is 

4transport-agnostic and can be adapted to ASGI, WSGI, or other transports. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable 

10 

11if TYPE_CHECKING: 

12 from collections.abc import Awaitable, Callable, Coroutine 

13 

14T = TypeVar("T") 

15TContext = TypeVar("TContext") 

16TResult = TypeVar("TResult") 

17 

18 

19@runtime_checkable 

20class MiddlewareProtocol(Protocol[TContext, TResult]): 

21 """Universal middleware contract. 

22 

23 Middleware wraps async operations with cross-cutting concerns 

24 (logging, metrics, security, etc.). This protocol is transport-agnostic. 

25 

26 Example: 

27 ```python 

28 class LoggingMiddleware(MiddlewareProtocol[dict[str, Any], Any]): 

29 async def __call__( 

30 self, 

31 context: dict[str, Any], 

32 next: Callable[[dict[str, Any]], Awaitable[Any]] 

33 ) -> Any: 

34 print(f"Before: {context}") 

35 result = await next(context) 

36 print(f"After: {result}") 

37 return result 

38 ``` 

39 """ 

40 

41 async def __call__( 

42 self, 

43 context: TContext, 

44 next: Callable[[TContext], Awaitable[TResult]], 

45 ) -> TResult: 

46 """Process the context and call the next middleware/handler. 

47 

48 Args: 

49 context: Arbitrary context (dict, request, scope, etc.) 

50 next: Callable to invoke the next middleware or handler 

51 

52 Returns: 

53 TResult: Result from the next middleware/handler 

54 """ 

55 ... 

56 

57 

58Middleware = MiddlewareProtocol[dict[str, Any], Any] 

59 

60 

61@runtime_checkable 

62class ExceptionFilterChainProtocol(Protocol): 

63 """Protocol for exception filter chains.""" 

64 

65 def add(self, filter_: Any) -> ExceptionFilterChainProtocol: 

66 """Add a filter to the chain. 

67 

68 Args: 

69 filter_: The filter to add (should satisfy ExceptionFilterProtocol protocol). 

70 

71 Returns: 

72 The filter chain for fluent chaining. 

73 """ 

74 ... 

75 

76 async def handle( 

77 self, 

78 exc: Exception, 

79 request: Any, 

80 fallback: Callable[..., Any] | None = None, 

81 ) -> Any: 

82 """Handle an exception by passing it through the chain. 

83 

84 Args: 

85 exc: The exception to handle. 

86 request: The request context. 

87 fallback: Optional fallback handler. 

88 

89 Returns: 

90 The result of the filter chain (e.g. a Response object). 

91 """ 

92 ... 

93 

94 

95@runtime_checkable 

96class MiddlewarePipelineProtocol(Protocol): 

97 """Protocol for an immutable, composable middleware pipeline. 

98 

99 Middleware pipelines chain ``async (context, next_handler) -> result`` 

100 callables so that each middleware wraps the next. Implementations must 

101 be immutable — ``add`` returns a **new** pipeline rather than mutating 

102 the existing one. 

103 

104 Example: 

105 ```python 

106 pipeline = MiddlewarePipeline() 

107 pipeline = pipeline.add(LoggingMiddleware()) 

108 pipeline = pipeline.add(MetricsMiddleware()) 

109 result = await pipeline.execute(context, handler) 

110 ``` 

111 """ 

112 

113 def add(self, middleware: Any) -> MiddlewarePipelineProtocol: 

114 """Return a new pipeline with *middleware* appended. 

115 

116 Args: 

117 middleware: An ``async (context, next_handler) -> result`` callable. 

118 

119 Returns: 

120 A new pipeline with the middleware added. 

121 """ 

122 ... 

123 

124 async def execute( 

125 self, 

126 context: Any, 

127 handler: Callable[..., Coroutine[Any, Any, Any]], 

128 ) -> Any: 

129 """Execute the pipeline, passing *context* through all middleware. 

130 

131 Args: 

132 context: The request/context object to process. 

133 handler: The final ``async (context) -> result`` handler at the 

134 end of the chain. 

135 

136 Returns: 

137 The result from the handler (possibly transformed by middleware). 

138 """ 

139 ... 

140 

141 def __len__(self) -> int: 

142 """Return the number of middleware in the pipeline.""" 

143 ... 

144 

145 

146__all__ = [ 

147 "ExceptionFilterChainProtocol", 

148 "Middleware", 

149 "MiddlewarePipelineProtocol", 

150 "MiddlewareProtocol", 

151]