معرفی شرکت ها
celery-message-consumer-1.2.1
تبلیغات ما
مشتریان به طور فزاینده ای آنلاین هستند. تبلیغات می تواند به آنها کمک کند تا کسب و کار شما را پیدا کنند.
مشاهده بیشتر
تبلیغات ما
مشتریان به طور فزاینده ای آنلاین هستند. تبلیغات می تواند به آنها کمک کند تا کسب و کار شما را پیدا کنند.
مشاهده بیشتر
تبلیغات ما
مشتریان به طور فزاینده ای آنلاین هستند. تبلیغات می تواند به آنها کمک کند تا کسب و کار شما را پیدا کنند.
مشاهده بیشتر
تبلیغات ما
مشتریان به طور فزاینده ای آنلاین هستند. تبلیغات می تواند به آنها کمک کند تا کسب و کار شما را پیدا کنند.
مشاهده بیشتر
تبلیغات ما
مشتریان به طور فزاینده ای آنلاین هستند. تبلیغات می تواند به آنها کمک کند تا کسب و کار شما را پیدا کنند.
مشاهده بیشترتوضیحات
Tool for using the bin/celery worker to consume vanilla AMQP messages (i.e. not Celery tasks)
| ویژگی | مقدار |
|---|---|
| سیستم عامل | - |
| نام فایل | celery-message-consumer-1.2.1 |
| نام | celery-message-consumer |
| نسخه کتابخانه | 1.2.1 |
| نگهدارنده | [] |
| ایمیل نگهدارنده | [] |
| نویسنده | Depop |
| ایمیل نویسنده | dev@depop.com |
| آدرس صفحه اصلی | https://github.com/depop/celery-message-consumer |
| آدرس اینترنتی | https://pypi.org/project/celery-message-consumer/ |
| مجوز | Apache 2.0 |
celery-message-consumer
=======================
|PyPI Version| |Build Status|
.. |PyPI Version| image:: http://img.shields.io/pypi/v/celery-message-consumer.svg?style=flat
:target: https://pypi.python.org/pypi/celery-message-consumer/
:alt: Latest PyPI version
.. |Build Status| image:: https://circleci.com/gh/depop/celery-message-consumer.svg?style=shield&circle-token=a9ea2909c5cbc4cb32a87f50444ca79b99e3b09c
:alt: Build Status
Tool for using the ``bin/celery`` worker to consume vanilla AMQP
messages (i.e. not Celery tasks)
While `writing a simple consumer
script <https://medium.com/python-pandemonium/building-robust-rabbitmq-consumers-with-python-and-kombu-part-1-ccd660d17271>`__
using Kombu can be quite easy, the Celery worker provides many features
around process pools, queue/routing connections etc as well as being
known to run reliably over long term.
It seems safer to re-use this battle-tested consumer than try to write
our own and have to learn from scratch all the ways that such a thing
can fail.
Usage
-----
.. code:: bash
pip install celery-message-consumer
Handlers
~~~~~~~~
In your code, you can define a message handler by decorating a python
function, in much the same way as you would a Celery task:
.. code:: python
from event_consumer import message_handler
@message_handler('my.routing.key')
def process_message(body):
# `body` has been deserialized for us by the Celery worker
print(body)
@message_handler(['my.routing.key1', 'my.routing.key2'])
def process_messages(body):
# you can register handler for multiple routing keys
@message_handler('my.routing.*')
def process_all_messages(body):
# or wildcard routing keys, if using a 'topic' exchange
Like a Celery task, the module it is defined in must actually get
imported at some point for the handler to be registered.
A queue (in fact, three queues - see below) will be created to receive
messages matching the routing key.
Celery
~~~~~~
Elsewhere in your code you will need to instantiate a Celery app and
apply our custom 'ConsumerStep' which hooks our message handlers into
the worker. If you are already using Celery *as Celery* in your project
then you probably want separate Celery apps for tasks and for the
message consumer.
.. code:: python
from celery import Celery
from event_consumer.handlers import AMQPRetryConsumerStep
main_app = Celery()
consumer_app = Celery()
consumer_app.steps['consumer'].add(AMQPRetryConsumerStep)
You likely will want separate config for each app. See
`Celery docs <http://docs.celeryproject.org/en/latest/userguide/application.html#configuration>`__.
In the config for your message consumer app, add the modules containing
your decorated message handler functions to ``CELERY_IMPORTS``, exactly
as you would for Celery tasks - this ensures they get imported and
registered when the worker starts up.
Then from the command-line, run the Celery worker just like you usually
would, attaching to the consumer app:
.. code:: bash
bin/celery worker -A myproject.mymodule:consumer_app
Configuration
~~~~~~~~~~~~~
Settings are intended to be configured primarily via a python file, such
as your existing Django ``settings.py`` or Celery ``celeryconfig.py``.
To bootstrap this, there are a couple of env vars to control how config
is loaded:
- ``EVENT_CONSUMER_APP_CONFIG``
should be an import path to a python module, for example:
``EVENT_CONSUMER_APP_CONFIG=django.conf.settings``
- ``EVENT_CONSUMER_CONFIG_NAMESPACE``
Sets the prefix used for loading further config values from env and
config file. Defaults to ``EVENT_CONSUMER``.
See source of ``event_consumer/conf/`` for more details.
Some useful config keys (all of which are prefixed with
``EVENT_CONSUMER_`` by default):
- ``SERIALIZER`` this is the name of a Celery serializer name, e.g.
``'json'``. The consumer will only accept messages serialized in this
format.
- ``QUEUE_NAME_PREFIX`` if using default queue name (routing-key) then
this prefix will be added to the queue name. If you supply a custom
queue name in the handler decorator the prefix will not be applied.
- ``MAX_RETRIES`` defaults to ``4`` (i.e. 1 attempt + 4 retries = 5
strikes)
- ``BACKOFF_FUNC`` takes a function ``(int) -> float`` which returns
the retry delay (in seconds) based on current retry counter for the
message.
- ``ARCHIVE_EXPIRY`` time in milliseconds to keep messages in the
"archive" queue, after which the exchange will delete them. Defaults
to 24 days.
- ``USE_DJANGO`` set to ``True`` if your message handler uses the
Django db connection, so that the worker is able to cope with the
dreaded *"current transaction is aborted"* error and continue.
- ``EXCHANGES`` if you need your message handlers to connect their
queues to specific exchanges then you can provide a dict like:
.. code:: python
EXCHANGES = {
# a reference name for this config, used when attaching handlers
'default': {
'name': 'data', # actual name of exchange in RabbitMQ
'type': 'topic', # an AMQP exchange type
},
'other': {
...
},
...
}
The ``'default'`` config will be used... by default. You can attach
handler to a specific exchange when decorating:
.. code:: python
@message_handler('my.routing.key', exchange='other')
def process_message(body):
pass
Queue layout
------------
While all of the broker, exchange and queue naming is configurable (see
source code) this project implements a *very specific queue pattern*.
Briefly: for each routing key it listens to, the consumer sets up
*three* queues and a 'dead-letter exchange' (DLX).
#. The "main" message queue
#. If any unhandled exceptions occur, and we have retried less than
``settings.MAX_RETRIES``, the message will be put on the "retry"
queue with a TTL. After the TTL expires, the DLX will put the message
back on the main queue.
#. If all retries are exhausted (or ``PermanentFailure`` is raised) then
the consumer will put the message on the "archive" queue. This gives
opportunity for someone to manually retry the archived messages,
perhaps after a code fix has been deployed.
| You will of course note that this is *totally different and separate*
from Celery's own ``task.retry`` mechanism.
| **Pros:** matches pattern we were already using for non-Celery,
non-Python apps, "archive" queue provides an extra safety net.
| **Cons:** Relies on RabbitMQ-specific feature, more queues (more
complicated).
Compatibility
-------------
Python 2.7 and 3.6-3.8 are both supported.
**Only** RabbitMQ transport is supported.
We depend on Celery and Kombu. Their versioning seems to be loosely in
step so that Celery 3.x goes with Kombu 3.x and Celery 4.x goes with
Kombu 4.x. We test against both v3 and v4.
Django is not required, but when used we have some extra integration
which is needed if your event handlers use the Django db connection.
This must be enabled if required via the ``settings.USE_DJANGO`` flag.
This project is tested against:
=========== ============ ============= ================== ==================
x Django 1.11 Django 2.2 Celery/Kombu 3.x Celery/Kombu 4.x
=========== ============ ============= ================== ==================
Python 2.7 * * *
Python 3.6 * * * *
Python 3.7 * * *
Python 3.8 * * *
=========== ============ ============= ================== ==================
Running the tests
-----------------
CircleCI
~~~~~~~~
| The easiest way to test the full version matrix is to install the
CircleCI command line app:
| https://circleci.com/docs/2.0/local-jobs/
| (requires Docker)
The cli does not support 'workflows' at the moment so you have to run
the two Python version jobs separately:
.. code:: bash
circleci build --job python-2.7
.. code:: bash
circleci build --job python-3.6
py.test (single combination of dependency versions)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
It's also possible to run the tests locally, allowing for debugging of
errors that occur.
We rely on some RabbitMQ features for our retry queues so we need a
rabbit instance to test against. A ``docker-compose.yml`` file is
provided.
.. code:: bash
docker-compose up -d
export BROKER_HOST=$(docker-machine ip default)
(adjust the last line to suit your local Docker installation)
The ``rabbitmqadmin`` web UI is available to aid in debugging queue issues:
.. code:: bash
http://{BROKER_HOST}:15672/
Now decide which version combination from the matrix you're going to
test and set up your virtualenv accordingly:
.. code:: bash
pyenv virtualenv 3.6.2 celery-message-consumer
You will need to edit ``requirements.txt`` and ``requirements-test.txt``
for the specific versions of dependencies you want to test against. Then
you can install everything via:
.. code:: bash
pip install -r requirements-test.txt
Set an env to point to the target Django version's settings in the test
app (for Django-dependent tests) and for general app settings:
.. code:: bash
export DJANGO_SETTINGS_MODULE=test_app.dj111.settings
export EVENT_CONSUMER_APP_CONFIG=test_app.settings
Now we can run the tests:
.. code:: bash
PYTHONPATH=. py.test -v -s --pdb tests/
tox (all version combinations for current Python)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
You'll notice in the CircleCI config we run tests against the matrix
dependency versions using ``tox``.
There are `some warts <https://github.com/pyenv/pyenv-virtualenv/issues/202#issuecomment-339624649>`__
around using ``tox`` with ``pyenv-virtualenv`` so if you created a Python 3.6
virtualenv using the instructions above the best thing to do is delete it and
recreate it like this:
.. code:: bash
pyenv virtualenv -p python3.6 myenv
pip install tox
(it's actually easier not to use a virtualenv at all - tox creates its
own virtualenvs anyway, but that does mean you'd have to install tox
globally)
You need the Docker container running:
.. code:: bash
docker-compose up -d
export BROKER_HOST=$(docker-machine ip default)
You can now run tests for any versions compatible with your virtualenv
python version, e.g.
.. code:: bash
tox -e py36-dj111-cel4
To run the full version matrix you need to have both Python 2.7 and 3.6. The
easiest way is via ``pyenv``. You will also need to make both Python versions
'global' (or 'local') via pyenv, and then install and run ``tox`` outside of
any virtualenv.
.. code:: bash
source deactivate
pyenv global 2.7.14 3.6.2
pip install tox
tox
نحوه نصب
نصب پکیج whl celery-message-consumer-1.2.1:
pip install celery-message-consumer-1.2.1.whl
نصب پکیج tar.gz celery-message-consumer-1.2.1:
pip install celery-message-consumer-1.2.1.tar.gz