Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/starlette/concurrency.py: 44%

1from __future__ import annotations 

2 

3import functools 

4import warnings 

5from collections.abc import AsyncIterator, Callable, Coroutine, Iterable, Iterator 

6from typing import ParamSpec, TypeVar 

7 

8import anyio.to_thread 

9 

10from starlette.exceptions import StarletteDeprecationWarning 

11 

12P = ParamSpec("P") 

13T = TypeVar("T") 

14 

15 

16async def run_until_first_complete(*args: tuple[Callable, dict]) -> None: # type: ignore[type-arg] 

17 warnings.warn( 

18 "run_until_first_complete is deprecated and will be removed in a future version.", 

19 StarletteDeprecationWarning, 

20 ) 

21 

22 async with anyio.create_task_group() as task_group: 

23 

24 async def run(func: Callable[[], Coroutine]) -> None: # type: ignore[type-arg] 

25 await func() 

26 task_group.cancel_scope.cancel() 

27 

28 for func, kwargs in args: 

29 task_group.start_soon(run, functools.partial(func, **kwargs)) 

30 

31 

32async def run_in_threadpool(func: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> T: 

33 func = functools.partial(func, *args, **kwargs) 

34 return await anyio.to_thread.run_sync(func) 

35 

36 

37class _StopIteration(Exception): 

38 pass 

39 

40 

41def _next(iterator: Iterator[T]) -> T: 

42 # We can't raise `StopIteration` from within the threadpool iterator 

43 # and catch it outside that context, so we coerce them into a different 

44 # exception type. 

45 try: 

46 return next(iterator) 

47 except StopIteration: 

48 raise _StopIteration 

49 

50 

51async def iterate_in_threadpool( 

52 iterator: Iterable[T], 

53) -> AsyncIterator[T]: 

54 as_iterator = iter(iterator) 

55 while True: 

56 try: 

57 yield await anyio.to_thread.run_sync(_next, as_iterator) 

58 except _StopIteration: 

59 break