Skip to main content

Twisted based pipeline framework for AMQP

Project description

https://travis-ci.org/Fatal1ty/amqpipe.svg?branch=master Requirements Status https://img.shields.io/pypi/v/amqpipe.svg https://img.shields.io/pypi/pyversions/amqpipe.svg https://img.shields.io/badge/license-MIT-blue.svg

Twisted based pipeline framework for AMQP. It allow you to create fast asynchronous services which follow ideology:

  • get message from queue

  • doing something with message

  • publish some result

Installation

Install via pip:

pip install amqpipe

Basic usage

The minimal module based on AMQPipe is:

from amqpipe import AMQPipe

pipe = AMQPipe()
pipe.run()

It will simply get all messages from one RabbitMQ queue and publish them to other RabbitMQ exchange.

Now we define some action on messages:

import hashlib
from amqpipe import AMQPipe

def action(message):
    return hashlib.md5(message).hexdigest()

pipe = AMQPipe(action=action)
pipe.run()

It will publish md5 checksum for every message as result.

If messages in input queue are in predefined format then you can define converter-function:

import hashlib
from amqpipe import AMQPipe

def converter(message):
    return message['text']

def action(text):
    return hashlib.md5(text).hexdigest()

pipe = AMQPipe(converter=converter, action=action)
pipe.run()

You can define service-specific arguments:

import hashlib
from amqpipe import AMQPipe

class Processor:
    def set_field(self, field):
        self.field = field

processor = Processor()

def init(args):
    processor.set_field(args.field)

def converter(message):
    return message.get(processor.field)

def action(text):
    return hashlib.md5(text).hexdigest()

pipe = AMQPipe(converter, action, init)
pipe.parser.add_argument('--field', default='text', help='Field name for retrieving message value')
pipe.run()

You can connect to database in init function or do some other things for initialization.

If your action returns Deferred then result would be published to RabbitMQ when this Deferred will be resolved:

import logging
from twisted.internet import defer
from amqpipe import AMQPipe

logger = logging.getLogger(__name__)

class Processor:
    def set_field(self, field):
        self.field = field

processor = Processor()

def init(args):
    connect_to_db()
    ...

def converter(message):
    return message.get(processor.field)

@defer.inlineCallbacks
def action(text):
    result = yield db_query(text)
    logger.info('Get from db: %s', result)
    defer.returnValue(result)

pipe = AMQPipe(converter, action, init)
pipe.parser.add_argument('--field', default='text', help='Field name for retrieving message value')
pipe.run()

Init function may return Deferred too.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

amqpipe-1.2.tar.gz (6.4 kB view details)

Uploaded Source

File details

Details for the file amqpipe-1.2.tar.gz.

File metadata

  • Download URL: amqpipe-1.2.tar.gz
  • Upload date:
  • Size: 6.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No

File hashes

Hashes for amqpipe-1.2.tar.gz
Algorithm Hash digest
SHA256 9a2862c0a9e317cf1a0f2b1de7e68b16a7b8778df38f384f424e934bb4d96f33
MD5 31869a576afc9ecb8fde5593b1a7e6da
BLAKE2b-256 9eb3949ee1fe77f1490f9b971a493feddfacb32de9ec98f6ad2a24f3055c17f6

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page