Coverage for src/lexigram/web/routing/parameter_binder.py: 17%

1"""Parameter binder for web handlers. 

2 

3Resolves handler arguments from HTTP requests, supporting multiple sources 

4(Path, Query, Body, Header, Cookie), DI, and special types. 

5""" 

6 

7from __future__ import annotations 

8 

9import inspect 

10from typing import Any, cast 

11 

12from lexigram.logging import get_logger 

13from lexigram.web.exceptions import DependencyResolutionError 

14from lexigram.web.protocols import ExecutionContextProtocol 

15from lexigram.web.routing.execution_context import WebExecutionContext 

16 

17logger = get_logger(__name__) 

18 

19 

20class ParameterBinder: 

21 """Binds request data to handler parameters.""" 

22 

23 def __init__(self, sig: inspect.Signature, hints: dict[str, Any]): 

24 """Initialize the binder. 

25 

26 Args: 

27 sig: The handler signature. 

28 hints: Resolved type hints for the handler. 

29 """ 

30 self.sig = sig 

31 self.hints = hints 

32 

33 async def bind(self, context: WebExecutionContext) -> dict[str, Any]: 

34 """Bind request data to parameters for the current context. 

35 

36 Args: 

37 context: The execution context containing the request and container. 

38 

39 Returns: 

40 Dictionary of keyword arguments for the handler call. 

41 """ 

42 from lexigram.web.routing.validation import validate_and_merge_request 

43 

44 request = context.request 

45 handler = context.handler 

46 container = context.container 

47 

48 kwargs = {} 

49 path_params = getattr(request, "path_params", {}) or {} 

50 

51 # 1. Pydantic Validation & Merging (Unified Sources) 

52 # This handles Body DTOs, Query params, and Path params that are part of a model. 

53 parsed_all = None 

54 try: 

55 parsed_all = await validate_and_merge_request( 

56 request, 

57 handler, 

58 path_params, 

59 ) 

60 except (ValueError, TypeError): 

61 # Re-raise for exception filters 

62 raise 

63 

64 # 2. Map parameters to handler signature 

65 for param_name, param in self.sig.parameters.items(): 

66 if param_name == "self": 

67 continue 

68 

69 annotation = self.hints.get(param_name, param.annotation) 

70 value = None 

71 

72 # Resolve ForwardRef strings if needed 

73 if isinstance(annotation, str): 

74 resolved = handler.__globals__.get(annotation) 

75 if resolved is not None: 

76 annotation = resolved 

77 

78 # A. Check parsed validation result (Path/Query/Body DTOs) 

79 if parsed_all and hasattr(parsed_all, param_name): 

80 value = getattr(parsed_all, param_name) 

81 

82 # B. Special Type: Request or ExecutionContextProtocol 

83 elif self._is_request_type(annotation, param_name): 

84 value = request 

85 

86 elif annotation in (WebExecutionContext, ExecutionContextProtocol): 

87 value = context 

88 

89 # C. Dependency Injection 

90 elif ( 

91 container 

92 and hasattr(container, "resolve") 

93 and annotation is not inspect.Parameter.empty 

94 ): 

95 # Only resolve if NOT already found in path/query (to allow overrides) 

96 if ( 

97 param_name not in path_params 

98 and param_name not in request.query_params 

99 ): 

100 try: 

101 value = await container.resolve(annotation) 

102 except Exception as e: # noqa: BLE001 — container.resolve may raise any DI error; wrap into typed DependencyResolutionError 

103 if param.default is inspect.Parameter.empty: 

104 raise DependencyResolutionError( 

105 param=param_name, 

106 service_type=annotation, 

107 cause=e, 

108 ) from e 

109 

110 # D. Manual Fallbacks (Direct extraction) 

111 if value is None: 

112 if param_name in path_params: 

113 value = path_params[param_name] 

114 elif param_name in request.query_params: 

115 value = request.query_params[param_name] 

116 elif param.default is not inspect.Parameter.empty: 

117 # If it's a decorator, get its default 

118 if hasattr(param.default, "_lexigram_param_info"): 

119 value = param.default._lexigram_param_info.get("default") 

120 else: 

121 value = param.default 

122 

123 if value is ...: 

124 value = None 

125 

126 # 3. Execute Pipes 

127 value = await self._run_pipes(param_name, value, param, annotation, context) 

128 kwargs[param_name] = value 

129 

130 return kwargs 

131 

132 async def _run_pipes( 

133 self, 

134 name: str, 

135 value: Any, 

136 param: inspect.Parameter, 

137 annotation: Any, 

138 context: WebExecutionContext, 

139 ) -> Any: 

140 """Execute pipes for a single parameter.""" 

141 from lexigram.web.protocols import ParamMetadata 

142 

143 # Collect pipes 

144 pipes = [] 

145 

146 # 1. Parameter-level pipes 

147 info = getattr(param.default, "_lexigram_param_info", {}) 

148 if info and "pipes" in info and info["pipes"]: 

149 pipes.extend(info["pipes"]) 

150 

151 # 2. Handler-level pipes (@use_pipes) 

152 handler_pipes = getattr(context.handler, "_lexigram_pipes", []) 

153 pipes.extend(handler_pipes) 

154 

155 if not pipes: 

156 return value 

157 

158 # Create metadata for pipes 

159 metadata = ParamMetadata( 

160 name=name, 

161 param_type=info.get("type", "unknown"), 

162 expected_type=annotation 

163 if annotation is not inspect.Parameter.empty 

164 else None, 

165 default=info.get("default") if info.get("default") is not ... else None, 

166 alias=info.get("alias"), 

167 ) 

168 

169 result = value 

170 for pipe in pipes: 

171 # If pipe is a class, instantiate it (simple DI could be added here later) 

172 pipe_instance = cast("Any", pipe()) if inspect.isclass(pipe) else pipe 

173 result = await pipe_instance.transform(result, metadata) 

174 

175 return result 

176 

177 def _is_request_type(self, annotation: Any, name: str) -> bool: 

178 """Helper to determine if a parameter represents the Request object.""" 

179 try: 

180 annotation_name = getattr(annotation, "__name__", None) 

181 except AttributeError: 

182 annotation_name = None 

183 

184 from starlette.requests import Request as StarletteRequest 

185 

186 return ( 

187 annotation is StarletteRequest 

188 or annotation_name == "Request" 

189 or str(annotation).startswith("Request") 

190 or name == "request" 

191 ) 

192 

193 

194__all__ = ["ParameterBinder"]