anddr anddr - 5 months ago 57
Python Question

How to create a parallel loop using aiohttp or asyncio in Python?

I would like to use rethinkdb .changes() feature to push some messages to users. The messages should send without any requests from the users.

I am using rethinkdb with aiohttp and websockets. How it works:

  1. User sends message

  2. Server puts it into rethinkdb

  3. What I need: an additional loop uses rethinkdb
    function to send updates to connected users

This is how I initiate the application:

def init(loop):
app = Application(loop=loop)
app['sockets'] = []
app['susers'] = []
app.router.add_route('GET', '/', wshandler)
handler = app.make_handler()
srv = yield from loop.create_server(handler, '', 9080)
print("Server started at")
return app, srv, handler

In the
I have a loop, which processes incoming messages:

def wshandler(request):
resp = WebSocketResponse()
if not resp.can_prepare(request):
return Response(
body=bytes(json.dumps({"error_code": 401}), 'utf-8'),
yield from resp.prepare(request)['sockets'].append(resp)
print('Someone connected')
while True:
msg = yield from resp.receive()
if == MsgType.text:
runCommand(msg, resp, request)
print('Someone disconnected.')
return resp

How to create a second loop sending messages to the same pool of opened connections? How to make it thread-safe?


Generally speaking, you should try to avoid threads as much as a possible whenever running an event loop.

Unfortunately rethinkdb does not support asyncio out-of-the-box, but it does support the Tornado & Twisted frameworks. So, you could bridge Tornado & asyncio and make it work without using threads.


As Andrew pointed out rethinkdb does support asyncio. After 2.1.0 you can presumably do:


And then in your web handlers:

res = await rethinkdb.table(tbl).changes().run(connection)
while await res.fetch_next():