Coverage for src/lexigram/web/routing/pipeline.py: 14%

1"""Web request execution pipeline. 

2 

3Orchestrates the execution of guards, interceptors, and handlers. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING, Any 

9 

10from lexigram.web.interceptors.pipeline import InterceptorChain 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.web.guard import GuardProtocol 

14 from lexigram.contracts.web.protocols import ExceptionFilterProtocol 

15 from lexigram.web.filters.pipeline import FilterPipeline 

16 from lexigram.web.protocols import WebInterceptorProtocol 

17 from lexigram.web.routing.execution_context import WebExecutionContext 

18 from lexigram.web.serialization.serializers import ResponseSerializer 

19 

20 

21class RequestPipeline: 

22 """Orchestrates the request execution lifecycle.""" 

23 

24 def __init__( 

25 self, 

26 guards: list[GuardProtocol] | None = None, 

27 interceptors: list[WebInterceptorProtocol] | None = None, 

28 filters: list[ExceptionFilterProtocol] | None = None, 

29 fallback_pipeline: FilterPipeline | None = None, 

30 response_serializer: ResponseSerializer | None = None, 

31 ): 

32 """Initialize the pipeline. 

33 

34 Args: 

35 guards: List of guards to execute. 

36 interceptors: List of interceptors to execute. 

37 filters: List of exception filters to execute. 

38 fallback_pipeline: Optional fallback pipeline (e.g. from the Router). 

39 response_serializer: Optional pre-wired ResponseSerializer instance. 

40 """ 

41 self.guards = guards or [] 

42 self.interceptors = interceptors or [] 

43 self.filters = filters or [] 

44 self.fallback_pipeline = fallback_pipeline 

45 self.response_serializer = response_serializer 

46 

47 async def execute(self, context: WebExecutionContext) -> Any: 

48 """Execute the pipeline for the given context. 

49 

50 Args: 

51 context: The execution context. 

52 

53 Returns: 

54 The result of the handler (possibly transformed). 

55 """ 

56 try: 

57 # 1. Execute Guards 

58 for guard in self.guards: 

59 if not await guard.can_activate(context): 

60 from lexigram.web.exceptions import ForbiddenError 

61 

62 raise ForbiddenError("Access denied by guard") 

63 

64 # 2. Execute Interceptors & Handler 

65 # We wrap the final handler call in a CallHandlerProtocol to be used by interceptors 

66 async def final_handler() -> Any: 

67 from lexigram.web.routing.parameter_binder import ParameterBinder 

68 

69 # Resolve parameters using binder 

70 sig_info = context.route_metadata.get("sig_info") 

71 if not sig_info: 

72 # Fallback to computing on the fly (less efficient) 

73 import inspect 

74 

75 from lexigram.web.routing.validation import ( 

76 _cached_get_type_hints_for_handler, 

77 ) 

78 

79 sig = inspect.signature(context.handler) 

80 hints = _cached_get_type_hints_for_handler(context.handler) 

81 sig_info = {"sig": sig, "hints": hints} 

82 

83 binder = ParameterBinder(sig_info["sig"], sig_info["hints"]) 

84 kwargs = await binder.bind(context) 

85 

86 return await context.handler(**kwargs) 

87 

88 # Build interceptor chain 

89 chain = InterceptorChain(self.interceptors, 0, final_handler, context) 

90 result = await chain.handle() 

91 

92 # 3. Post-execution: Serialize result to Response if needed 

93 from starlette.responses import Response as StarletteResponse 

94 

95 if isinstance(result, StarletteResponse): 

96 return result 

97 

98 from lexigram.web.serialization.serializers import ResponseSerializer 

99 

100 route = context.route_metadata.get("route") 

101 serializer = self.response_serializer or ResponseSerializer() 

102 

103 return await serializer.serialize( 

104 result, 

105 context.request, 

106 route=route, 

107 ) 

108 

109 except Exception as e: # noqa: BLE001 — pipeline safety net; dispatches any error to registered exception filters 

110 # Match and execute appropriate filter 

111 # Local filters take precedence 

112 for f in reversed(self.filters): 

113 if f.can_handle(e): 

114 return await f.handle(e, context.request) 

115 

116 # Check for fallback pipeline (e.g. from the Router) 

117 if self.fallback_pipeline: 

118 return await self.fallback_pipeline.handle(e, context.request) 

119 

120 # Check for truly global filters if no fallback handled it 

121 from lexigram.web.filters.pipeline import get_filter_pipeline 

122 

123 global_pipeline = get_filter_pipeline(context.container) 

124 if global_pipeline and global_pipeline is not self.fallback_pipeline: 

125 return await global_pipeline.handle(e, context.request) 

126 

127 # Re-raise if no filter found 

128 raise 

129 

130 

131__all__ = ["RequestPipeline"]