Coverage for fast_depends/use.py: 100%

91 statements  

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

1from collections.abc import AsyncIterator, Callable, Iterator, Sequence 1defghibjkclmopqrstan

2from contextlib import AsyncExitStack, ExitStack 1defghibjkclmopqrstan

3from functools import partial, wraps 1defghibjkclmopqrstan

4from typing import ( 1defghibjkclmopqrstan

5 TYPE_CHECKING, 

6 Any, 

7 Optional, 

8 Protocol, 

9 TypeVar, 

10 Union, 

11 cast, 

12 overload, 

13) 

14 

15from typing_extensions import ParamSpec 1defghibjkclmopqrstan

16 

17from fast_depends.core import CallModel, build_call_model 1defghibjkclmopqrstan

18from fast_depends.dependencies import Dependant, Provider 1defghibjkclmopqrstan

19from fast_depends.library.serializer import SerializerProto 1defghibjkclmopqrstan

20 

21SerializerCls: Optional["SerializerProto"] = None 1defghibjkclmopqrstan

22 

23try: 1defghibjkclmopqrstan

24 from fast_depends.pydantic import PydanticSerializer 1defghibjkclmopqrstan

25 

26 SerializerCls = PydanticSerializer() 1defghijklmpqrst

27except ImportError: 1bcoan

28 pass 1bcoan

29 

30if SerializerCls is None: 1defghibjkclmopqrstan

31 try: 1bcoan

32 from fast_depends.msgspec import MsgSpecSerializer 1bcoan

33 

34 SerializerCls = MsgSpecSerializer() 1bco

35 except ImportError: 1an

36 pass 1an

37 

38 

39P = ParamSpec("P") 1defghibjkclmopqrstan

40T = TypeVar("T") 1defghibjkclmopqrstan

41 

42if TYPE_CHECKING: 1defghibjkclmopqrstan

43 from fast_depends.library.serializer import SerializerProto 

44 

45 class InjectWrapper(Protocol[P, T]): 

46 def __call__( 

47 self, 

48 func: Callable[P, T], 

49 model: CallModel | None = None, 

50 ) -> Callable[P, T]: ... 

51 

52 

53global_provider = Provider() 1defghibjkclmopqrstan

54 

55 

56def Depends( 1defghibjkclmopqrstan

57 dependency: Callable[..., Any], 

58 *, 

59 use_cache: bool = True, 

60 cast: bool = True, 

61 cast_result: bool = False, 

62) -> Any: 

63 return Dependant( 1defghibjkclmopqrstan

64 dependency=dependency, 

65 use_cache=use_cache, 

66 cast=cast, 

67 cast_result=cast_result, 

68 ) 

69 

70 

71@overload 1defghibjkclmopqrstan

72def inject( 1defghibjkclmopqrstan

73 func: Callable[P, T], 1defghibjkclma

74 *, 

75 cast: bool = True, 1defghibjkclmopqrstan

76 cast_result: bool = True, 1defghibjkclmopqrstan

77 extra_dependencies: Sequence["Dependant"] = (), 1defghibjkclmopqrstan

78 dependency_provider: Optional["Provider"] = None, 1defghibjkclmopqrstan

79 wrap_model: Callable[["CallModel"], "CallModel"] = lambda x: x, 1defghibjkclmopqrstan

80 serializer_cls: Optional["SerializerProto"] = SerializerCls, 1defghibjkclmopqrstan

81 **call_extra: Any, 1defghibjkclma

82) -> Callable[P, T]: ... 1defghibjkclma

83 

84 

85@overload 1defghibjkclmopqrstan

86def inject( 1defghibjkclmopqrstan

87 func: None = None, 1defghibjkclma

88 *, 

89 cast: bool = True, 1defghibjkclmopqrstan

90 cast_result: bool = True, 1defghibjkclmopqrstan

91 extra_dependencies: Sequence["Dependant"] = (), 1defghibjkclmopqrstan

92 dependency_provider: Optional["Provider"] = None, 1defghibjkclmopqrstan

93 wrap_model: Callable[["CallModel"], "CallModel"] = lambda x: x, 1defghibjkclmopqrstan

94 serializer_cls: Optional["SerializerProto"] = SerializerCls, 1defghibjkclmopqrstan

95 **call_extra: Any, 1defghibjkclma

96) -> "InjectWrapper[..., Any]": ... 1defghibjkclma

