Utilities

Various utilities for working with ZyncIO.

enum Mode(value)

ZyncIO execution mode.

Valid values are as follows:

SYNC = <Mode.SYNC: 'sync'>

ASYNC = <Mode.ASYNC: 'async'>

SYNC: Final = Mode.SYNC

Module-level alias.

ASYNC: Final = Mode.ASYNC

Module-level alias.

ZYNC_MODE_CACHE_ATTR: Final = '__zync_cached_mode__'

Name of the attribute used by zyncio.get_mode to cache the mode of a ZyncDelegator object.

You can use this constant to allow caching when defining __slots__.

get_mode(obj)

Get the Mode of obj, if it can be determined.

If obj is an instance of zyncio.SyncMixin or zyncio.AsyncMixin, returns the corresponding mode.

If obj implements the zyncio.ZyncDelegator protocol, this function will be called recursively with the object returned by the method. The result will be cached on obj, and any intermediate objects, if possible.

Parameters:

obj (object) – The object to inspect.

Return type:

Mode | None

Returns:

The Mode of obj if it has one, otherwise None.

is_sync(obj)

Check if obj is a subclass of zyncio.SyncMixin or implements zyncio.ZyncDelegator[zyncio.SyncMixin]`.

Parameters:

obj (object) – The object to inspect.

Return type:

TypeIs[SyncMixin | ZyncDelegator[SyncMixin]]

is_async(obj)

Check if obj is a subclass of zyncio.AsyncMixin or implements zyncio.ZyncDelegator[zyncio.SyncMixin].

Parameters:

obj (object) – The object to inspect.

Return type:

TypeIs[AsyncMixin | ZyncDelegator[AsyncMixin]]

run_sync(coro)

Run a coroutine synchronously.

The coroutine must only await other coroutines, recursively. Awaiting any non-coroutine (such as an asyncio.Future), at any point in the call chain, will cause the function to fail.

Parameters:

coro (Coroutine[Any, Any, ReturnT_co]) – The sync coroutine to run.

Return type:

ReturnT_co

Returns:

The return value of the coroutine.

make_sync(func)

Wrap an async function make it run synchronously using zyncio.run_sync.

This function is useful for overloaded functions and methods, whose signatures can’t be captured properly by zyncio.zmethod etc.

Example

class BaseClient(zyncio.ZyncBase):
    @overload
    async def _overloaded_method(self, x: str) -> bytes: ...
    @overload
    async def _overloaded_method(self, x: int) -> bool: ...
    async def _overloaded_method(self, x: str | int) -> bytes | bool:
        ...

class SyncClient(zyncio.SyncMixin, BaseClient):
    overloaded_method = zyncio.make_sync(BaseClient._overloaded_method)

class AsyncClient(zyncio.AsyncMixin, BaseClient):
    overloaded_method = BaseClient._overloaded_method

If any of your overload signatures use Self, you may need to define your method outside of the class for type checkers to infer the correct type for Self in subclasses:

class BaseClient(zyncio.ZyncBase):
    ...

ClientT = TypeVar('ClientT', bound=BaseClient)

@overload
async def overloaded_method(self: ClientT, x: str) -> ClientT: ...
@overload
async def overloaded_method(self, x: int) -> bool: ...
async def overloaded_method(self: ClientT, x: str | int) -> ClientT | bool:
    ...

class SyncClient(zyncio.SyncMixin, BaseClient):
    overloaded_method = zyncio.make_sync(overloaded_method)

class AsyncClient(zyncio.AsyncMixin, BaseClient):
    overloaded_method = overloaded_method

Parameters:

func (async (*P.args, **P.kwargs) -> ReturnT_co) – The function to wrap.

Return type:

(*P.args, **P.kwargs) -> ReturnT_co