Write dual sync/async interfaces with minimal duplication.

---

What is ZyncIO?

If I had a nickel for every variation of my library that I maintain, I'd have two nickels... which isn't a lot, but it's weird that I had to write everything twice.

– Dr. Doofenshmirtz, before discovering ZyncIO.

ZyncIO allows you to write interfaces that can be used synchronously and asynchronously, while avoiding the code duplication this usually entails.

How does it work?

ZyncIO works due to the fact that in Python you can actually run a coroutine without an event loop, as long as your chain of awaits consists exclusively of other coroutines (i.e. no Futures or Tasks):

The behavior of await coroutine is effectively the same as invoking a regular, synchronous Python function.

– A Conceptual Overview of asyncio

To run such a coroutine, we simply call send(None), catch the StopIteration, and extract its value:

coro = pure_coroutine_func()
try:
    coro.send(None)
except StopIteration as e:
    ret = e.value

This means that a single async def function can be made to run in both synchronous and asynchronous contexts, as long as we have a way to determine which mode we're currently using:

async def zync_sleep(zync_mode: zyncio.Mode, secs: float) -> None:
    if zync_mode is zyncio.SYNC:
        time.sleep(secs)
    else:
        await asyncio.sleep(secs)

But this isn't very convenient; you need to pass an additional parameter, and running in sync mode is pretty clunky. That's where zyncio.zfunc comes in:

@zyncio.zfunc
async def zync_sleep(zync_mode: zyncio.Mode, secs: float) -> None:
    ...

zync_sleep.call_sync(3)
asyncio.run(zync_sleep.call_async(3))

@zyncio.zfunc
async def sleep_3(zync_mode: zyncio.Mode) -> None:
    await zync_sleep.call_zync(zync_mode, 3)

The real magic: the class-based API

The real power of ZyncIO comes out when implementing client interfaces:

1. Implement a single base client, using, zyncio.zmethod, zyncio.zproperty and the other ZyncIO method decorators.

2. Create two subclasses a sync client and an async client, adding the zyncio.SyncMixin and zyncio.AsyncMixin mixins respectively.

All of the zyncio.zmethods automatically become sync methods on the sync client and async methods on the async client. zyncio.zpropertys become properties on the sync client, and async methods on the async client.

class BaseClient:
    def __init__(self, sock: socket.socket) -> None:
        self.sock: socket.socket = sock

    @zyncio.zmethod
    async def send_msg(self, data: bytes) -> None:
        if zyncio.is_sync(self):
            self.sock.sendall(data)
        else:
            loop = asyncio.get_running_loop()
            await loop.sock_sendall(self.sock, data)

    @zyncio.zmethod
    async def recv_msg(self, n: int) -> bytes:
        buf = b''
        if zyncio.is_sync(self):
            while len(buf) < n:
                buf += self.sock.recv(n)
        else:
            loop = asyncio.get_running_loop()
            while len(buf) < n:
                buf += await loop.sock_recv(self.sock, n)
        return buf

    @zyncio.zmethod
    async def do_handshake(self) -> None:
        # `.z` (or `.call_zync`) on bound `zmethod`'s (and similar callables)
        # always returns a coroutine, so you can `await` it regardless of the
        # running mode.
        await self.send_msg.z(HANDSHAKE_REQ)
        response = await self.recv_msg.z(len(HANDSHAKE_RESP))
        if response != HANDSHAKE_RESP:
            raise RuntimeError('Handshake failed')

    @zyncio.zproperty
    async def status(self) -> str:
        await self.send_msg.z(STATUS_REQ)
        return (await self.recv_msg.z(STATUS_RESP_LEN)).decode()


class SyncClient(BaseClient, zyncio.SyncMixin):
    pass


class AsyncClient(BaseClient, zyncio.AsyncMixin):
    def __init__(self, sock: socket.socket) -> None:
        super().__init__(sock)
        self.sock.setblocking(False)


sync_client = SyncClient(sock)
sync_client.do_handshake()  # Magically sync!
print('Status:', sync_client.status)  # Sync property


async def use_async_client():
    async_client = AsyncClient(sock)
    await async_client.do_handshake()  # Magically async!
    print('Status:', await sync_client.status())  # Async method

asyncio.run(use_async_client())

Further Reading

Check out the documentation for a tutorial and the full API.

Typing

ZyncIO is fully typed, and built specifically for typed projects. If you're getting unexepcted type checking errors, please open an issue.

License

ZyncIO is distributed under the terms of the MIT license.