97 

98 

99def inject( 1defghibjkclmopqrstan

100 func: Callable[P, T] | None = None, 

101 *, 

102 cast: bool = True, 

103 cast_result: bool = True, 

104 extra_dependencies: Sequence[Dependant] = (), 

105 dependency_provider: Optional["Provider"] = None, 

106 wrap_model: Callable[["CallModel"], "CallModel"] = lambda x: x, 

107 serializer_cls: Optional["SerializerProto"] = SerializerCls, 

108 **call_extra: Any, 

109) -> Union[Callable[P, T], "InjectWrapper[P, T]"]: 

110 if dependency_provider is None: 1defghibjkclmopqrstan

111 dependency_provider = global_provider 1defghibjkclmopqrstan

112 

113 if not cast: 1defghibjkclmopqrstan

114 serializer_cls = None 1defghibjkclmopqrstan

115 

116 decorator: InjectWrapper[P, T] = _wrap_inject( 1defghibjkclmopqrstan

117 dependency_provider=dependency_provider, 

118 wrap_model=wrap_model, 

119 extra_dependencies=extra_dependencies, 

120 serializer_cls=serializer_cls, 

121 cast_result=cast_result, 

122 **call_extra, 

123 ) 

124 

125 if func is None: 1defghibjkclmopqrstan

126 return decorator 1defghibjkclmopqrstan

127 return decorator(func) 1defghibjkclmopqrstan

128 

129 

130def _wrap_inject( 1defghibjkclmopqrstan

131 *, 

132 dependency_provider: "Provider", 

133 wrap_model: Callable[["CallModel"], "CallModel"], 

134 extra_dependencies: Sequence[Dependant], 

135 serializer_cls: Optional["SerializerProto"], 

136 cast_result: bool, 

137 **call_extra: Any, 

138) -> "InjectWrapper[P, T]": 

139 def func_wrapper( 1defghibjkclmopqrstan

140 func: Callable[P, T], 

141 model: Optional["CallModel"] = None, 

142 ) -> Callable[P, T]: 

143 if model is None: 1defghibjkclmopqrstan

144 real_model = wrap_model( 1defghibjkclmopqrstan

145 build_call_model( 

146 call=func, 

147 extra_dependencies=extra_dependencies, 

148 dependency_provider=dependency_provider, 

149 serializer_cls=serializer_cls, 

150 serialize_result=cast_result, 

151 ) 

152 ) 

153 else: 

154 real_model = model 1defghibjkclmopqrstan

155 

156 if real_model.is_async: 1defghibjkclmopqrstan

157 injected_wrapper: Callable[P, T] 

158 

159 if real_model.is_generator: 1defghibjkclmopqrstan

160 injected_wrapper = partial( # type: ignore[assignment] 1defghibjkclmopqrstan

161 solve_async_gen, 

162 real_model, 

163 ) 

164 

165 else: 

166 

167 async def injected_wrapper(*args: P.args, **kwargs: P.kwargs) -> T: # type: ignore[misc] 1defghibjkclmopqrstan

168 async with AsyncExitStack() as stack: 1defghibjkclmopqrstan

169 return await real_model.asolve( # type: ignore[no-any-return] 1defghibjkclmopqrstan

170 *args, 

171 stack=stack, 

172 cache_dependencies={}, 

173 nested=False, 

174 **(call_extra | kwargs), 

175 ) 

176 

177 raise AssertionError("unreachable") 

178 

179 else: 

180 if real_model.is_generator: 1defghibjkclmopqrstan

181 injected_wrapper = partial( # type: ignore[assignment] 1defghibjkclmopqrstan

182 solve_gen, 

183 real_model, 

184 ) 

185 

186 else: 

187 

188 def injected_wrapper(*args: P.args, **kwargs: P.kwargs) -> T: 1defghibjkclmopqrstan

189 with ExitStack() as stack: 1defghibjkclmopqrstan

