Coverage for fast_depends/core/builder.py: 100%

115 statements  

« prev     ^ index     » next       coverage.py v7.17.3, created at 2026-11-13 05:19 +0000

1import inspect 1klabmcpndqoerfghijst

2from collections.abc import Callable, Sequence 1klabmcpndqoerfghijst

3from copy import deepcopy 1klabmcpndqoerfghijst

4from typing import ( 1klabmcpndqoerfghijst

5 TYPE_CHECKING, 

6 Annotated, 

7 Any, 

8 Optional, 

9 TypeVar, 

10 get_args, 

11 get_origin, 

12) 

13 

14from typing_extensions import ( 1klabmcpndqoerfghijst

15 ParamSpec, 

16) 

17 

18from fast_depends.dependencies.model import Dependant 1klabmcpndqoerfghijst

19from fast_depends.library import CustomField 1klabmcpndqoerfghijst

20from fast_depends.library.serializer import OptionItem, Serializer, SerializerProto 1klabmcpndqoerfghijst

21from fast_depends.utils import ( 1klabmcpndqoerfghijst

22 get_typed_signature, 

23 is_async_gen_callable, 

24 is_coroutine_callable, 

25 is_gen_callable, 

26) 

27 

28from .model import CallModel 1klabmcpndqoerfghijst

29 

30if TYPE_CHECKING: 1klabmcpndqoerfghijst

31 from fast_depends.dependencies.provider import Key, Provider 

32 

33 

34CUSTOM_ANNOTATIONS = ( 1klabmcpndqoerfghijst

35 Dependant, 

36 CustomField, 

37) 

38 

39 

40P = ParamSpec("P") 1klabmcpndqoerfghijst

41T = TypeVar("T") 1klabmcpndqoerfghijst

42 

43 

44def build_call_model( 1klabmcpndqoerfghijst

45 call: Callable[..., Any], 

46 *, 

47 dependency_provider: "Provider", 

48 use_cache: bool = True, 

49 is_sync: bool | None = None, 

50 extra_dependencies: Sequence[Dependant] = (), 

51 serializer_cls: Optional["SerializerProto"] = None, 

52 serialize_result: bool = True, 

53) -> CallModel: 

54 if hasattr(call, "_fastdepends_call_") and not hasattr(call, "_mock_name"): 1klabmcpndqoerfghijst

55 call = call._fastdepends_call_ 1klabmcpndqoerfghijst

56 

57 name = getattr(inspect.unwrap(call), "__name__", type(call).__name__) 1klabmcpndqoerfghijst

58 

59 is_call_async = is_coroutine_callable(call) or is_async_gen_callable(call) 1klabmcpndqoerfghijst

60 if is_sync is None: 1klabmcpndqoerfghijst

61 is_sync = not is_call_async 1klabmcpndqoerfghijst

62 else: 

63 assert not (is_sync and is_call_async), ( 1klabmcpndqoerfghijst

64 f"You cannot use async dependency `{name}` at sync main" 

65 ) 

66 

67 typed_params, return_annotation = get_typed_signature(call) 1klabmcpndqoerfghijst

68 if (is_call_generator := is_gen_callable(call) or is_async_gen_callable(call)) and ( 1klabmcpndqoerfghijst

69 return_args := get_args(return_annotation) 

70 ): 

71 return_annotation = return_args[0] 1klabmcpndqoerfghijst

72 

73 if not serialize_result: 1klabmcpndqoerfghijst

74 return_annotation = inspect.Parameter.empty 1klabmcpndqoerfghijst

75 

76 class_fields: list[OptionItem] = [] 1klabmcpndqoerfghijst

77 dependencies: dict[str, Key] = {} 1klabmcpndqoerfghijst

78 custom_fields: dict[str, CustomField] = {} 1klabmcpndqoerfghijst

79 positional_args: list[str] = [] 1klabmcpndqoerfghijst

80 keyword_args: list[str] = [] 1klabmcpndqoerfghijst

81 args_name: str | None = None 1klabmcpndqoerfghijst

82 kwargs_name: str | None = None 1klabmcpndqoerfghijst

83 

84 for param_name, param in typed_params.parameters.items(): 1klabmcpndqoerfghijst

85 dep: Dependant | None = None 1klabmcpndqoerfghijst

86 custom: CustomField | None = None 1klabmcpndqoerfghijst

87 

88 if param.annotation is inspect.Parameter.empty: 1klabmcpndqoerfghijst

89 annotation = Any 1klabmcpndqoerfghijst

90 

91 elif get_origin(param.annotation) is Annotated: 1klabmcpndqoerfghijst

92 annotated_args = get_args(param.annotation) 1klabmcpndqoerfghijst

93 type_annotation = annotated_args[0] 1klabmcpndqoerfghijst

