asyncio compatibility

[synapticarbors]

I was curious as to whether it would be possible to swap out requests for something like aiohttp in a clean way to enable use of the library with asyncio. I guess I could always opt to call everything via run_in_executor.

Has anyone attempted this or thought about the problem?

[limx0]

#106 would allow you to implement this

[synapticarbors]

That would be perfect. What happened to #107? Was the design rejected. I'm +1 for the idea.

[limx0]

I think @rozzac90 and I were the only ones interested and @liampauling wasn't sold. Keen to reopen it.

[liampauling]

With the asyncio API changing in almost every py release I was a bit nervous about implementing it into lightweight. Are there any libraries out there that have done this successfully? I assume with #106 the actual requests would be done outside lightweight, it would be great if it could instead be somehow included by overriding the baseendpoint or provide an http client instead.

I also know that requests had plans to provide concurrency through twisted but haven’t seen anything yet.

[limx0]

I agree, we don't want to depend on asyncio, #106 allows the flexibility to do whatever the user chooses. I don't believe there is any nice way to override the base http client to make it play well with aiohttp. Rather than nice clean async code such as

req = api.insert_order(..., return_kwargs=True)
await aiohttp.request(**req)

You end up having to create a bunch of executors and wrapping synchronous in futures. This leads to poor readability and having to spawn thread or process pools (not how asyncio is supposed to work)

[liampauling]

I see were you coming from but is this the only design pattern that can be used when integrating async libraries?

[limx0]

That I'm not 100% sure on, but IO should really be decoupled from function logic, which is what this would enable. I will do some more investigating for options.

[liampauling]

Bumping this as I am looking into having an easy way to use urllib3 connection pool, is anyone using #107?

[HMaker]

Can't it be ran in a background thread and janus be used to interface it with async code?

[HMaker]

@liampauling @limx0 callbacks could be used to avoid code duplication between sync and async versions, here follows an example for streaming:

# newstream.py

import inspect
import asyncio
import typing as t


class Deferred:
    __slots__ = ('_fut', '_callback')

    def __init__(self, fut: t.Awaitable):
        self._fut = fut
        self._callback = None

    def then(self, callback: t.Callable[[t.Any], t.Any]):
        self._callback = callback
        return self

    def __await__(self):
        result = yield from self._fut.__await__()
        if self._callback is not None:
            return self._callback(result)
        return result


class DoneDeferred:
    __slots__ = ('_result',)

    def __init__(self, result):
        self._result = result

    def then(self, callback: t.Callable[[t.Any], t.Any]):
        return callback(self._result)


def maydefer(obj):
    if inspect.isawaitable(obj):
        return Deferred(obj)
    return DoneDeferred(obj)


class OldBetfairStream:

    # the code as it is actually

    def authenticate(self) -> int:
        """Authentication request."""
        unique_id = 1
        message = {
            "op": "authentication",
            "id": unique_id,
            "appKey": 'app_key',
            "session": 'session_token',
        }
        self._send(message)
        return unique_id

    def _send(self, msg):
        pass


class NewBetfairStream:

    # proposed change

    def authenticate(self) -> int:
        """Authentication request."""
        unique_id = 1
        message = {
            "op": "authentication",
            "id": unique_id,
            "appKey": 'app_key',
            "session": 'session_token',
        }
        return maydefer(self._send(message)).then(lambda _: unique_id)

    def _send(self, msg):
        pass
    

class AsyncBetfairStream(NewBetfairStream):

    # async version of proposed change
    # note that this subclass overrides only _send(), not authenticate()

    async def _send(self, msg):
        pass


async def main():
    assert NewBetfairStream().authenticate() == await AsyncBetfairStream().authenticate() == 1


if __name__ == '__main__':
    asyncio.run(main())

but there is a cost, it makes calls to authenticate() almost 7x slower:

$ python -m timeit -n 1000000 -s "import newstream as ns; authenticate = ns.OldBetfairStream().authenticate" "authenticate()" 
1000000 loops, best of 5: 295 nsec per loop
$ python -m timeit -n 1000000 -s "import newstream as ns; authenticate = ns.NewBetfairStream().authenticate" "authenticate()"
1000000 loops, best of 5: 2.04 usec per loop

it won't be a problem if those kind of functions don't need to be called for each received message from streaming API. For HTTP API it's not a problem since request time will be way greater than that.

[HMaker]

@liampauling you don't have any thoughts on the above example of async support?

[liampauling]

I don't understand py async and I don't think I ever will, feel free to join the slack to see what others think