This is a sharer of resources that need to run in a safe synchronous manner but may need to be shared in an asynchronous environment.

sharer.py

# Copyright (c) 2021 by Kristoffer Paulsson. <[email protected]>
import asyncio
from concurrent.futures.thread import ThreadPoolExecutor


class share:
    """Decorator for a method that is being accessed from an event loop and executed in a worker thread."""

    def __init__(self, exe):
        self._exe = exe

    async def __call__(self, *args, **kwargs):
        future = self._obj.pool.submit(self._exe, self._obj, *args, **kwargs)
        await asyncio.sleep(0)
        return future.result()

    def __get__(self, instance, owner):
        self._obj = instance
        return self.__call__


class SharerMixin:
    """Mixin to be inherited by the resource that has to be shared in the asynchronous environment. By using a thread
    pool executor with one worker, all jobs are carried out one at a time. """

    def __init__(self):
        self._pool = ThreadPoolExecutor(max_workers=1)

    @property
    def pool(self) -> ThreadPoolExecutor:
        """Expose the threadpool executor."""
        return self._pool

    def __del__(self):
        self._pool.shutdown()


class Resource(SharerMixin):
    """Example resource to be shared."""

    @share
    def computation(self):
        return 100 * 100 / 2

    @share
    def arguments(self, one, two, three=None, four=None):
        return one == 1 and two == 2 and three == 3 and four == 4

    @share
    def crash(self):
        raise RuntimeWarning("Can you read this exception?")


async def task():
    resource = Resource()

    for _ in range(10):
        try:
            print(await resource.computation())
            print(await resource.arguments(1, 2, three=3, four=4))
            await resource.crash()
        except RuntimeWarning:
            pass
        else:
            raise RuntimeError("Resource should raise exception.")


def main():
    asyncio.run(task())


if __name__ == '__main__':
    main()