asyncio support

[lukesneeringer]

From @graingert on March 6, 2017 16:45

I want to use google-cloud-python in a Sanic web service, however I'm having to use run_in_executor() to use this library without stalling the eventloop.

This is much less efficient than an Asyncio native client.

Copied from original issue: googleapis/google-cloud-python#3103

[lukesneeringer]

From @graingert on March 6, 2017 18:35

@jonparrott it would be possible to support python 2.x and python 3.x with various event loops from the same source by using a custom deferred, future or IOAction implementation.

[lukesneeringer]

From @jonparrott on March 6, 2017 18:36

This might be possible once we exorcise httplib2 from this project. @lukesneeringer

[lukesneeringer]

That sounds right to me (and that has moved up in priority slightly).

Tagging #1998 for reference.

[lukesneeringer]

From @graingert on March 6, 2017 18:40

the core code would use Action composition:

return get_url(xys).map(process_request).flat_map(make_another_request)

Then there could be various, pluggable http transport implementations, that would take an Action instance and return a Future from various event systems (asyncio, twisted), or even use blocking IO.

[lukesneeringer]

From @graingert on March 8, 2017 9:59

import asyncio
import functools


class AsyncIOAction(object):
    def __init__(self, fn):
        this.exec = fn

    def map(self, fn):
        @functools.wraps(fn)
        async def exec():
            value = await self.exec()
            return fn(value)

        return AsyncIOAction(exec)

    def flat_map(self, fn):
        @functools.wraps(fn)
        async def exec():
            value = await self.exec()
            return await fn(value).exec()

        return AsyncIOAction(exec)

    @classmethod
    def all(items):
        async def exec():
            return await asyncio.wait([x.exec() for x in items])

        return AsyncIOAction(exec)

def aiohttp_action(request):
    async def exec():
        return await aiohttp.request(**request)

    return AsyncIOAction(exec)
        
# blocking.py
import functools

class BlockingIOAction(object):
    def __init__(self, fn):
        this.exec = fn

    def map(self, fn):
        @functools.wraps(self, fn)
        def exec():
            value = self.exec()
            return fn(value)
        return BlockingIOAction(exec)

    def flat_map(self, fn):
        @functools.wraps(fn)
        def exec():
            value = self.exec()
            return fn(value).exec()
        
        return BlockingIOAction(exec)

    @classmethod
    def all(items):
        def exec():
            return [x.exec() for x in items]

        return BlockingIOAction(exec)

def requests_action(request): # or httplib2 etc.
    def exec():
        requests.request(**request)

    return BlockingIOAction(exec)

# twisted.py

class TwistedIOAction(object):
    ...

Should be pretty easy with twisted deferred too. Idea being, calling code would choose an IO type, configure the google cloud platform library appropriately, call a method on some API endpoint and then be able to call .exec() on the returned action. The calling code would then be able to receive the value, await and receive the value, or yield it in twisted inline callbacks.

[lukesneeringer]

From @graingert on March 8, 2017 13:57

Here's how you might to inline callbacks: https://gist.github.com/graingert/ca6cdd5d9ae2e18ca917b4594ac8a633#file-asyncio_and_blocking_io-py-L141-L148

[lukesneeringer]

From @graingert on March 8, 2017 14:0

inside google code platform:

def some_api_method(IOAction, ham, spam):
    @IOAction.inline_callbacks
    def _some_api_method():
        v = yield some_other_method(IOAction, ham)
        v2 = yield another_method(IOAction, spam, v)
        Return(v + v2)
        
return _some_api_method()

In some calling code:

async def async_caller():
    await some_api_method(AsyncIOAction, ham="ham", spam="spam").exec()

def blocking_caller():
    some_api_method(BlockingIOAction, ham="ham", spam="spam").exec()

[lukesneeringer]

From @graingert on July 24, 2017 12:8

of course https://pypi.python.org/pypi/txaio might be a better way of implementing it.

[lukesneeringer]

From @jonparrott on July 24, 2017 16:9

@graingert oh that's super interesting. However, this seems to re-enforce the fact that we can't really support this without breaking our public interface and without seriously re-thinking all of our internals. We'd have to start from scratch with this in mind. :/

[lukesneeringer]

From @graingert on July 24, 2017 16:11

@jonparrott with crossbario/txaio#113 and crossbario/txaio#110

you probably won't have to change that much.

And you'd be able to keep the same public api for people using requests.

[lukesneeringer]

From @jonparrott on July 24, 2017 16:14

inlinecallbacks would help, but this is an enormous undertaking and a lot of our code that deals with IO is non-trivial (resumable-media).

I'm open to doing this, but from my perspective it seems like the tooling isn't quite ready and it would take an enormous amount of engineering effort to get all of our plumbing ready- plus, we have a consider how gRPC plays into all this.

[lukesneeringer]

From @dhermes on July 24, 2017 16:15

FWIW, resumable-media would be the easiest thing of all to add support for because the actual I/O is "isolated" (I took the sans-I/O approach when writing it)

[lukesneeringer]

From @graingert on July 24, 2017 16:16

@jonparrott @dhermes yeah a sans-io approach for this whole library would be the best option!