94 

95 custom_annotations = [] 1klabmcpndqoerfghijst

96 regular_annotations = [] 1klabmcpndqoerfghijst

97 for arg in annotated_args[1:]: 1klabmcpndqoerfghijst

98 if isinstance(arg, CUSTOM_ANNOTATIONS): 1klabmcpndqoerfghijst

99 custom_annotations.append(arg) 1klabmcpndqoerfghijst

100 else: 

101 regular_annotations.append(arg) 1klabmcndoefghij

102 

103 assert len(custom_annotations) <= 1, ( 1klabmcpndqoerfghijst

104 f"Cannot specify multiple `Annotated` Custom arguments for `{param_name}`!" 

105 ) 

106 

107 next_custom = next(iter(custom_annotations), None) 1klabmcpndqoerfghijst

108 if next_custom is not None: 1klabmcpndqoerfghijst

109 if isinstance(next_custom, Dependant): 1klabmcpndqoerfghijst

110 dep = next_custom 1klabmcpndqoerfghijst

111 elif isinstance(next_custom, CustomField): 1klabmcpndqoerfghijst

112 custom = deepcopy(next_custom) 1klabmcpndqoerfghijst

113 else: # pragma: no cover 

114 raise AssertionError("unreachable") 

115 

116 if regular_annotations: 1klabmcpndqoerfghijst

117 annotation = param.annotation 1abcdefghij

118 else: 

119 annotation = type_annotation 1klabmcpndqoerfghijst

120 else: 

121 annotation = param.annotation 1klabmcndoefghij

122 else: 

123 annotation = param.annotation 1klabmcpndqoerfghijst

124 

125 default: Any 

126 if param.kind is inspect.Parameter.VAR_POSITIONAL: 1klabmcpndqoerfghijst

127 default = () 1klabmcpndqoerfghijst

128 elif param.kind is inspect.Parameter.VAR_KEYWORD: 1klabmcpndqoerfghijst

129 default = {} 1klabmcpndqoerfghijst

130 else: 

131 default = param.default 1klabmcpndqoerfghijst

132 

133 if isinstance(default, Dependant): 1klabmcpndqoerfghijst

134 assert not dep, "You can not use `Depends` with `Annotated` and default both" 1klabmcpndqoerfghijst

135 dep, default = default, Ellipsis 1klabmcpndqoerfghijst

136 

137 elif isinstance(default, CustomField): 1klabmcpndqoerfghijst

138 assert not custom, ( 1klabmcpndqoerfghijst

139 "You can not use `CustomField` with `Annotated` and default both" 

140 ) 

141 custom, default = default, Ellipsis 1klabmcpndqoerfghijst

142 

143 elif not dep and not custom: 1klabmcpndqoerfghijst

144 class_fields.append( 1klabmcpndqoerfghijst

145 OptionItem( 

146 field_name=param_name, 

147 field_type=annotation, 

148 default_value=Ellipsis 

149 if default is inspect.Parameter.empty 

150 else default, 

151 kind=param.kind, 

152 ) 

153 ) 

154 

155 if dep: 1klabmcpndqoerfghijst

156 dependency = build_call_model( 1klabmcpndqoerfghijst

157 dep.dependency, 

158 dependency_provider=dependency_provider, 

159 use_cache=dep.use_cache, 

160 is_sync=is_sync, 

161 serializer_cls=serializer_cls, 

162 serialize_result=dep.cast_result, 

163 ) 

164 

165 key = dependency_provider.add_dependant(dependency) 1klabmcpndqoerfghijst

166 

167 _rebuild_override_model( 1klabmcpndqoerfghijst

168 dependency_provider=dependency_provider, 

169 dependency=dependency, 

170 key=key, 

171 serializer_cls=serializer_cls, 

172 ) 

173 

174 overrided_dependency = dependency_provider.get_dependant(key) 1klabmcpndqoerfghijst

175 

176 assert not (is_sync and is_coroutine_callable(overrided_dependency.call)), ( 1klabmcpndqoerfghijst

177 f"You cannot use async dependency `{overrided_dependency.call_name}` at sync main" 

178 ) 

179 

180 dependencies[param_name] = key 1klabmcpndqoerfghijst

181 

182 if not dep.cast: 1klabmcpndqoerfghijst

183 annotation = Any 1klabmcpndqoerfghij

184 

185 class_fields.append( 1klabmcpndqoerfghijst

186 OptionItem( 

187 field_name=param_name, 

188 field_type=annotation, 

189 source=dep, 

190 kind=param.kind, 

191 ) 

192 ) 

193 

194 keyword_args.append(param_name) 1klabmcpndqoerfghijst

195 

196 elif custom: 1klabmcpndqoerfghijst

