Test aiohttp with a simple chat

Introduction

Last fall, I was able to visit several python meetups in Kiev.
At one of them, Nikolay Novik spoke and talked about the new asynchronous framework aiohttp , which runs on the library for asynchronous calls asyncio in 3 versions of the python interpreter. I was interested in this framework by the fact that it was created by core python developers and positioned as a python concept framework for the web.

Now there is a huge number of different frameworks, each of which has its own philosophy,
syntax and implementation of common web templates. Hope over time, all this variety
will be on the same basis - aiohttp.

Structure

In order to test the full potential of aiohttp to the maximum, I tried to develop a simple chat on web sockets. The basis of aiohttp is an endless loop in which handlers are spinning. Handler - the so-called coroutine, an object that does not block input / output (I / O). This type of object appeared in python 3.4 in the asyncio library. Until all the calculations in this object take place, he as if falls asleep, and at this time the interpreter can process other objects. To make it clear, give an example. Often, all server delays occur when it waits for a response from the database and until this response comes and is processed, other objects are waiting for their turn. In this case, other objects will be processed until the answer comes from the database. But to implement this, you need an asynchronous driver.
Currently, asynchronous drivers and wrappers are implemented for aiohttp for most popular databases ( postgresql , mysql , redis )
For mongodb there is Motor , which is used in chat.

The chat entry point is app.py. It creates an app object.

import asyncio from aiohttp import web loop = asyncio.get_event_loop() app = web.Application(loop=loop, middlewares=[ session_middleware(EncryptedCookieStorage(SECRET_KEY)), authorize, db_handler, ]) 

As you can see, upon initialization, the loop is passed to the app, as well as a list of middleware, which will be discussed later.

Routes

Unlike the flask that aiohttp is very similar to, routes are added to an already initialized app app.

 app.router.add_route('GET', '/{name}', handler) 

By the way, the explanation of Andrei Svetlov , why it is so implemented.

Filling routes rendered in a separate file routes.py .

 from chat.views import ChatList, WebSocket from auth.views import Login, SignIn, SignOut routes = [ ('GET', '/', ChatList, 'main'), ('GET', '/ws', WebSocket, 'chat'), ('*', '/login', Login, 'login'), ('*', '/signin', SignIn, 'signin'), ('*', '/signout', SignOut, 'signout'), ] 

The first element is the http method, then the url is located, the third in the tuple is the handler object, and finally the route name, so that it is convenient to call it in the code.

Next, the routes list is imported into app.py and they are populated with a simple loop into the application.

 from routes import routes for route in routes: app.router.add_route(route[0], route[1], route[2], name=route[3]) 

Everything is simple and logical

Handlers, Request and Response

I decided to do requests processing following the example of the Django framework. The auth folder contains everything related to users, authorization, the processing of creating a user and his login. And in the chat folder is the logic of the chat, respectively. In aiohttp, you can implement a handler as both a function and a class.
We select implementation through a class.

 class Login(web.View): async def get(self): session = await get_session(self.request) if session.get('user'): url = request.app.router['main'].url() raise web.HTTPFound(url) return b'Please enter login or email' 

About the session will be written below, and everything else I think is clear and true. I want to note that redirection occurs either by returning (return) or by throwing an exception in the form of the web.HTTPFound () object, to which the path is passed as a parameter. Http methods in a class are implemented through asynchronous functions get, post, and so on. There are some features if you need to work with query parameters.

 data = await self.request.post() 

Configuration settings

All settings are stored in the settings.py file. I use envparse to store sensitive data. This utility allows you to read data from environment variables, as well as to parse a special file where these variables are stored.

 if isfile('.env'): env.read_envfile('.env') 

Firstly, it was necessary to raise the project on Heroku, and secondly, it turned out to be also very convenient. At first I used a local database, and then I tested it on a remote one, and the switch consisted of changing just one line in the .env file.

Middlewares

When you initialize the application, you can set the middleware. Here they are in a separate file . The standard implementation is a decorator function in which you can do checks or any other actions with the request.

