AsyncioPatch

A mixin written by me to use to enable async on all your non async classes! It's not only a patch to use it but actually really executes the code in the event loop!

How to use


# (C) retoor 2024-12-01

import os 
import AsyncioPatch from asyncio_patch.py

# Non async traditional class.
class Reader:

    def readline(self):
        return os.stdin.readline()

# Extend the class with the mixin.
class AsyncReader(Reader, AsyncioPatch):
    pass

# Use it for blocking code.
async def main():
    reader = AsyncReader()
    
    # method 1
    print(await reader.areadline())
    
    # method 2
    print(await reader.readline_async())

asyncio.run(main())

Source code

import asyncio

class AsyncioPatch:
    def __getattr__(self, name):
        if name.endswith("_async"):
            name = name.split("_async")[0]
        elif name.startswith("a"):
            name = name[1:]
        else:
            return self.__dict__.get(name, None)
        return self.patch_async(getattr(self, name))


    def patch_async(self, method, *args, **kwargs):
        async def patched_async(*args, **kwargs):
            loop = asyncio.get_running_loop()
            import functools 
            patched_call = functools.partial(method, *args, **kwargs)
            return await loop.run_in_executor(None, patched_call)
        return patched_async 
    
    async def __aenter__(self):
        fn = self.patch_async(self.__enter__)
        return await fn()

    async def __aexit__(self, *args, **kwargs):
        fn = self.patch_async(self.__exit__,*args, **kwargs)
        return await fn()