Coverage for faststream / _internal / utils / functions.py: 86%

1import asyncio (empty)

2from collections.abc import AsyncIterator, Awaitable, Callable (empty)

3from concurrent.futures import Executor (empty)

4from contextlib import asynccontextmanager (empty)

5from functools import partial, wraps (empty)

6from typing import ( (empty)

7 TYPE_CHECKING, 

8 Any, 

9 Optional, 

10 TypeVar, 

11 cast, 

12 overload, 

13) 

14 

15from fast_depends.utils import ( (empty)

16 is_coroutine_callable, 

17 run_async as call_or_await, 

18 run_in_threadpool, 

19) 

20from typing_extensions import ParamSpec, Self (empty)

21 

22from faststream._internal.basic_types import F_Return, F_Spec (empty)

23 

24if TYPE_CHECKING: (empty)

25 from types import TracebackType 

26 

27__all__ = ( (empty)

28 "call_or_await", 

29 "fake_context", 

30 "to_async", 

31) 

32 

33P = ParamSpec("P") (empty)

34T = TypeVar("T") (empty)

35 

36 

37@overload (empty)

38def to_async( (empty)

39 func: Callable[F_Spec, Awaitable[F_Return]], (empty)

40) -> Callable[F_Spec, Awaitable[F_Return]]: ... (empty)

41 

42 

43@overload (empty)

44def to_async( (empty)

45 func: Callable[F_Spec, F_Return], (empty)

46) -> Callable[F_Spec, Awaitable[F_Return]]: ... (empty)

47 

48 

49def to_async( (empty)

50 func: Callable[F_Spec, F_Return] | Callable[F_Spec, Awaitable[F_Return]], 

51) -> Callable[F_Spec, Awaitable[F_Return]]: 

52 """Converts a synchronous function to an asynchronous function.""" 

53 if is_coroutine_callable(func): (empty)

54 return cast("Callable[F_Spec, Awaitable[F_Return]]", func) (empty)

55 

56 func = cast("Callable[F_Spec, F_Return]", func) (empty)

57 

58 @wraps(func) (empty)

59 async def to_async_wrapper(*args: F_Spec.args, **kwargs: F_Spec.kwargs) -> F_Return: (empty)

60 """Wraps a function to make it asynchronous.""" 

61 return await run_in_threadpool(func, *args, **kwargs) (empty)

62 

63 return to_async_wrapper (empty)

64 

65 

66@asynccontextmanager (empty)

67async def fake_context(*args: Any, **kwargs: Any) -> AsyncIterator[None]: (empty)

68 yield None (empty)

69 

70 

71class FakeContext: (empty)

72 def __init__(self, *args: Any, **kwargs: Any) -> None: (empty)

73 pass (empty)

74 

75 def __enter__(self) -> Self: (empty)

76 return self (empty)

77 

78 def __exit__( (empty)

79 self, 

80 exc_type: type[BaseException] | None = None, 

81 exc_val: BaseException | None = None, 

82 exc_tb: Optional["TracebackType"] = None, 

83 ) -> None: 

84 if exc_val: (empty)

85 raise exc_val (empty)

86 

87 async def __aenter__(self) -> Self: (empty)

88 return self 

89 

90 async def __aexit__( (empty)

91 self, 

92 exc_type: type[BaseException] | None = None, 

93 exc_val: BaseException | None = None, 

94 exc_tb: Optional["TracebackType"] = None, 

95 ) -> None: 

96 if exc_val: 

97 raise exc_val 

98 

99 

100async def return_input(x: Any) -> Any: (empty)

101 return x (empty)

102 

103 

104async def run_in_executor( (empty)

105 executor: Executor | None, 

106 func: Callable[P, T], 

107 *args: P.args, 

108 **kwargs: P.kwargs, 

109) -> T: 

110 loop = asyncio.get_running_loop() (empty)

111 return await loop.run_in_executor(executor, partial(func, *args, **kwargs)) (empty)