An example of an authorization check

 async def authorize(app, handler): async def middleware(request): def check_path(path): result = True for r in ['/login', '/static/', '/signin', '/signout', '/_debugtoolbar/']: if path.startswith(r): result = False return result session = await get_session(request) if session.get("user"): return await handler(request) elif check_path(request.path): url = request.app.router['login'].url() raise web.HTTPFound(url) return handler(request) else: return await handler(request) return middleware 

There is also a middleware for connecting the database.

 async def db_handler(app, handler): async def middleware(request): if request.path.startswith('/static/') or request.path.startswith('/_debugtoolbar'): response = await handler(request) return response request.db = app.db response = await handler(request) return response return middleware 

Connection details below.

Database

For chat use Mongodb and asynchronous driver Motor. Connection to the database occurs when the application is initialized.

 app.client = ma.AsyncIOMotorClient(MONGO_HOST) app.db = app.client[MONGO_DB_NAME] 

And the connection is closed in the special function shutdown.

 async def shutdown(server, app, handler): server.close() await server.wait_closed() app.client.close() # database connection close await app.shutdown() await handler.finish_connections(10.0) await app.cleanup() 

I want to note that in the case of an asynchronous server, you need to correctly complete all parallel tasks.

A little more about the creation of event loop.

 loop = asyncio.get_event_loop() serv_generator, handler, app = loop.run_until_complete(init(loop)) serv = loop.run_until_complete(serv_generator) log.debug('start server', serv.sockets[0].getsockname()) try: loop.run_forever() except KeyboardInterrupt: log.debug(' Stop server begin') finally: loop.run_until_complete(shutdown(serv, app, handler)) loop.close() log.debug('Stop server end') 

The loop itself is created from asyncio.

 serv_generator, handler, app = loop.run_until_complete(init(loop)) 

The run_until_complete method adds corutines to the loop. In this case, it adds the function to initialize the application.

 try: loop.run_forever() except KeyboardInterrupt: log.debug(' Stop server begin') finally: loop.run_until_complete(shutdown(serv, app, handler)) loop.close() 

Actually the implementation of an infinite loop, which is interrupted in case of an exception. Before closing, the shutdown function is called, which terminates all connections and correctly stops the server.

Now we need to figure out how to query, retrieve and modify data.

 class Message(): def __init__(self, db, **kwargs): self.collection = db[MESSAGE_COLLECTION] async def save(self, user, msg, **kw): result = await self.collection.insert({'user': user, 'msg': msg, 'time': datetime.now()}) return result async def get_messages(self): messages = self.collection.find().sort([('time', 1)]) return await messages.to_list(length=None) 

Although I’m not involved in ORM, it’s more convenient to do database queries in separate classes. The files.py file was created in the chat folder, where the Message class is located. In the get_messages method, a query is created that retrieves all the stored messages, sorted by time. The save method creates a request to save a message to the database.

Templates

For aiohttp several asynchronous wrappers are written for popular templating engines, in particular aiohttp_jinja2 and aiohttp_mako . For chat I use jinja2.

 aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader('templates')) 

This is how template support is initialized in the application.
FileSystemLoader ('templates') tells jinja2 that our templates are in the templates folder.

 class ChatList(web.View): @aiohttp_jinja2.template('chat/index.html') async def get(self): message = Message(self.request.db) messages = await message.get_messages() return {'messages': messages} 

Through the decorator, we specify which template we will use in views, and to fill the context, we return a dictionary with variables that we later work with in the template.

Sessions, authorization

To work with sessions there is aiohttp_session library. It is possible to store sessions in Redis or in cookies in encrypted form using cryptography. The storage method is specified when the library is installed.

 aiohttp_session[secure] 

To initialize the session, add it to the middleware.

 session_middleware(EncryptedCookieStorage(SECRET_KEY)), 

To get or put values ​​into a session, you must first extract it from the request.

 session = await get_session(request) 

To authorize a user, add his id to the session, and then check its presence in middleware. Of course, for security, more checks are needed, but for testing the concept, that will be enough.

Static

A folder with static content is connected to a separate route when the application is initialized.

 app.router.add_static('/static', 'static', name='static') 

To use it in the template, you need to get it from the app.

 <script src="{{ app.router.static.url(filename='js/main.js') }}"></script> 

It's simple, nothing complicated.

