Rather than trying to rewrite Pyramid, Aiopyramid provides a set of features that will allow you to run existing code asynchronously where possible.


Aiopyramid provides three view mappers for calling view callables:

When you include Aiopyramid, the default view mapper is replaced with the CoroutineOrExecutorMapper which detects whether your view callable is a coroutine and does a yield from to call it asynchronously. If your view callable is not a coroutine, it will run it in a separate thread to avoid blocking the thread with the main loop. asyncio is not thread-safe, so you will need to guarantee that either in memory resources are not shared between view callables running in the executor or that such resources are synchronized.

This means that you should not necessarily have to change existing views. Also, it is possible to restore the default view mapper, but note that this will mean that coroutine views that do not specify CoroutineMapper as their view mapper will fail.

If most of your view needs to be a coroutine but you want to call out to code that blocks, you can always use run_in_executor. Aiopyramid also provides a decorator, use_executor(), for specifying declaratively that a particular routine should run in a separate thread.

For example:

import asyncio
from aiopyramid.helpers import use_executor

class DatabaseUtilies:

    @use_executor  # query_it is now a coroutine
    def query_it():
        # some code that blocks


If you are using the default authorization policy, then you will generally not need to make any modifications to authorize users with Aiopyramid. The exception is if you want to use a callable that performs some io for your __acl__. In that case you will simply need to use a synchronized coroutine so that the authorization policy can call your coroutine like a normal Python function during view lookup.

For example:

import asyncio

from aiopyramid.helpers import synchronize

class MyResource:
    This resource uses a callable for it's
    __acl__ that accesses the db.

    # this
    __acl__ = synchronize(my_coroutine)

    # or this

    def __acl__(self):

    # will work

If you are using a custom authorization policy, most likely it will work with Aiopyramid in the same fashion, but it is up to you to guarantee that it does.


Authentication poses a problem because the interface for authentication policies uses normal Python methods that the framework expects to call noramlly but at the same time it is usually necessary to perform some io to retrieve relevant information. The built-in authentication policies generally accept a callback function that delegates retrieving principals to the application, but this callback function is also expected to be called in the regular fashion. So, it is necessary to use a synchronized coroutine as a callback function.

The final problem is that synchronized coroutines are expected to be called from within a child greenlet, or in other words from within framework code (see Architecture). However, it is often the case that we will want to access the policy through pyramid.request.Request.authenticated_userid or by calling remember(), etc. from within another coroutine such as a view callable.

To handle both situations, Aiopyramid provides tools for wrapping a callback-based authentication policy to work asynchronously. For example, the following code in your app constructor will allow you to use a coroutine as a callback.

from pyramid.authentication import AuthTktAuthenticationPolicy
from aiopyramid.auth import authn_policy_factory

from .myauth import get_principals


