Aiohttp: Sharing the event loop

[flip111]

I would like to use aiohttp to make a python program that can:

• act as webserver
• act as webserver with websockets
• act as telnet server
• act as telnet client
• offer a CLI interface https://github.com/jonathanslenders/python-prompt-toolkit/blob/2.0/docs/pages/advanced_topics/asyncio.rst https://github.com/jonathanslenders/python-prompt-toolkit/blob/2.0/examples/prompts/asyncio-prompt.py

Because i want to monitor (and use the web to visualize this) and inject traffic on the telnet connection (this python program sits in between two other programs). For telnet i would like to use this library https://github.com/jquast/telnetlib3 which also works with asyncio. I've searched the documentation of aiohttp to see if i can do something with the event loop directly and found this page https://docs.aiohttp.org/en/stable/web_lowlevel.html However it does not fit my use case looking by the first line: Sometimes user don’t need high-level concepts introduced in Server Usage: applications, routers, middlewares and signals.. I DO want all the high-level concepts and use routers, middleware and signals for all my http related stuff. I just like to mix in telnet as well. My goal is stream values between telnet and websocket. How can i have access to the low-level event loop but keep all the high-level good stuff that aiohttp gives me? I would also like to prevent aiohttp from blocking the terminal so i can provide a CLI interface with python-prompt-toolkit.

[asvetlov]

Take a look on https://github.com/aio-libs/aiohttp/blob/master/aiohttp/web_runner.py

Documentation will be ready up to aiohttp 3.0 release date

[flip111]

Documentation will be ready up to aiohttp 3.0 release date

Do you mean a 3.0 release is coming out which has documentation about this?

[flip111]

Take a look on https://github.com/aio-libs/aiohttp/blob/master/aiohttp/web_runner.py

So i see a bunch of classes that use asyncio. I guess they will all be sharing the same loop because they all do asyncio.get_event_loop(). I would think it's up to me to define such classes and then instantiate them and then it's suppose to work just like this? Is that right or is there more to it?

[flip111]

In the tutorial there is a line web.run_app(app, host='127.0.0.1', port=8080). I don't see host or port in the AppRunner class. Am i suppose to pass my app to both the AppRunner and web.run_app() ?

[asvetlov]

