Unasyncd

A tool to transform asynchronous Python code to synchronous Python code.

Why?

Unasyncd is largely inspired by unasync, and a detailed discussion about this approach can be found here.

Its purpose is to reduce to burden of having to maintain both a synchronous and an asynchronous version of otherwise functionally identical code. The idea behind simply "taking out the async" is that often, synchronous and asynchronous code only differ slightly: A few awaits, async defs, async withs, and a couple of different method names. The unasync approach makes use of this by treating the asynchronous version as a source of truth from wich the synchronous version is then generated.

Why unasyncd?

The original unasync works by simply replacing certain token, which is enough for most basic use cases, but can be somewhat restrictive in the way the code can be written. More complex cases such as exclusion of functions / classes or transformations (such as AsyncExitStack to ExitStack wich have not only different names but also different method names that then need to be replaced only within a certain scope) are not possible. This can lead to the introduction of shims, introducing additional complexity.

Unasyncd's goal is to impose as little restrictions as possible to the way the asynchronous code can be written, as long as it maps to a functionally equivalent synchronous version.

To achieve this, unasyncd leverages libcst, enabling a more granular control and complex transformations.

Unasyncd features:

1. Transformation of arbitrary modules, not bound to any specific directory structure
2. (Per-file) Exclusion of (nested) functions, classes and methods
3. Optional transformation of docstrings
4. Replacements based on fully qualified names (e.g. typing.AsyncGenerator is different than foo.typing.AsyncGenerator)
5. Transformation of constructs like asyncio.TaskGroup to a thread based equivalent

A full list of supported transformations is available below.

Table of contents

<!-- TOC -->

• Unasyncd
  • Why?
  • Why unasyncd?
  • Table of contents
  • What can be transformed?
    • Asynchronous functions
    • await
    • Asynchronous iterators, iterables and generators
    • Asynchronous iteration
    • Asynchronous context managers
    • contextlib.AsyncExitStack
    • asyncio.TaskGroup
    • anyio.create_task_group
    • asyncio.sleep / anyio.sleep
    • asyncio.Lock / anyio.Lock
    • asyncio.Event / anyio.Event
    • asyncio.Barrier
    • anyio.Path
    • Type annotations
    • Docstrings
  • Usage
    • Installation
    • CLI
    • As a pre-commit hook
    • Configuration
      • File
      • CLI options
      • Exclusions
      • Extending name replacements
    • Handling of imports
    • Integration with linters
    • Limitations
    • Disclaimer

<!-- TOC -->

What can be transformed?

Unasyncd supports a wide variety of transformation, ranging from simple name replacements to more complex transformations such as task groups.

Asynchronous functions

Async

async def foo() -> str:
    return "hello"

Sync

def foo() -> str:
    return "hello"

await

Async

await foo()

Sync

foo()

Asynchronous iterators, iterables and generators

Async

from typing import AsyncGenerator

async def foo() -> AsyncGenerator[str, None]:
    yield "hello"

Sync

from typing import Generator

def foo() -> Generator[str, None, None]:
    yield "hello"

Async

from typing import AsyncIterator

class Foo:
    async def __aiter__(self) -> AsyncIterator[str]:
        ...

    async def __anext__(self) -> str:
        raise StopAsyncIteration

Sync

from typing import Iterator

class Foo:
    def __next__(self) -> str:
        raise StopIteration

    def __iter__(self) -> Iterator[str]:
        ...

Async

x = aiter(foo)

Sync

x = iter(foo)

Async

x = await anext(foo)

Sync

x = next(foo)

Asynchronous iteration

Async

async for x in foo():
    pass

Sync

for x in foo():
    pass

Async

[x async for x in foo()]

Sync

[x for x in foo()]

Asynchronous context managers

Async

async with foo() as something:
    pass

Sync

with foo() as something:
    pass

Async

class Foo:
    async def __aenter__(self):
        ...

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        ...

Sync