# In the includeme or constructor
authentication = authn_policy_factory(

Relevant authentication tools will now return a coroutine when called from another coroutine, so you would access the authentication policy using yield from in your view callable since it performs io.

from pyramid.security import remember, forget


# in some coroutine

maybe = yield from request.unauthenticated_userid
checked = yield from request.authenticated_userid
principals = yield from request.effective_principals
headers = yield from remember(request, 'george')
fheaders = yield from forget(request)


If you don’t perform asynchronous io or wrap the authentication policy as above, then don’t use yield from in your view. This approach only works for coroutine views. If you have both coroutine views and legacy views running in an executor, you will probably need to write a custom authentication policy.


Pyramid allows you to write tweens which wrap the request/response chain. Most existing tweens expect those tweens above and below them to run synchronously. Therefore, if you have a tween that needs to run asynchronously (e.g. it looks up some data from a database for each request), then you will need to write that tween so that it can wait without other tweens needing to explicitly yield from it. For example:

import asyncio

from aiopyramid.helpers import synchronize

def coroutine_logger_tween_factory(handler, registry):
    Example of an asynchronous tween that delegates
    a synchronous function to a child thread.
    This tween asynchronously logs all requests and responses.

    # We use the synchronize decorator because we will call this
    # coroutine from a normal python context
    # this is a coroutine
    def _async_print(content):
        # print doesn't really need to be run in a separate thread
        # but it works for demonstration purposes

        yield from asyncio.get_event_loop().run_in_executor(

    def coroutine_logger_tween(request):
        # The following calls are guaranteed to happen in order
        # but they do not block the event loop

        # print the request on the aio event loop
        # without needing to say yield
        # at this point,
        # other coroutines and requests can be handled

        # get response, this should be done in this greenlet
        # and not as a coroutine because this will call
        # the next tween and subsequently yield if necessary
        response = handler(request)

        # print the response on the aio event loop

        # return response after logging is done
        return response

    return coroutine_logger_tween


When using Pyramid’s traversal view lookup, it is often the case that you will want to make some io calls to a database or storage when traversing via __getitem__. When using the default traverser, Pyramid will call __getitem__ as a normal Python function. Therefore, it is necessary to synchronize __getitem__ on any asynchronous resources like so:

import asyncio

from aiopyramid.helpers import synchronize

class MyResource:
    """ This resource performs some asynchronous io. """

    __name__ = "example"
    __parent__ = None

    def __getitem__(self, key):
        yield from self.example_coroutine()
        return self  # no matter the path, this is the context

    def example_coroutine(self):
        yield from asyncio.sleep(0.1)
        print('I am some async task.')


Aiopyramid supports both asynchronous gunicorn and the uWSGI asyncio plugin.

Example gunicorn config:

use = egg:gunicorn#main
host =
port = 6543
worker_class = aiopyramid.gunicorn.worker.AsyncGunicornWorker

Example uWSGI config:

http-socket =
workers = 1
plugins =
    asyncio = 50

For those setting up Aiopyramid on a Mac, Ander Ustarroz’s tutorial may prove useful. Rickert Mulder has also provided a fork of uWSGI that allows for quick installation by running pip install git+git://github.com/circlingthesun/uwsgi.git in a virtualenv.


Aiopyramid provides additional view mappers for handling websocket connections with either gunicorn or uWSGI. Websockets with gunicorn use the websockets library whereas uWSGI has native websocket support. In either case, the interface is the same.

A function view callable for a websocket connection follows this pattern:

def websocket_callable(ws):
    # do stuff with ws

The ws argument passed to the callable has three methods for communicating with the websocket recv(), send(), and close() methods, which correspond to similar methods in the websockets library. A websocket connection that echoes all messages using gunicorn would be:

from pyramid.view import view_config
from aiopyramid.websocket.config import WebsocketMapper

@view_config(route_name="ws", mapper=WebsocketMapper)
def echo(ws):
    while True:
        message = yield from ws.recv()
        if message is None:
        yield from ws.send(message)

Aiopyramid also provides a view callable class WebsocketConnectionView that has on_message(), on_open(), and on_close() callbacks. Class-based websocket views also have a send() convenience method, otherwise the underyling ws may be accessed as self.ws. Simply extend WebsocketConnectionView specifying the correct view mapper for your server either via the __view_mapper__ attribute or the view_config decorator. The above example could be rewritten in a larger project, this time using uWSGI, as follows:

from pyramid.view import view_config
from aiopyramid.websocket.view import WebsocketConnectionView
from aiopyramid.websocket.config import UWSGIWebsocketMapper

from myproject.resources import MyWebsocketContext

class MyWebsocket(WebsocketConnectionView):
    __view_mapper__ = UWSGIWebsocketMapper

class EchoWebsocket(MyWebsocket):

    def on_message(self, message):
        yield from self.send(message)

The underyling websocket implementations of uWSGI and websockets differ in how they pass on the WebSocket message. uWSGI always sends bytes even when the WebSocket frame indicates that the message is text, whereas websockets decodes text messages to str. Aiopyramid attempts to match the behavior of websockets by default, which means that it coerces messages from uWSGI to str where possible. To adjust this behavior, you can set the use_str flag to False, or alternatively to coerce websockets messages back to bytes, set the use_bytes flag to True:

# In your app constructor
from aiopyramid.websocket.config import WebsocketMapper

WebsocketMapper.use_bytes = True

uWSGI Special Note

Aiopyramid uses a special WebsocketClosed exception to disconnect a greenlet after a websocket has been closed. This exception will be visible in log ouput when using uWSGI. In order to squelch this message, wrap the wsgi application in the ignore_websocket_closed() middleware in your application’s constructor like so:

from aiopyramid.websocket.helpers import ignore_websocket_closed

app = config.make_wsgi_app()
return ignore_websocket_closed(app)