The problem is that run_app is blocking function.
You can use app.make_handler() though (see https://github.com/aio-libs/aiohttp/blob/2.3/aiohttp/web.py#L432) but proper initialization/finalizing code is not trivial.

In aiohttp 3.0 run_app is reimplemented on top of AppRunner: https://github.com/aio-libs/aiohttp/blob/master/aiohttp/web.py#L39
Runner is a container for running application, sites like TCPSite is responsive for handling specific TCP host/port .

[flip111]

So run_app is blocking in 2.* and non-blocking in 3.0 ? I tried using aiohttp 3.0 by doing pip install git+https://github.com/aio-libs/aiohttp.git@master. But when i use this code

print("A")
web.run_app(app, host='localhost', port=8080)
print("B")

i get

======== Running on http://linvm418:8080 ========
(Press CTRL+C to quit)
^CB

Looks to me run_app is still blocking (because i need to send sigINT to unblock) .. Did i misunderstood something here?

[asvetlov]

run_app is still blocking, but implementation is significantly simplified by AppRunner/sites usage.
Look on run_app as on AppRunner/TCPSite usage example.

[flip111]

Ok i looked at the implementation of run_app and i split it up in two functions (setup_app and run_loop) so i have the opportunity to add more asyncio stuff in between with loop.run_until_complete(). The run_loop function is a bit dirty right now because it just cleans up the only runner i pass into it and not any additional stuff which i can register between setup_app and run_loop. But as a proof of concept .. is this the way to go?

EDIT: maybe another idea would be to pass a few additional sites to run_app ?

[asvetlov]

Sorry, I don't follow.

[flip111]

Sorry, i should have added the calling code to the gist. I updated the code to add a snippet of app.py. This is just an idea of how i could add other asyncio stuff in the same loop together with aiohttp. Another idea is not call run_app with empty sites but pass them to the function as argument. Can you make a recommendation?

[asvetlov]

runner = AppRunner(app)
await runner.setup()
site = TCPSiter(runner, host, port)
await site.start()

[flip111]

Hello, thanks for the code. I have my app defined as app = web.Application() from the tutorial. I looked in the code of the Application class but i didn't see a setup method. Did you mean await runner.setup() on the second line?

I saw that in the run_app code or my own setup_app code the runner.setup() method is being called as part of a call to run_until_complete (loop.run_until_complete(runner.setup())). I think i understand the concept of async code and promises. But why do you choose to replace run_until_complete with await? I guess i could put your code in an async function and then let it run by the loop.

On the first line you use AppRunner does that mean this code no longer makes use of the logic in run_app? This logic seems very handy to me, i'd like to explore ways to make it reusable .. but if that's not possible or just in general a bad idea i will do without it.

[asvetlov]

Yes, I did mean runner.setup().

[sxpert]

I just got it to work with a trivial patch to TCPSite :

#!/usr/bin/env python3.6
# -*- coding: utf-8 -*-
import asyncio
from aiohttp import web, web_runner


class TCPSite_patched(web_runner.BaseSite):
    """
    this is a copy paste of the TCPSite code except where shown
    """
    __slots__ = ('_host', '_port', '_reuse_address', '_reuse_port')

    def __init__(self, runner, host=None, port=None, *,
                 shutdown_timeout=60.0, ssl_context=None,
                 backlog=128, reuse_address=None,
                 reuse_port=None,
                 # >>> added this
                 loop=None
                 #
                 ):
        super().__init__(runner, shutdown_timeout=shutdown_timeout,
                         ssl_context=ssl_context, backlog=backlog)
        # >>> added this
        if loop is None:
            loop = asyncio.get_event_loop()
        self.loop = loop
        #
        if host is None:
            host = "0.0.0.0"
        self._host = host
        if port is None:
            port = 8443 if self._ssl_context else 8080
        self._port = port
        self._reuse_address = reuse_address
        self._reuse_port = reuse_port

    @property
    def name(self):
        scheme = 'https' if self._ssl_context else 'http'
        return str(URL.build(scheme=scheme, host=self._host, port=self._port))

    async def start(self):
        await super().start()
        # <<< removed this
        # loop = asyncio.get_event_loop()
        #
        # <<< modified this
        # self._server = await loop.create_server(
        # >>> into this
        self._server = await self.loop.create_server(
            #
            self._runner.server, self._host, self._port,
            ssl=self._ssl_context, backlog=self._backlog,
            reuse_address=self._reuse_address,
            reuse_port=self._reuse_port)

# ----
# below is sample code


class WebServer(object):
    def __init__(self, address='127.0.0.1', port=8080, loop=None):
        self.address = address
        self.port = port
        if loop is None:
            loop = asyncio.get_event_loop()
        self.loop = loop
        asyncio.ensure_future(self.start(), loop=self.loop)

    async def start(self):
        self.app = web.Application(loop=self.loop, debug=True)
        self.setup_routes()
        self.runner = web.AppRunner(self.app)
        await self.runner.setup()
        self.site = TCPSite_patched(self.runner,
                                    self.address, self.port,
                                    loop=self.loop)
        await self.site.start()
        print('------ serving on %s:%d ------'
              % (self.address, self.port))

    def setup_routes(self):
        self.app.router.add_get('/', self.index)

    async def index(self, request):
        return web.Response(text='Hello Aiohttp!!')


class LoopTester(object):
    def __init__(self, loop=None):
        if loop is None:
            loop = asyncio.get_event_loop()
        self.loop = loop
        self.counter = 0
        asyncio.ensure_future(self.run(), loop=self.loop)

    async def run(self):
        while self.loop.is_running():
            print('basic test %d' % (self.counter))
            self.counter += 1
            await asyncio.sleep(1)

if __name__ == '__main__':
    import logging

    loop = asyncio.get_event_loop()
    logging.basicConfig(level=logging.DEBUG)
    loop.set_debug(False)
    lt = LoopTester(loop=loop)
    ws = WebServer(loop=loop)
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        tasks = asyncio.gather(
                    *asyncio.Task.all_tasks(loop=loop),
                    loop=loop,
                    return_exceptions=True)
        tasks.add_done_callback(lambda t: loop.stop())
        tasks.cancel()

[tartavull]

is this the simplest way of injecting a loop into aiohttp.web.run_app ?