190 return real_model.solve( # type: ignore[no-any-return] 1defghibjkclmopqrstan

191 *args, 

192 stack=stack, 

193 cache_dependencies={}, 

194 nested=False, 

195 **(call_extra | kwargs), 

196 ) 

197 

198 raise AssertionError("unreachable") 

199 

200 injected_wrapper._fastdepends_call_ = real_model.call # type: ignore[attr-defined] 1defghibjkclmopqrstan

201 return wraps(func)(injected_wrapper) 1defghibjkclmopqrstan

202 

203 return func_wrapper 1defghibjkclmopqrstan

204 

205 

206class solve_async_gen: 1defghibjkclmopqrstan

207 _iter: AsyncIterator[Any] | None = None 1defghibjkclmopqrstan

208 

209 def __init__( 1defghibjkclmopqrstan

210 self, 

211 model: "CallModel", 

212 *args: Any, 

213 **kwargs: Any, 

214 ): 

215 self.call = model 1defghibjkclmopqrstan

216 self.args = args 1defghibjkclmopqrstan

217 self.kwargs = kwargs 1defghibjkclmopqrstan

218 

219 def __aiter__(self) -> "solve_async_gen": 1defghibjkclmopqrstan

220 self.stack = AsyncExitStack() 1defghibjkclmopqrstan

221 return self 1defghibjkclmopqrstan

222 

223 async def __anext__(self) -> Any: 1defghibjkclmopqrstan

224 if self._iter is None: 1defghibjkclmopqrstan

225 stack = self.stack = AsyncExitStack() 1defghibjkclmopqrstan

226 await self.stack.__aenter__() 1defghibjkclmopqrstan

227 self._iter = cast( 1defghibjkclmopqrstan

228 AsyncIterator[Any], 

229 ( 

230 await self.call.asolve( 

231 *self.args, 

232 stack=stack, 

233 cache_dependencies={}, 

234 nested=False, 

235 **self.kwargs, 

236 ) 

237 ).__aiter__(), 

238 ) 

239 

240 try: 1defghibjkclmopqrstan

241 r = await self._iter.__anext__() 1defghibjkclmopqrstan

242 except StopAsyncIteration: 1defghibjkclmopqrstan

243 await self.stack.__aexit__(None, None, None) 1defghibjkclmopqrstan

244 raise 1defghibjkclmopqrstan

245 else: 

246 return r 1defghibjkclmopqrstan

247 

248 

249class solve_gen: 1defghibjkclmopqrstan

250 _iter: Iterator[Any] | None = None 1defghibjkclmopqrstan

251 

252 def __init__( 1defghibjkclmopqrstan

253 self, 

254 model: "CallModel", 

255 *args: Any, 

256 **kwargs: Any, 

257 ): 

258 self.call = model 1defghibjkclmopqrstan

259 self.args = args 1defghibjkclmopqrstan

260 self.kwargs = kwargs 1defghibjkclmopqrstan

261 

262 def __iter__(self) -> "solve_gen": 1defghibjkclmopqrstan

263 self.stack = ExitStack() 1defghibjkclmopqrstan

264 return self 1defghibjkclmopqrstan

265 

266 def __next__(self) -> Any: 1defghibjkclmopqrstan

267 if self._iter is None: 1defghibjkclmopqrstan

268 stack = self.stack = ExitStack() 1defghibjkclmopqrstan

269 self.stack.__enter__() 1defghibjkclmopqrstan

270 self._iter = cast( 1defghibjkclmopqrstan

271 Iterator[Any], 

272 iter( 

273 self.call.solve( 

274 *self.args, 

275 stack=stack, 

276 cache_dependencies={}, 

277 nested=False, 

278 **self.kwargs, 

279 ) 

280 ), 

281 ) 

282 

283 try: 1defghibjkclmopqrstan

284 r = next(self._iter) 1defghibjkclmopqrstan

285 except StopIteration: 1defghibjkclmopqrstan

286 self.stack.__exit__(None, None, None) 1defghibjkclmopqrstan

287 raise 1defghibjkclmopqrstan

288 else: 

289 return r 1defghibjkclmopqrstan