@pquentin I would find this useful for my projects... For the time being, I just post-process my code to get the same effect.

A key use-case is for when the async/sync implementation is fundamentally different... particularly when I need the resulting code to pass mypy type checking.

As a small example, I have two functions that provide sync/async implementations: async_create_process and sync_create_process.

async def async_create_process(runargs: list[str]) -> asyncio.subprocess.Process:
    return await asyncio.subprocess.create_subprocess_exec(*runargs, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)


def sync_create_process(runargs: list[str]) -> subprocess.Popen[bytes]:
    return subprocess.Popen(runargs, stdin=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE)

In my async code that gets processed by unasync, I call the async version of the function... (There's also actually a class to wrap the process objects and abstract some of the other API differences, but omitting that for brevity)

The problem is that, after rewriting I end up with this:

def sync_create_process(runargs: list[str]) -> asyncio.subprocess.Process:
    return asyncio.subprocess.create_subprocess_exec(*runargs, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)


def sync_create_process(runargs: list[str]) -> subprocess.Popen[bytes]:
    return subprocess.Popen(runargs, stdin=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE)

Technically, this code works because the second function overwrites the first one and all callsites for async_create_process now call the correct sync_create_process. However, mypy hates this, for a number of reasons. Not only does the first definition fail type checking due to the return annotation, but it also carries to mypy's inspection of all the callsites.

To fix this, I created a post-processor that looks for # unasync: remove and removes the node in the sync code.

So if I add the comment like this...

async def async_create_process(runargs: list[str]) -> asyncio.subprocess.Process: # unasync: remove

The resulting code has just the sync implementation...

def sync_create_process(runargs: list[str]) -> subprocess.Popen[bytes]:
    return subprocess.Popen(runargs, stdin=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE)

And mypy is happy everywhere!

Implementing this might also be a solution/workaround for issues like #74 and #64

If anyone is interested in the (largely untested) code, it's available in this gist

Uh oh!

Option to exclude code #67

Description

Metadata

Metadata

Assignees

Labels

Type

Fields

Projects

Milestone

Relationships

Development

Issue actions