Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]

Groups > comp.lang.python > #102197 > unrolled thread

Re: Question about asyncio and blocking operations

Back to article view | Back to comp.lang.python

This discussion starts older than the indexed window; earlier articles aren't shown. The article labeled Started by below is the oldest one visible, not the original post.

Contents

  Re: Question about asyncio and blocking operations Maxime S <maxischmeii@gmail.com> - 2016-01-28 22:23 +0100

#102197 — Re: Question about asyncio and blocking operations

2016-01-28 17:53 GMT+01:00 Ian Kelly <ian.g.kelly@gmail.com>:

> On Thu, Jan 28, 2016 at 9:40 AM, Frank Millman <frank@chagford.com> wrote:
>
> > The caller requests some data from the database like this.
> >
> >    return_queue = asyncio.Queue()
> >    sql = 'SELECT ...'
> >    request_queue.put((return_queue, sql))
>
> Note that since this is a queue.Queue, the put call has the potential
> to block your entire event loop.
>
>
Actually, I don't think you actually need an asyncio.Queue.

You could use a simple deque as a buffer, and call fetchmany() when it is
empty, like that (untested):

class AsyncCursor:
    """Wraps a DB cursor and provide async method for blocking operations"""
    def __init__(self, cur, loop=None):
        if loop is None:
            loop = asyncio.get_event_loop()
        self._loop = loop
        self._cur = cur
        self._queue = deque()

    def __getattr__(self, attr):
        return getattr(self._cur, attr)

    def __setattr__(self, attr, value):
        return setattr(self._cur, attr, value)

    async def execute(self, operation, params):
        return await self._loop.run_in_executor(self._cur.execute,
operation, params)

    async def fetchall(self):
        return await self._loop.run_in_executor(self._cur.fetchall)


    async def fetchone(self):
        return await self._loop.run_in_executor(self._cur.fetchone)

    async def fetchmany(self, size=None):
        return await self._loop.run_in_executor(self._cur.fetchmany, size)


    async def __aiter__(self):
        return self

    async def __anext__(self):
        if self._queue.empty():
            rows = await self.fetchmany()
            if not rows:
                raise StopAsyncIteration()
            self._queue.extend(rows)

        return self._queue.popleft()

[toc] | [standalone]

Back to top | Article view | comp.lang.python

csiph-web