init

Author: gistart

.gitignore

@@ -0,0 +1,15 @@
+*.swo
+*.swp
+*.pyc
+*.pyo
+*.log
+*.lock
+.env*
+.pypyenv*
+.pytest_*
+.git
+__pycache__
+local_settings.py
+
+!.gitkeep
+!/.gitignore

README.md

@@ -0,0 +1,3 @@
+# asyncio-pool
+
+TODO

__init__.py

@@ -0,0 +1,3 @@
+# coding: utf8
+
+from .pool import AioPool

asyncio_pool/__init__.py

@@ -0,0 +1,85 @@
+# coding: utf8
+
+import traceback
+import asyncio as aio
+
+
+class AioPool(object):
+
+    def __init__(self, size=1024, loop=None):
+        self._size = size
+        self._waiting = 0
+        self._executed = 0
+        self.semaphore = aio.Semaphore(value=self._size)
+        self.loop = loop or aio.get_event_loop()
+        self._all_done = self.loop.create_future()
+
+    @property
+    def is_empty(self):
+        return 0 == self._waiting == (self._size - self.semaphore._value)
+
+    @property
+    def is_full(self):
+        return self._waiting + (self._size - self.semaphore._value) >= self._size
+
+    async def join(self):
+        await self._all_done
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, ext_type, exc, tb):
+        await self.join()
+
+    async def _acquire(self):
+        if self._all_done.done():
+            self._all_done = self.loop.create_future()
+        self._waiting += 1
+        await self.semaphore.acquire()
+        self._waiting -= 1
+
+    async def _wrap(self, coro, future, cb=None, ctx=None):
+        res, exc, tb = None, None, None
+        try:
+            res = await coro
+            future.set_result(res)
+        except Exception as exc:
+            future.set_exception(exc)
+            tb = traceback.format_exc()
+        finally:
+            self.semaphore.release()
+            self._executed += 1
+
+        if cb:
+            await self.spawn(cb(res, (exc, tb), ctx))
+        elif self.is_empty:
+            self._all_done.set_result('done')
+
+    async def spawn(self, coro, cb=None, ctx=None):
+        await self._acquire()
+        future = self.loop.create_future()
+        wrapped = self._wrap(coro, future, cb=cb, ctx=ctx)
+        self.loop.create_task(wrapped)
+        return future
+
+    async def exec(self, coro):
+        return await (await self.spawn(coro))
+
+    async def itermap(self, fn, iterable):
+        for it in iterable:
+            yield (await self.spawn(fn(it)))
+
+    async def map(self, fn, iterable, exc_as_result=True):
+        futures = []
+        async for fut in self.itermap(fn, iterable):
+            futures.append(fut)
+        await self.join()
+
+        result = []
+        for fut in futures:
+            if fut.exception():
+                res = fut.exception() if exc_as_result else None
+            else:
+                res = fut.result()
+            result.append(res)
+        return result

asyncio_pool/pool.py

@@ -0,0 +1,40 @@
+# coding: utf8
+
+import logging
+import asyncio as aio
+import json as myformat   # just for a demo
+
+
+def parse_res(res):
+    doc = myformat.loads(res)
+    return doc.get('files') or [], doc.get('folders') or []
+
+
+async def ls_cb(res, err, ctx):
+    apipool, dbpool, api, db, log = ctx
+    if err:
+        log.error()
+        return
+
+    files, folders = parse_res(res)
+    for folder in folders:
+        await apipool.spawn(api.ls(folder['path']), cb=ls_cb, ctx=ctx)
+        await dbpool.spawn(db.save(folder))
+    for file in files:
+        await dbpool.spawn(db.save(file))
+
+
+async def example_ls():
+    loop = aio.get_event_loop()
+    api = await create_client(loop)
+    db = await create_connection(dsn, loop)
+    log = logging.getLogger('example_ls')
+
+    with AioPool(size=10) as dbpool, \
+            AioPool(size=10) as apipool:
+        ctx = (apipool, dbpool, api, db, log)
+        await apipool.spawn(api.ls('/'), cb=ls_cb, ctx=ctx)
+
+
+if __name__ == "__main__":
+    aio.get_event_loop().run_until_complete(example_ls())

examples/__init__.py

@@ -0,0 +1,3 @@
+ipython
+pytest
+pytest-asyncio

examples/ls_remote.py

@@ -0,0 +1,25 @@
+#! /bin/sh
+
+py35=python3.5
+py36=python3.6
+py37=python3.7
+pypy3=/opt/pypy3/pypy3/bin/pypy3
+
+default_env=$py36
+
+
+for py in $py35 $py36 $py37 $pypy3
+do
+    if [ -x "$(command -v $py)" ]; then
+        pyname="$(basename $py)"
+        envname=".env_$pyname"
+
+        if ! [ -d $envname ]; then
+            echo "$pyname: virtual env does not exist"
+        else
+            $envname/bin/pytest tests
+        fi
+    else
+        echo "$py: not found"
+    fi
+done