How-to guides: advanced

How to create an extension

You can use extensions to achieve a lot of enhancements of the base framework.

Basically, an extension is a function listening to events, for instance:

def cors(app, value='*'):

    async def add_cors_headers(response, request):
        response.headers['Access-Control-Allow-Origin'] = value

Here the cors extension can be applied to the Roll app object. It listens to the response event and for each of those add a custom header. The name of the inner function is not relevant but explicit is always a bonus. The response object is modified in place.

Note: more extensions are available by default. Make sure to check these out!

How to deal with content negociation

The content_negociation extension is made for this purpose, you can use it that way:


@app.route('/test', accepts=['text/html', 'application/json'])
async def get(req, resp):
    if req.headers['ACCEPT'] == 'text/html':
        resp.headers['Content-Type'] = 'text/html'
        resp.body = '<h1>accepted</h1>'
    elif req.headers['ACCEPT'] == 'application/json':
        resp.json = {'status': 'accepted'}

Requests with Accept header not matching text/html or application/json will be honored with a 406 Not Acceptable response.

How to subclass Roll itself

Let’s say you want your own Query parser to deal with GET parameters that should be converted as objects.

What you can do is subclass the Roll class to set your custom Query class:

from datetime import date

from roll import Roll, Query
from roll.extensions import simple_server

class MyQuery(Query):

    def date(self):
        return date(int(self.get('year')),

class MyRoll(Roll):
    Query = MyQuery

app = MyRoll()

async def hello(request, response):
    response.body =

if __name__ == '__main__':

And now when you pass appropriated parameters (for the sake of brievety, no error handling is performed but hopefully you get the point!):

$ http :3579/hello/ year==2017 month==9 day==20
HTTP/1.1 200 OK
Content-Length: 10


How to deploy Roll into production

The recommended way to deploy Roll is using Gunicorn.

First install gunicorn in your virtualenv:

pip install gunicorn

To run it, you need to pass it the pythonpath to your roll project application. For example, if you have created a module in your package mypackage, where you create your application with app = Roll(), then you need to issue this command line:

gunicorn mypackage.core:app --worker-class roll.worker.Worker

See gunicorn documentation for more details about the available arguments.

Note: it's also recommended to install uvloop as a faster asyncio event loop replacement:

pip install uvloop

How to send custom events

Roll has a very small API for listening and sending events. It's possible to use it in your project for your own events.

Events are useful when you want other users to extend your own code, whether it's a Roll extension, or a full project built with Roll. They differ from configuration in that they are more adapted for dynamic modularity.

For example, say we develop a DB pooling extension for Roll. We would use a simple configuration parameter to let users change the connection credentials (host, username, password…). But if we want users to run some code each time a new connection is created, we may use a custom event.

Our extension usage would look like this:

app = Roll()
db_pool_extension(app, dbname='mydb', username='foo', password='bar')

def listener(connection):
    # dosomething with the connection,
    # for example register some PostgreSQL custom types.

Then, in our extension, when creating a new connection, we'd do something like that:

app.hook('new_connection', connection=connection)

How to protect a view with a decorator

Here is a small example of a WWW-Authenticate protection using a decorator. Of course, the decorator pattern can be used to any kind of more advanced authentication process.

from base64 import b64decode

from roll import Roll

def auth_required(func):

    async def wrapper(request, response, *args, **kwargs):
        auth = request.headers.get('AUTHORIZATION', '')
        # This is really naive, never do that at home!
        if b64decode(auth[6:]) != b'user:pwd':
            response.status = HTTPStatus.UNAUTHORIZED
            response.headers['WWW-Authenticate'] = 'Basic'
            await func(request, response, *args, **kwargs)

    return wrapper

app = Roll()

async def hello(request, response):

How to work with Websockets pings and pongs

While most clients will keep the connection alive and won't expect heartbeats (read ping), some can be more pedantic and ask for a regular keep-alive ping.

import asyncio

async def keep_me_alive(request, ws, **params):
    while True:
            msg = await asyncio.wait_for(ws.recv(), timeout=20)
        except asyncio.TimeoutError:
            # No data in 20 seconds, check the connection.
                pong_waiter = await
                await asyncio.wait_for(pong_waiter, timeout=10)
            except asyncio.TimeoutError:
                # No response to ping in 10 seconds, disconnect.
            # do something with msg

How to consume a request body the asynchronous way

Let’s say you are waiting for big file uploads. You might want to consume the request iteratively to keep your memory consumption low. Here is how to achieve that:

# lazy_body parameter will ask Roll not to load the body automatically
@app.route('/path', lazy_body=True)
async def my_handler(request, response):
    # Prior to accept the upload you can check for headers:
    if headers.get("Authorization") != "Dummy OK":
        # raise, redirect, 401, whatever

    # In case image is a file object you can write onto.
    async for chunk in request:

How to serve a chunked response

In some situations, it's useful to send a chunked response, for example for an unknown sized response body, maybe a file generated on the fly, or to prevent loading a big file in memory.

This is a good occasion to take advantage of using an async library: Roll will automatically serve a chunked response if Response.body is an async generator, or more specifically if it defines the __aiter__ method.

Here is a theoretical example:

async def my_handler(request, response):
    response.body = my_async_generator

Now a more concrete example:

from aiofile import AIOFile, Reader

async def serve_file(path):
    async with AIOFile(path, 'rb') as afp:
        reader = Reader(afp, chunk_size=4096)
        async for data in reader:
            yield data

async def my_handler(request, response):
    response.body = serve_file(path_to_file)
    response.headers['Content-Disposition'] = "attachment; filename=filename.mp3"

Note: the header Transfert-Encoding will be set to chunked, and each chunk length will be calculated and added to the chunk body by Roll.