Skip to content

Convert

convert

async_to_sync 

async_to_sync(
    func_or_awaitable: Callable[_P, Awaitable[_T]],
) -> Callable[_P, _T]

async_to_sync(
    func_or_awaitable: Awaitable[_T],
) -> Callable[[], _T]

async_to_sync(
    func_or_awaitable: (
        Callable[..., Awaitable[_T]] | Awaitable[_T]
    ),
) -> Callable[..., _T]

async_to_sync(
    func_or_awaitable: (
        Callable[_P, Awaitable[_T]] | Awaitable[_T]
    ),
) -> Callable[_P, _T] | Callable[[], _T]

Convert an awaitable function or awaitable object to a synchronous function.

If used within an asynchronous context, attempts to use the same backend. Defaults to asyncio.

Parameters:

Name Type Description Default

func_or_awaitable

Callable[_P, Awaitable[_T]] | Awaitable[_T]

An awaitable function or awaitable object.

 required

Returns:

Type Description
Callable[_P, _T] | Callable[[], _T]

A synchronous function.
Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import asyncio
import time

import anyio
import sniffio

from async_wrapper import async_to_sync


@async_to_sync
async def test(x: int) -> int:
    backend = sniffio.current_async_library()
    if backend == "asyncio":
        loop = asyncio.get_running_loop()
        print(backend, loop)
    else:
        print(backend)
    await anyio.sleep(1)
    return x


def main() -> None:
    start = time.perf_counter()
    result = test(1)
    end = time.perf_counter()
    assert result == 1
    assert end - start < 1.1


async def async_main() -> None:
    start = time.perf_counter()
    result = test(1)
    end = time.perf_counter()
    assert result == 1
    assert end - start < 1.1


if __name__ == "__main__":
    main()
    anyio.run(
        async_main,
        backend="asyncio",
        backend_options={"use_uvloop": True},
    )
    anyio.run(
        async_main,
        backend="asyncio",
        backend_options={"use_uvloop": True},
    )
    anyio.run(async_main, backend="trio")
Source code in src/async_wrapper/convert/_sync/main.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def async_to_sync(
    func_or_awaitable: Callable[_P, Awaitable[_T]] | Awaitable[_T],
) -> Callable[_P, _T] | Callable[[], _T]:
    """
    Convert an awaitable function or awaitable object to a synchronous function.

    If used within an asynchronous context, attempts to use the same backend.
    Defaults to asyncio.

    Args:
        func_or_awaitable: An awaitable function or awaitable object.

    Returns:
        A synchronous function.

    Example:
        ```python
        import asyncio
        import time

        import anyio
        import sniffio

        from async_wrapper import async_to_sync


        @async_to_sync
        async def test(x: int) -> int:
            backend = sniffio.current_async_library()
            if backend == "asyncio":
                loop = asyncio.get_running_loop()
                print(backend, loop)
            else:
                print(backend)
            await anyio.sleep(1)
            return x


        def main() -> None:
            start = time.perf_counter()
            result = test(1)
            end = time.perf_counter()
            assert result == 1
            assert end - start < 1.1


        async def async_main() -> None:
            start = time.perf_counter()
            result = test(1)
            end = time.perf_counter()
            assert result == 1
            assert end - start < 1.1


        if __name__ == "__main__":
            main()
            anyio.run(
                async_main,
                backend="asyncio",
                backend_options={"use_uvloop": True},
            )
            anyio.run(
                async_main,
                backend="asyncio",
                backend_options={"use_uvloop": True},
            )
            anyio.run(async_main, backend="trio")
        ```
    """
    if callable(func_or_awaitable):
        from async_wrapper.convert._async import Async  # noqa: PLC0415

        if isinstance(func_or_awaitable, Async):
            return func_or_awaitable._func  # noqa: SLF001
        return Sync(func_or_awaitable)

    if has_sqlalchemy:
        result = run_sa_greenlet(func_or_awaitable)
        if not check_is_unset(result):
            return result  # pyright: ignore[reportReturnType]

    awaitable_func = _awaitable_to_function(func_or_awaitable)
    return _async_func_to_sync(awaitable_func)

