Skip to main content

Simplified PostgreSQL client built upon Psycopg2

Project description

An opinionated `Rejected consumer <>`_ class that
automatically decodes messages sent as `Avro <>`_

For applications that can share schema files, Avro datum provide small, contract
based binary serialization format. Leveraging AMQP's ``Type`` message property
to convey the Avro schema file for decoding the datum, the ``DatumFileSchemaConsumer``
and ``DatumConsulSchemaConsumer`` classes extend Rejected's
``rejected.consumer.Consumer`` class adding automated deserialization of AMQP
messages serialized as Avro datums.

|Version| |Downloads| |License|

avroconsumer is available on the `Python package index <>`_.


To use the ``DatumFileSchemaConsumer``, first you need to set the ``schema_path``
configuration setting in the rejected configuration file. The following snippet
demonstrates an example configuration:

.. code:: yaml

consumer: app.Consumer
connections: [rabbit1]
qty: 1
queue: datum
qos_prefetch: 1
ack: True
max_errors: 5
schema_path: /etc/avro-schemas/

If messages are published with a AMQP ``type`` message property of ``foo`` and
a ``content-type`` property of ``application/vnd.apache.avro.datum``, the
``DatumConsumer`` will use the Avro schema file ``/etc/avro-schemas/foo.avsc``
to deserialize messages

The following example code shows how implement the ``DatumConsumer``.

.. code:: python

import avroconsumer

class MyConsumer(avroconsumer.DatumFileSchemaConsumer):

def process(self):
LOGGER.debug('Decoded avro datum data: %r', self.body)

As with any instance of ``rejected.consumer.Consumer``, the
``avroconsumer.DatumFileSchemaConsumer`` can automatically rejected messages
based upon the ``type`` message property. Simply set the ``MESSAGE_TYPE`` attribute
of your consumer and any messages received that do not match that message type
will be rejected. The following example demonstrates setting the message type:

.. code:: python

import avroconsumer

class MyConsumer(avroconsumer.DatumConsumer):


def process(self):
LOGGER.debug('Decoded avro datum data: %r', self.body)

The ``avroconsumer.DatumConsulSchemaConsumer`` is an opinionated wrapper that
loads versioned schema files from a `consul <>`_ Key/Value
database. The schema files should be stored as keys in the format
``/schemas/avro/<schema-name>/<schema-version>.avsc``. When messages are sent
to consumers extending ``avroconsumer.DatumConsulSchemaConsumer`` the AMQP `
``type`` message property should be in ``<schema-name>.<schema-version>`` format.

You can alter which Consul server is used to retrieve the schema by setting
the ``CONSUL_HOST`` and ``CONSUL_PORT`` environment variables. They default
to ``localhost`` and ``8500`` respectively.

- `avro <>`_
- `rejected <>`_

.. |Version| image::

.. |Downloads| image::

.. |License| image::

Project details

Download files

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

Files for avroconsumer, version 0.3.0
Filename, size File type Python version Upload date Hashes
Filename, size avroconsumer-0.3.0.tar.gz (5.1 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page