Coverage for src/lexigram/web/interceptors/pipeline.py: 45%

1"""Interceptor pipeline for chaining interceptors. 

2 

3The pipeline executes interceptors in order, passing control to the next 

4interceptor until the handler is reached. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from lexigram.web.protocols import ( 

12 CallHandlerProtocol, 

13 ExecutionContextProtocol, 

14 WebInterceptorProtocol, 

15) 

16 

17if TYPE_CHECKING: 

18 from collections.abc import Callable 

19 

20 

21class DefaultCallHandler(CallHandlerProtocol): 

22 """Default call handler that executes the actual handler function.""" 

23 

24 def __init__(self, handler: Callable): 

25 """Initialize with the handler function. 

26 

27 Args: 

28 handler: The actual handler function to call. 

29 """ 

30 self._handler = handler 

31 

32 async def handle(self) -> Any: 

33 """Execute the handler and return its result.""" 

34 return await self._handler() 

35 

36 

37class InterceptorChain(CallHandlerProtocol): 

38 """Internal chain that links interceptors together.""" 

39 

40 def __init__( 

41 self, 

42 interceptors: list[WebInterceptorProtocol], 

43 index: int, 

44 final_handler: Callable, 

45 context: ExecutionContextProtocol, 

46 ): 

47 """Initialize the chain. 

48 

49 Args: 

50 interceptors: List of interceptors to chain. 

51 index: Current index in the interceptor list. 

52 final_handler: The handler to call when all interceptors complete. 

53 context: The execution context. 

54 """ 

55 self._interceptors = interceptors 

56 self._index = index 

57 self._final_handler = final_handler 

58 self._context = context 

59 

60 async def handle(self) -> Any: 

61 """Execute the next interceptor or final handler.""" 

62 if self._index >= len(self._interceptors): 

63 # All interceptors passed, execute final handler 

64 return await self._final_handler() 

65 

66 # Get current interceptor and create next in chain 

67 interceptor = self._interceptors[self._index] 

68 next_chain = InterceptorChain( 

69 self._interceptors, 

70 self._index + 1, 

71 self._final_handler, 

72 self._context, 

73 ) 

74 

75 # Call the interceptor 

76 return await interceptor.intercept(self._context, next_chain) 

77 

78 

79class InterceptorPipeline: 

80 """Pipeline that manages and executes interceptors. 

81 

82 Interceptors are executed in order, each having the opportunity to: 

83 - Execute code before the handler 

84 - Call the next interceptor/handler 

85 - Execute code after the handler (by transforming the result) 

86 - Short-circuit by returning early without calling next 

87 """ 

88 

89 def __init__(self, interceptors: list[WebInterceptorProtocol] | None = None): 

90 """Initialize the interceptor pipeline. 

91 

92 Args: 

93 interceptors: Optional list of interceptors to register. 

94 """ 

95 self._interceptors: list[WebInterceptorProtocol] = ( 

96 list(interceptors) if interceptors else [] 

97 ) 

98 

99 def add_interceptor(self, interceptor: WebInterceptorProtocol) -> None: 

100 """Add an interceptor to the pipeline. 

101 

102 Interceptors are executed in the order they are added. 

103 

104 Args: 

105 interceptor: The interceptor to add. 

106 """ 

107 if interceptor not in self._interceptors: 

108 self._interceptors.append(interceptor) 

109 

110 def remove_interceptor(self, interceptor: WebInterceptorProtocol) -> None: 

111 """Remove an interceptor from the pipeline. 

112 

113 Args: 

114 interceptor: The interceptor to remove. 

115 """ 

116 if interceptor in self._interceptors: 

117 self._interceptors.remove(interceptor) 

118 

119 def clear(self) -> None: 

120 """Remove all interceptors from the pipeline.""" 

121 self._interceptors.clear() 

122 

123 @property 

124 def interceptors(self) -> list[WebInterceptorProtocol]: 

125 """Get the list of registered interceptors.""" 

126 return list(self._interceptors) 

127 

128 async def execute( 

129 self, 

130 context: ExecutionContextProtocol, 

131 handler: Callable, 

132 ) -> Any: 

133 """Execute the interceptor pipeline. 

134 

135 Args: 

136 context: The execution context. 

137 handler: The final handler to execute. 

138 

139 Returns: 

140 The result of the handler (possibly transformed by interceptors). 

141 """ 

142 if not self._interceptors: 

143 # No interceptors, execute handler directly 

144 return await handler() 

145 

146 # Create the chain starting with first interceptor 

147 chain = InterceptorChain( 

148 self._interceptors, 

149 0, 

150 handler, 

151 context, 

152 ) 

153 

154 return await chain.handle() 

155 

156 def __len__(self) -> int: 

157 """Return the number of interceptors.""" 

158 return len(self._interceptors) 

159 

160 def __bool__(self) -> bool: 

161 """Return True if there are interceptors.""" 

162 return bool(self._interceptors) 

163 

164 

165__all__ = ["DefaultCallHandler", "InterceptorChain", "InterceptorPipeline"]