Coverage for fast_depends/utils.py: 100%
91 statements
« prev ^ index » next coverage.py v7.17.3, created at 2026-11-13 05:19 +0000
1import functools 1ghijklabcdefnopqrsmt
2import inspect 1ghijklabcdefnopqrsmt
3import sys 1ghijklabcdefnopqrsmt
4from collections.abc import AsyncGenerator, AsyncIterable, Awaitable, Callable 1ghijklabcdefnopqrsmt
5from contextlib import ( 1ghijklabcdefnopqrsmt
6 AbstractContextManager,
7 AsyncExitStack,
8 ExitStack,
9 asynccontextmanager,
10 contextmanager,
11)
12from typing import ( 1ghijklabcdefnopqrsmt
13 TYPE_CHECKING,
14 Annotated,
15 Any,
16 ForwardRef,
17 TypeVar,
18 cast,
19 get_args,
20 get_origin,
21 overload,
22)
24if sys.version_info >= (3, 12): 1ghijklabcdefnopqrsmt
25 # to support PydanticV1 we should switch it expicitly
26 from typing import TypeAliasType 1abcdefnopqrsmt
27else:
28 from typing_extensions import TypeAliasType 1ghijkl
30import anyio 1ghijklabcdefnopqrsmt
31from typing_extensions import ParamSpec 1ghijklabcdefnopqrsmt
33from fast_depends._compat import evaluate_forwardref 1ghijklabcdefnopqrsmt
35if TYPE_CHECKING: 1ghijklabcdefnopqrsmt
36 from types import FrameType
38P = ParamSpec("P") 1ghijklabcdefnopqrsmt
39T = TypeVar("T") 1ghijklabcdefnopqrsmt
42@overload 1ghijklabcdefnopqrsmt
43async def run_async( 1ghijklabcdefnopqrsmt
44 func: Callable[P, Awaitable[T]], 1ghijklabcdefm
45 *args: P.args, 1ghijklabcdefm
46 **kwargs: P.kwargs, 1ghijklabcdefm
47) -> T: ... 1ghijklabcdefm
50@overload 1ghijklabcdefnopqrsmt
51async def run_async( 1ghijklabcdefnopqrsmt
52 func: Callable[P, T], 1ghijklabcdefm
53 *args: P.args, 1ghijklabcdefm
54 **kwargs: P.kwargs, 1ghijklabcdefm
55) -> T: ... 1ghijklabcdefm
58async def run_async( 1ghijklabcdefnopqrsmt
59 func: Callable[P, T] | Callable[P, Awaitable[T]],
60 *args: P.args,
61 **kwargs: P.kwargs,
62) -> T:
63 if is_coroutine_callable(func): 1ghijklabcdefnopqrsmt
64 return await cast(Callable[P, Awaitable[T]], func)(*args, **kwargs) 1ghijklabcdefnopqrsmt
65 else:
66 return await run_in_threadpool(cast(Callable[P, T], func), *args, **kwargs) 1ghijklabcdefnopqrsmt
69async def run_in_threadpool( 1ghijklabcdefnopqrsmt
70 func: Callable[P, T],
71 *args: P.args,
72 **kwargs: P.kwargs,
73) -> T:
74 if kwargs: 1ghijklabcdefnopqrsmt
75 func = functools.partial(func, **kwargs) 1ghijklabcdefnopqrsmt
76 return await anyio.to_thread.run_sync(func, *args) 1ghijklabcdefnopqrsmt
79async def solve_generator_async( 1ghijklabcdefnopqrsmt
80 *sub_args: Any,
81 call: Callable[..., Any],
82 stack: AsyncExitStack,
83 **sub_values: Any,
84) -> Any:
85 if is_gen_callable(call): 1ghijklabcdefnopqrsmt
86 cm = contextmanager_in_threadpool(contextmanager(call)(**sub_values)) 1ghijklabcdefnopqrsmt
87 elif is_async_gen_callable(call): # pragma: no branch 1ghijklabcdefnopqrsmt
88 cm = asynccontextmanager(call)(*sub_args, **sub_values) 1ghijklabcdefnopqrsmt
89 return await stack.enter_async_context(cm) 1ghijklabcdefnopqrsmt
92def solve_generator_sync( 1ghijklabcdefnopqrsmt
93 *sub_args: Any,
94 call: Callable[..., Any],
95 stack: ExitStack,
96 **sub_values: Any,
97) -> Any:
98 cm = contextmanager(call)(*sub_args, **sub_values) 1ghijklabcdefnopqrsmt
99 return stack.enter_context(cm) 1ghijklabcdefnopqrsmt
102def get_typed_signature(call: Callable[..., Any]) -> tuple[inspect.Signature, Any]: 1ghijklabcdefnopqrsmt
103 signature = inspect.signature(call) 1ghijklabcdefnopqrsmt
105 locals = collect_outer_stack_locals() 1ghijklabcdefnopqrsmt
107 # We unwrap call to get the original unwrapped function
108 call = inspect.unwrap(call) 1ghijklabcdefnopqrsmt
110 type_params = getattr(call, "__type_params__", ()) or None 1ghijklabcdefnopqrsmt
112 globalns = getattr(call, "__globals__", {}) 1ghijklabcdefnopqrsmt
113 typed_params = [ 1ghijklabcdefnopqrsmt
114 inspect.Parameter(
115 name=param.name,
116 kind=param.kind,
117 default=param.default,
118 annotation=get_typed_annotation(
119 param.annotation,
120 globalns,
121 locals,
122 type_params=type_params,
123 ),
124 )
125 for param in signature.parameters.values()
126 ]
128 return inspect.Signature(typed_params), get_typed_annotation( 1ghijklabcdefnopqrsmt
129 signature.return_annotation,
130 globalns,
131 locals,
132 type_params=type_params,
133 )
136def collect_outer_stack_locals() -> dict[str, Any]: 1ghijklabcdefnopqrsmt
137 frame = inspect.currentframe() 1ghijklabcdefnopqrsmt
139 frames: list[FrameType] = [] 1ghijklabcdefnopqrsmt
140 while frame is not None: 1ghijklabcdefnopqrsmt
141 if "fast_depends" not in frame.f_code.co_filename: 1ghijklabcdefnopqrsmt
142 frames.append(frame) 1ghijklabcdefnopqrsmt
143 frame = frame.f_back 1ghijklabcdefnopqrsmt
145 locals = {} 1ghijklabcdefnopqrsmt
146 for f in frames[::-1]: 1ghijklabcdefnopqrsmt
147 locals.update(f.f_locals) 1ghijklabcdefnopqrsmt
149 return locals 1ghijklabcdefnopqrsmt
152def get_typed_annotation( 1ghijklabcdefnopqrsmt
153 annotation: Any,
154 globalns: dict[str, Any],
155 locals: dict[str, Any],
156 type_params: tuple[Any, ...] | None = None,
157) -> Any:
158 if isinstance(annotation, TypeAliasType): 1ghijklabcdefnopqrsmt
159 annotation = annotation.__value__ 1abcdefnopqrsmt
161 if isinstance(annotation, str): 1ghijklabcdefnopqrsmt
162 annotation = ForwardRef(annotation) 1ghijklabcdefnopqrs
164 if isinstance(annotation, ForwardRef): 1ghijklabcdefnopqrsmt
165 annotation = evaluate_forwardref( 1ghijklabcdefnopqrsmt
166 annotation, globalns, locals, type_params=type_params
167 )
169 if get_origin(annotation) is Annotated and (args := get_args(annotation)): 1ghijklabcdefnopqrsmt
170 solved_args = [ 1ghijklabcdefnopqrsmt
171 get_typed_annotation(x, globalns, locals, type_params=type_params)
172 for x in args
173 ]
175 annotation.__origin__, annotation.__metadata__ = ( 1ghijklabcdefnopqrsmt
176 solved_args[0],
177 tuple(solved_args[1:]),
178 )
180 return annotation 1ghijklabcdefnopqrsmt
183@asynccontextmanager 1ghijklabcdefnopqrsmt
184async def contextmanager_in_threadpool( 1ghijklabcdefnopqrsmt
185 cm: AbstractContextManager[T],
186) -> AsyncGenerator[T, None]:
187 exit_limiter = anyio.CapacityLimiter(1) 1ghijklabcdefnopqrsmt
188 try: 1ghijklabcdefnopqrsmt
189 yield await run_in_threadpool(cm.__enter__) 1ghijklabcdefnopqrsmt
190 except Exception as e: 1ghijklabcdefnopqrsmt
191 ok = bool( 1ghijklabcdefnopqrsmt
192 await anyio.to_thread.run_sync(
193 cm.__exit__, type(e), e, None, limiter=exit_limiter
194 )
195 )
196 if not ok: # pragma: no branch 1ghijklabcdefnopqrsmt
197 raise e 1ghijklabcdefnopqrsmt
198 else:
199 await anyio.to_thread.run_sync( 1ghijklabcdefnopqrsmt
200 cm.__exit__, None, None, None, limiter=exit_limiter
201 )
204def is_gen_callable(call: Callable[..., Any]) -> bool: 1ghijklabcdefnopqrsmt
205 if inspect.isgeneratorfunction(call): 1ghijklabcdefnopqrsmt
206 return True 1ghijklabcdefnopqrsmt
207 dunder_call = getattr(call, "__call__", None) # noqa: B004 1ghijklabcdefnopqrsmt
208 return inspect.isgeneratorfunction(dunder_call) 1ghijklabcdefnopqrsmt
211def is_async_gen_callable(call: Callable[..., Any]) -> bool: 1ghijklabcdefnopqrsmt
212 if inspect.isasyncgenfunction(call): 1ghijklabcdefnopqrsmt
213 return True 1ghijklabcdefnopqrsmt
214 dunder_call = getattr(call, "__call__", None) # noqa: B004 1ghijklabcdefnopqrsmt
215 return inspect.isasyncgenfunction(dunder_call) 1ghijklabcdefnopqrsmt
218def is_coroutine_callable(call: Callable[..., Any]) -> bool: 1ghijklabcdefnopqrsmt
219 if inspect.isclass(call): 1ghijklabcdefnopqrsmt
220 return False 1ghijklabcdefnopqrsmt
222 if inspect.iscoroutinefunction(call): 1ghijklabcdefnopqrsmt
223 return True 1ghijklabcdefnopqrsmt
225 dunder_call = getattr(call, "__call__", None) # noqa: B004 1ghijklabcdefnopqrsmt
226 return inspect.iscoroutinefunction(dunder_call) 1ghijklabcdefnopqrsmt
229async def async_map( 1ghijklabcdefnopqrsmt
230 func: Callable[..., T], async_iterable: AsyncIterable[Any]
231) -> AsyncIterable[T]:
232 async for i in async_iterable: 1ghijklabcdefnopqrs
233 yield func(i) 1ghijklabcdefnopqrs