197 assert not (is_sync and is_coroutine_callable(custom.use)), ( 1klabmcpndqoerfghijst

198 f"You cannot use async custom field `{type(custom).__name__}` at sync `{name}`" 

199 ) 

200 

201 custom.set_param_name(param_name) 1klabmcpndqoerfghijst

202 custom_fields[param_name] = custom 1klabmcpndqoerfghijst

203 

204 if not custom.cast: 1klabmcpndqoerfghijst

205 annotation = Any 1klabmcpndqoerfghijst

206 

207 if custom.required: 1klabmcpndqoerfghijst

208 class_fields.append( 1klabmcpndqoerfghijst

209 OptionItem( 

210 field_name=param_name, 

211 field_type=annotation, 

212 default_value=Ellipsis 

213 if default is inspect.Parameter.empty 

214 else default, 

215 source=custom, 

216 kind=param.kind, 

217 ) 

218 ) 

219 

220 else: 

221 class_fields.append( 1klabmcpndqoerfghij

222 OptionItem( 

223 field_name=param_name, 

224 field_type=Optional[annotation], # noqa: UP045 

225 default_value=None 

226 if (default is inspect.Parameter.empty or default is Ellipsis) 

227 else default, 

228 source=custom, 

229 kind=param.kind, 

230 ) 

231 ) 

232 

233 keyword_args.append(param_name) 1klabmcpndqoerfghijst

234 

235 else: 

236 if param.kind is param.KEYWORD_ONLY: 1klabmcpndqoerfghijst

237 keyword_args.append(param_name) 1klabmcpndqoerfghijst

238 elif param.kind is param.VAR_KEYWORD: 1klabmcpndqoerfghijst

239 kwargs_name = param_name 1klabmcpndqoerfghijst

240 elif param.kind is param.VAR_POSITIONAL: 1klabmcpndqoerfghijst

241 args_name = param_name 1klabmcpndqoerfghijst

242 else: 

243 positional_args.append(param_name) 1klabmcpndqoerfghijst

244 

245 serializer: Serializer | None = None 1klabmcpndqoerfghijst

246 if serializer_cls is not None: 1klabmcpndqoerfghijst

247 serializer = serializer_cls( 1klabmcpndqoerfghijst

248 name=name, 

249 options=class_fields, 

250 response_type=return_annotation, 

251 ) 

252 

253 solved_extra_dependencies: list[Key] = [] 1klabmcpndqoerfghijst

254 for dep in extra_dependencies: 1klabmcpndqoerfghijst

255 dependency = build_call_model( 1klabmcpndqoerfghijst

256 dep.dependency, 

257 dependency_provider=dependency_provider, 

258 use_cache=dep.use_cache, 

259 is_sync=is_sync, 

260 serializer_cls=serializer_cls, 

261 ) 

262 

263 key = dependency_provider.add_dependant(dependency) 1klabmcpndqoerfghijst

264 

265 overrided_dependency = dependency_provider.get_dependant(key) 1klabmcpndqoerfghijst

266 

267 assert not (is_sync and is_coroutine_callable(overrided_dependency.call)), ( 1klabmcpndqoerfghijst

268 f"You cannot use async dependency `{overrided_dependency.call_name}` at sync main" 

269 ) 

270 

271 solved_extra_dependencies.append(key) 1klabmcpndqoerfghijst

272 

273 return CallModel( 1klabmcpndqoerfghijst

274 call=call, 

275 serializer=serializer, 

276 params=tuple( 

277 i 

278 for i in class_fields 

279 if (i.field_name not in dependencies and i.field_name not in custom_fields) 

280 ), 

281 use_cache=use_cache, 

282 is_async=is_call_async, 

283 is_generator=is_call_generator, 

284 dependencies=dependencies, 

285 custom_fields=custom_fields, 

286 positional_args=positional_args, 

287 keyword_args=keyword_args, 

288 args_name=args_name, 

289 kwargs_name=kwargs_name, 

290 extra_dependencies=solved_extra_dependencies, 

291 dependency_provider=dependency_provider, 

292 serializer_cls=serializer_cls, 

293 ) 

294 

295 

296def _rebuild_override_model( 1klabmcpndqoerfghijst

297 dependency_provider: "Provider", 

298 dependency: CallModel, 

299 key: "Key", 

300 serializer_cls: Optional["SerializerProto"], 

301) -> None: 

302 """Rebuild override model in case of a different serializer class""" 

303 override_model = dependency_provider.overrides.get(key) 1klabmcpndqoerfghijst

304 if override_model is not None and override_model.serializer_cls != serializer_cls: 1klabmcpndqoerfghijst

305 dependency_provider.override(dependency.call, override_model.call) 1klabmcpndqoerfghij