Skip to content

to_async

faststream.utils.functions.to_async #

to_async(func: Union[Callable[F_Spec, F_Return], Callable[F_Spec, Awaitable[F_Return]]]) -> Callable[F_Spec, Awaitable[F_Return]]

Converts a synchronous function to an asynchronous function.

PARAMETER DESCRIPTION
func

The synchronous function to be converted.

TYPE: Union[Callable[F_Spec, F_Return], Callable[F_Spec, Awaitable[F_Return]]]

RETURNS DESCRIPTION
Callable[F_Spec, Awaitable[F_Return]]

The asynchronous version of the input function.
Source code in faststream/utils/functions.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def to_async(
    func: Union[
        Callable[F_Spec, F_Return],
        Callable[F_Spec, Awaitable[F_Return]],
    ],
) -> Callable[F_Spec, Awaitable[F_Return]]:
    """Converts a synchronous function to an asynchronous function.

    Args:
        func: The synchronous function to be converted.

    Returns:
        The asynchronous version of the input function.
    """

    @wraps(func)
    async def to_async_wrapper(*args: F_Spec.args, **kwargs: F_Spec.kwargs) -> F_Return:
        """Wraps a function to make it asynchronous.

        Args:
            func: The function to be wrapped
            args: Positional arguments to be passed to the function
            kwargs: Keyword arguments to be passed to the function

        Returns:
            The result of the wrapped function

        Raises:
            Any exceptions raised by the wrapped function
        """
        return await call_or_await(func, *args, **kwargs)

    return to_async_wrapper