Websocket

Finally we got to the most delicious part of aiohttp). The implementation of socket is very simple. In javascript I added the minimum necessary functionality for its work.

 try{ var sock = new WebSocket('ws://' + window.location.host + '/ws'); } catch(err){ var sock = new WebSocket('wss://' + window.location.host + '/ws'); } // show message in div#subscribe function showMessage(message) { var messageElem = $('#subscribe'), height = 0, date = new Date(); options = {hour12: false}; messageElem.append($('<p>').html('[' + date.toLocaleTimeString('en-US', options) + '] ' + message + '\n')); messageElem.find('p').each(function(i, value){ height += parseInt($(this).height()); }); messageElem.animate({scrollTop: height}); } function sendMessage(){ var msg = $('#message'); sock.send(msg.val()); msg.val('').focus(); } sock.onopen = function(){ showMessage('Connection to server started') } // send message from form $('#submit').click(function() { sendMessage(); }); $('#message').keyup(function(e){ if(e.keyCode == 13){ sendMessage(); } }); // income message handler sock.onmessage = function(event) { showMessage(event.data); }; $('#signout').click(function(){ window.location.href = "signout" }); sock.onclose = function(event){ if(event.wasClean){ showMessage('Clean connection end') }else{ showMessage('Connection broken') } }; sock.onerror = function(error){ showMessage(error); } 

To implement the server side, I use the class WebSocket

 class WebSocket(web.View): async def get(self): ws = web.WebSocketResponse() await ws.prepare(self.request) session = await get_session(self.request) user = User(self.request.db, {'id': session.get('user')}) login = await user.get_login() for _ws in self.request.app['websockets']: _ws.send_str('%s joined' % login) self.request.app['websockets'].append(ws) async for msg in ws: if msg.tp == MsgType.text: if msg.data == 'close': await ws.close() else: message = Message(self.request.db) result = await message.save(user=login, msg=msg.data) log.debug(result) for _ws in self.request.app['websockets']: _ws.send_str('(%s) %s' % (login, msg.data)) elif msg.tp == MsgType.error: log.debug('ws connection closed with exception %s' % ws.exception()) self.request.app['websockets'].remove(ws) for _ws in self.request.app['websockets']: _ws.send_str('%s disconected' % login) log.debug('websocket connection closed') return ws 

The socket itself is created using the WebSocketResponse () function. Be sure to use it before cooking. The list of open sockets is stored in my application (so that when closing the server they can be correctly closed). When a new user is connected, all participants receive a notification that the new participant has joined the chat. Next we expect a message from the user. If it is valid, we save it in a database and send it to other members of the chat.
When the socket is closed, we remove it from the list and notify the chat that one of the participants has left it. Very simple implementation, visually in synchronous style, without a large number of callbacks, as in Tornado for example. Take it and use it).

Upload to Heroku

I posted the test chat on Heroku, for visual demonstration. During installation, several problems arose, in particular, to use their internal mongodb database, it was necessary to enter credit card information, which I did not want to do, so I used the services of MongoLab and created a base there. Then there were problems with installing the application itself. To install cryptography, you had to explicitly specify it in requirements.txt. Also, to specify the version of python, you need to create a runtime.txt file in the project root.

findings

In general, creating a chat, learning aiohttp, analyzing the work of sockets and some other technologies that I haven’t worked with before, took me about 3 weeks of work in the evenings and rarely on weekends.
The documentation in aiohttp is pretty good, many asynchronous drivers and wrappers are ready for testing.
Perhaps for production, not everything is ready yet, but development is very active (in 3 weeks aiohttp was updated from version 0.19 to 0.21).
If you need to add sockets to the project, this option is perfect, so as not to add a heavy Tornado depending.

Links

• Github

  
  

• Aiohttp

  
  

• Aiohttp libs

  
  

• Mongodb

  
  

• Motor

  
  

• envparse

  
  
• Writing a framework on asyncio, aiohttp and thinking about Python3 part one

All errors and omissions please send in PM :)

UPD

Some time has passed since the article was released, unfortunately the application on Heroku has long been unavailable, so testing the demo will not work anymore. There was some time to update the code and dependencies, but the article still has the old code.