(Tornado) Convert a callback-based async function to a coroutine, configuring the position/keyword the callback is expected. This can be used, for example, to convert the callback-based interface of the pika.TornadoConnection to a coroutine-based one, which is the standard async pattern suggested by Tornado nowadays.

coroutine_maker.py

from functools import wraps
from tornado.concurrent import return_future


class CoroutineMaker(object):
    """ Convert a callback-based async function to a coroutine.

    The Tornado utility gen.Task is quite useful to convert callback-based
    async functions to coroutines, but has an important drawback: it expects
    the callback to be passed to the function in a keyword parameter
    called 'callback'. This limits the versatility of this utility.

    This CoroutineMaker allows you to configure each function you will
    convert to a coroutine, telling for each one which parameter will
    expect the callback. It can be expected on a keyword parameter, in which
    case you configure it by giving the name of the parameter; or even on a
    position parameter, in which case you configure it by giving the position
    at which the parameter is expected.

    Take for instance this example, that uses the CoroutineMaker to convert
    the pika.TornadoConnection callback-based interface to a coroutine one:

    >>> coroutiner = CoroutineMaker({
        'queue_bind': 0,
        'queue_declare': 0,
        'TornadoConnection': 'on_open_callback',
        'channel': 'on_open_callback'})

    >>> connection = yield coroutiner.convert(pika.TornadoConnection)()
    >>> channel = yield coroutiner.convert(connection.channel)()
    >>> method_frame = yield coroutiner.convert(channel.queue_declare)('queue')
    >>> method_frame = yield coroutiner.convert(channel.queue_bind)('queue',
                                                                    'exchange',
                                                                    'to.pic')

    The configuration is straighforward: you give a dictionary that maps the
    name of the function with the parameter that expects the callback. For
    functions that expect the callback in a positional parameter, give the
    position (integer, starting from 0) at witch the callback is expected.
    For functions that expect a keyword argument, give the name of the
    parameter, with a string. 
    
    For convenience, this can also be used in functions that aren't previously
    configured, in which case it behaves just like gen.Task or @return_future, 
    expecting the callback in a keyword parameter named 'callback'.
    """
    def __init__(self, config):
        self._signatures = {}
        self._parse_config(config)

    def convert(self, func):
        """ Convert a callback-based async function into a coroutine.

        If the function has a configuration available, uses it. Otherwise,
        falls back to the default 'callback' keyword parameter as the
        callback parameter.
        """
        @return_future
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                args, kwargs = self._signatures[func.__name__](args, kwargs)
            except KeyError:
                # Function not configured. The default @return_future
                # behaviour will apply.
                pass
            func(*args, **kwargs)
        return wrapper

    def _parse_config(self, config):
        for function, param in config.items():
            if isinstance(param, int):
                # Positional parameter
                self._signatures[function] = self._positional_signature(param)
            elif isinstance(param, basestring):
                # keyword parameter
                self._signatures[function] = self._keyword_signature(param)
            else:
                raise Exception('Wrong Configuration')

    def _positional_signature(self, position):
        # pop the 'callback' kwarg that puts Tornado and insert it in the args,
        # at the configured position.
        def signature(args, kwargs):
            callback = kwargs.pop('callback')
            args = list(args)
            args.insert(position, callback)
            return args, kwargs
        return signature

    def _keyword_signature(self, keyword):
        # Just substitute the default 'callback' kwarg for the
        # configured keyword.
        def signature(args, kwargs):
            kwargs[keyword] = kwargs.pop('callback')
            return args, kwargs
        return signature

usage_example.py

import pika
from pika.adapters import TornadoConnection
from tornado import gen
from tornado.ioloop import IOLoop

from coroutine_maker import CoroutineMaker


def consume_message(channel, basic_deliver, properties, body):
    """ Called on message received """
    print "Message Received"
    channel.basic_ack(basic_deliver.delivery_tag)


@gen.coroutine
def setup_consumer():
    """ Setup the pika consumer asyncrhonously.

    Using the CoroutineMaker, the asynchronicity is achieved yielding
    the arguments that would had been passed to the callbacks in a
    normal pika setup.

    """
    QUEUE= 'queue_name'
    TOPIC = 'to.pic'
    EXCHANGE = 'exchange'

    coroutiner = CoroutineMaker({
        'queue_bind': 0,
        'queue_declare': 0,
        'exchange_declare': 0,
        'channel': 'on_open_callback',
        'TornadoConnection': 'on_open_callback'})

    connection = yield coroutiner.convert(TornadoConnection)()
    channel = yield coroutiner.convert(connection.channel)()
    yield coroutiner.convert(channel.exchange_declare)(EXCHANGE, 'topic')
    yield coroutiner.convert(channel.queue_declare)(QUEUE)
    yield coroutiner.convert(channel.queue_bind)(QUEUE, EXCHANGE, TOPIC)
    channel.basic_consume(consume_message, QUEUE)
    print "Pika Consumer setup complete"


if __name__ == '__main__':
    setup_consumer()
    IOLoop.instance().start()