aiofiles Add more os calls like os.stat

I'm just poking at this - came to it through a reference on https://github.com/MagicStack/uvloop/issues/1#issuecomment-157537973

Would you be interested in adding support to the project for more calls from the os module like os.stat?

e.g.

os.close
os.fstat
os.read
os.write
os.unlink
os.listdir
os.path.exists
os.rmdir

May 12 '16 18:05 clayg

If someone were to submit quality pull requests (with tests, of course) I wouldn't be opposed to merging them in. :)

May 13 '16 08:05 Tinche

@Tinche Do you have a preference for what aiofiles module os.path functions should go in?

Jun 24 '17 16:06 cjerdonek

@cjerdonek I guess aiofiles.os.path, so we mirror the sync versions closely.

Jul 02 '17 16:07 Tinche

havn't fully tested yet, but should this snippet work?

import asyncio
import inspect
from functools import wraps, partial


def wrap(func):
    @wraps(func)
    async def run(*args, loop=None, executor=None, **kwargs):
        if loop is None:
            loop = asyncio.get_event_loop()
        pfunc = partial(func, *args, **kwargs)
        return await loop.run_in_executor(executor, pfunc)

    return run


class Wrapper:
    pass


def aiowrap(obj):
    if callable(obj):
        return wrap(obj)
    elif inspect.ismodule(obj) or inspect.isclass(obj):
        wrapped_obj = Wrapper()
        if getattr(obj, '__all__'):
            attrnames = obj.__all__
        else:
            attrnames = dir(obj)
        for attrname in attrnames:
            if attrname.startswith('__'):
                continue
            original_obj = getattr(obj, attrname)
            setattr(wrapped_obj, attrname, aiowrap(original_obj))
        return wrapped_obj
    else:
        return obj

Usage

import os
from aiowrap import aiowrap

aios = aiowrap(os)

async def main():
    print(await aios.path.exists('/'))

if __name__ == '__main__':
    import asyncio as aio
    loop = aio.get_event_loop()
    loop.run_until_complete(main())