Path: csiph.com!fu-berlin.de!uni-berlin.de!not-for-mail
From: Maxime S <maxischmeii@gmail.com>
Newsgroups: comp.lang.python
Subject: Re: Question about asyncio and blocking operations
Date: Thu, 28 Jan 2016 22:23:13 +0100
Lines: 60
Sender: maximesteisel@gmail.com
Message-ID: <mailman.65.1454016215.2338.python-list@python.org>
References: <n8038j$575$1@ger.gmane.org> <n8818q$35e$1@ger.gmane.org> <CALwzidk-RBkB-vi6CgcEeoFHQrsoTFvqX9MqzDD=rnY5bOCRUg@mail.gmail.com> <n8aln3$fah$1@ger.gmane.org> <CALwzidn6TvN9W-2qnn2JYvJu8NHzn499nPtfjn9OHjdDcebVbA@mail.gmail.com> <CALwzidmmUeL3Sq9o6ZgKYM3-V22HCshG03wQqsXvhh6xy1otgQ@mail.gmail.com> <CALwzidkr-fT6S6wH2caNaxyQvUdAw=x7XDqkqOfNRrwzwnjvOQ@mail.gmail.com> <n8cm4k$ojk$1@ger.gmane.org> <CAPTjJmr162+K4LZeFpXruR6wxrHxbR-_wkrCLLDyR7kST+kjYg@mail.gmail.com> <n8ct18$fu4$1@ger.gmane.org> <CALwzidnGbz7kM=D7MKua2tA9-csFn9u0OHL0w-X5Bbixpcw4Ow@mail.gmail.com> <n8dgas$s1u$1@ger.gmane.org> <CALwzidn6NfT_O0cfHw1itWja81+MW3scHuEcADVCen3ix6z73w@mail.gmail.com>
Mime-Version: 1.0
Content-Type: text/plain; charset=UTF-8
In-Reply-To: <CALwzidn6NfT_O0cfHw1itWja81+MW3scHuEcADVCen3ix6z73w@mail.gmail.com>
Precedence: list
Xref: csiph.com comp.lang.python:102197

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()