sync_to_async 

sync_to_async(
    func: Callable[_P, _T],
) -> Callable[_P, Awaitable[_T]]

Convert a synchronous function to an asynchronous function.

Parameters:

Name Type Description Default

func

Callable[_P, _T]

The synchronous function to be converted.

 required

Returns:

Type Description
Callable[_P, Awaitable[_T]]

An asynchronous function
Callable[_P, Awaitable[_T]]

that behaves equivalently to the input synchronous function.

Examples:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import time

import anyio

from async_wrapper import sync_to_async


@sync_to_async
def test(x: int) -> int:
    print(f"[{x}] test: start")
    time.sleep(1)
    print(f"[{x}] test: end")
    return x


async def main() -> None:
    start = time.perf_counter()
    async with anyio.create_task_group() as task_group:
        for i in range(4):
            task_group.start_soon(test, i)
    end = time.perf_counter()
    assert end - start < 1.1


if __name__ == "__main__":
    anyio.run(main)
Source code in src/async_wrapper/convert/_async.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def sync_to_async(func: Callable[_P, _T]) -> Callable[_P, Awaitable[_T]]:
    """
    Convert a synchronous function to an asynchronous function.

    Args:
        func: The synchronous function to be converted.

    Returns:
        An asynchronous function
        that behaves equivalently to the input synchronous function.

    Examples:
        ```python
        import time

        import anyio

        from async_wrapper import sync_to_async


        @sync_to_async
        def test(x: int) -> int:
            print(f"[{x}] test: start")
            time.sleep(1)
            print(f"[{x}] test: end")
            return x


        async def main() -> None:
            start = time.perf_counter()
            async with anyio.create_task_group() as task_group:
                for i in range(4):
                    task_group.start_soon(test, i)
            end = time.perf_counter()
            assert end - start < 1.1


        if __name__ == "__main__":
            anyio.run(main)
        ```
    """
    from async_wrapper.convert._sync.main import Sync  # noqa: PLC0415

    if isinstance(func, Sync):
        return func._func  # noqa: SLF001
    return Async(func)

toggle_func 

toggle_func(
    func: Callable[_P, Coroutine[Any, Any, _T" optional hover>_T]],
) -> Callable[_P, _T]

toggle_func(
    func: Callable[_P, _T],
) -> Callable[_P, Awaitable[_T]]

toggle_func(
    func: (
        Callable[_T" optional hover>_T, _T]
        | Callable[_T" optional hover>_T, Coroutine[Any, Any, _T]]
    ),
) -> Callable[_P, _T] | Callable[_P, Awaitable[_T]]

Convert between synchronous and asynchronous functions.

Parameters:

Name Type Description Default

func

Callable[_P, _T] | Callable[_P, Coroutine[Any, Any, _T]]

A function that can be either synchronous or asynchronous.

 required

Returns:

Type Description
Callable[_P, _T] | Callable[_P, Awaitable[_T]]

A function that matches the desired synchronicity,
Callable[_P, _T] | Callable[_P, Awaitable[_T]]

either synchronous or asynchronous.
Source code in src/async_wrapper/convert/main.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def toggle_func(
    func: Callable[_P, _T] | Callable[_P, Coroutine[Any, Any, _T]],
) -> Callable[_P, _T] | Callable[_P, Awaitable[_T]]:
    """
    Convert between synchronous and asynchronous functions.

    Args:
        func: A function that can be either synchronous or asynchronous.

    Returns:
        A function that matches the desired synchronicity,
        either synchronous or asynchronous.
    """
    if isinstance(func, (Async, Sync)):
        return func._func  # noqa: SLF001
    if iscoroutinefunction(func):
        return async_to_sync(func)
    return sync_to_async(func)  # pyright: ignore[reportReturnType]