[FIXED] How to choose asynchronous or synchronous method variants in python?

Issue

Let's assume I have a class that is used to perform I/O operations:

class CommunicationStack:
    def __init__(self, socket):
        self.socket = socket

    def sync_get_data(self):
        ...

    def sync_send_data(self):
        ...

    async def async_get_data(self):
        ...

    async def async_send_data(self):
        ...

As you can see it has sync and async variant for the same operations but it would be slightly inconvenient to write async_get_data or sync_get_data manually. I am looking for a smart way to have the same interface like

def get_data(self):
    ...  # call sync variant or return an awaitable, depending on caller type

def send_data(self):
    ...  # call sync variant or return an awaitable, depending on caller type

so it can be conveniently used like:

stack = CommunicationStack(...)

def thread_procedure():
    data = stack.get_data()  # get_data returns data

async def task_procedure():
    data = await stack.get_data()  # get_data returns an awaitable

I believe it can be done in some tricky way with inspections or some black magic:

def is_caller_coroutine():
    return sys._getframe(2).f_code.co_flags & 0x380

to check if caller is a coroutine or a function but it seems like a bad design to mess with python's guts.

---

The question is: what is a good way to choose appropriate variant? Or is there a better way to design everything like using adapters or developing two independent AsyncCommunicationStack and SyncCommunicationStack classes?

Solution

If you want to call async functions same way as regular ones, you may be interested in using gevent.

asyncio wants you to explicitly mark functions as async and to explicitly use await everywhere async stuff happens. In other words asyncio wants you to have different interfaces for sync and async code.

This is intentional decision made to fight with concurrency problems which is much harder to achieve when async nature of a code is hidden (like in gevent).

So yes - two different independent AsyncCommunicationStack and CommunicationStack classes is a way to go if you want to support both worlds. Although once you have async version you can cast write critical code with it and make it sync just running it with asyncio.run(). Problem only is that you won't be able to return to async world after that.

Answered By - Mikhail Gerasimov