class Foo:
    def __enter__(self):
        ...

    def __exit__(self, exc_type, exc_val, exc_tb):
        ...

Async

from contextlib import asynccontextmanager
from typing import AsyncGenerator

@asynccontextmanager
async def foo() -> AsyncGenerator[str, None]:
    yield "hello"

Sync

from contextlib import contextmanager
from typing import Generator

@contextmanager
def foo() -> Generator[str, None, None]:
    yield "hello"

contextlib.AsyncExitStack

Async

import contextlib

async with contextlib.AsyncExitStack() as exit_stack:
    exit_stack.enter_context(context_manager_one())
    exit_stack.push(callback_one)
    exit_stack.callback(on_exit_one)

    await exit_stack.enter_async_context(context_manager_two())
    exit_stack.push_async_exit(on_exit_two)
    exit_stack.push_async_callback(callback_two)

    await exit_stack.aclose()

Sync

import contextlib

with contextlib.ExitStack() as exit_stack:
    exit_stack.enter_context(context_manager_one())
    exit_stack.push(callback_one)
    exit_stack.callback(on_exit_one)

    exit_stack.enter_context(context_manager_two())
    exit_stack.push(on_exit_two)
    exit_stack.callback(callback_two)

    exit_stack.close()

See limitations

asyncio.TaskGroup

Async

import asyncio

async with asyncio.TaskGroup() as task_group:
    task_group.create_task(something(1, 2, 3, this="that"))

Sync

import concurrent.futures

with concurrent.futures.ThreadPoolExecutor() as executor:
    executor.submit(something, 1, 2, 3, this="that")

See limitations

anyio.create_task_group

Async

import anyio

async with anyio.create_task_group() as task_group:
    task_group.start_soon(something, 1, 2, 3)

Sync

import concurrent.futures

with concurrent.futures.ThreadPoolExecutor() as executor:
    executor.submit(something, 1, 2, 3)

See limitations

asyncio.sleep / anyio.sleep

Calls to asyncio.sleep and anyio.sleep will be replaced with calls to time.sleep:

Async

import asyncio

await asyncio.sleep(1)

Sync

import time

time.sleep(1)

If the call argument is 0, the call will be replaced entirely:

import asyncio

await asyncio.sleep(0)

asyncio.Lock / anyio.Lock

Async

import asyncio
# import anyio

lock = asyncio.Lock()

async with lock:
    pass

Sync

import threading

lock = threading.Lock()

with lock:
    pass

asyncio.Event / anyio.Event

Async

import asyncio
# import anyio

event = asyncio.Event()

event.set()
await event.wait()
assert event.is_set()
event.clear()

Sync

import threading

event = threading.Event()

event.set()
event.wait()
assert event.is_set()
event.clear()

asyncio.Barrier

Async

import asyncio

barrier = asyncio.Barrier(2)

await barrier.wait()
await barrier.reset()
await barrier.abort()

barrier.parties
barrier.n_waiting
barrier.broken

Sync

import threading

barrier = threading.Barrier(2)

barrier.wait()
barrier.reset()
barrier.abort()

barrier.parties
barrier.n_waiting
barrier.broken

anyio.Path

Async

import anyio

await anyio.Path().read_bytes()

Sync

import pathlib

pathlib.Path().read_bytes()

Type annotations

| | | |--------------------------------------------|---------------------------------------------| | typing.AsyncIterable[int] | typing.Iterable[int] | | collections.abc.AsyncIterable[int] | collections.abc.Iterable[int] | | typing.AsyncIterator[int] | typing.Iterator[int] | | collections.abc.AsyncIterator[int] | collections.abc.Iterator[int] | | typing.AsyncGenerator[int, str] | typing.Generator[int, str, None] | | collections.abc.AsyncGenerator[int, str] | collections.abc.Generator[int, str, None] | | typing.Awaitable[str] | str | | collections.abc.Awaitable[str] | str |

Docstrings

Simply token replacement is available in docstrings:

Async

async def foo():
    """This