landregistry-rabbitmq-utilities 0.9.9
Standardised logging configuration for HMLR applications
License: MIT
Requires Python: >=3.9

# RabbitMQ Utiltiies

Utilities and helpers for interacting with RabbitMQ.

## Installation

```shell
pip3 install --index-url https://nexus.devops.stp.hmlr.zone/repository/pypi/simple landregistry-rabbitmq-utilities
```

This package depends on:

* Python (>= 3.9)
* Kombu (>= 5.0.0)
* Requests (>= 2.28.0)


## Helper methods

**broker_url** *(hostname, userid, password, port=DEFAULT_PORT, virtual_host=DEFAULT_VHOST, protocol=DEFAULT_PROTOCOL)*

Generates a broker URL from the upplied parameters

**publish_message** *(message, rabbit_url, exchange_name, routing_key, queue_name=None,exchange_type='direct', serializer="json", headers=None)*

Convenience wrapper for sending a single message.

```python
url = broker_url('localhost', 'guest', 'guest')
message = {'content': 'test message'}
publish_message(message, url, 'exchange', 'key')
```

**get_queue_count** *(rabbit_url, queue_name)*

Get the number of messages on the named queue.


## Worker

Helper class for writing consumers.

Usage example:

```python
queues = [
    Queue(
        config.QUEUE_NAME,
        exchange,
        bindings=[binding(exchange, routing_key=key) for key in config.ROUTING_KEYS]
    )
]

with Connection(config.RABBIT_URL, heartbeat=4) as conn:
    try:
        worker = Worker(conn, queues, process_message_callback, 'server')
        worker.run()
    except KeyboardInterrupt:
        logger.debug('KeyboardInterrupt')
    except Exception as e:
        logger.exception('Unhandled Exception: %s', repr(e))
```

and:

```python
def process_message_callback(body, message, logger, requests):
    try:
        # TODO add code here for processing the message.
        message.ack()
    except ConnectionError:
        logger.info("Connection was reset by RabbitMQ server. Raising this error so that the connection gets revived.")
        raise
    except Exception as e:
        # Decide what is needed here i.e. send the message to an error queue.
        # message.ack()  # maybe
        logger.exception("Error processing message - {}".format(repr(e)))
```

### Specifying prefetch_count

The `Worker` constructor takes an optional `prefetch_count` parameter.

```python
worker = Worker(conn, queues, process_message_callback, 'server', prefetch_count=1)
```


## Emitter

Helper class for writing robust producers.

An instance of this class will connect to the given AMQP broker,
and allow you to send messages to an exchange on that broker with
a given routing key, optionally also declaring a queue and and
binding it to the exchange with that routing key.  Upon connection
errors (either upon initial connection or message send) it will
retry several times before giving up, and if a connection is
interrupted it will recover gracefully on the next attempt to send
a message.

### Short-lived connections

To create a short-lived connection which automatically
releases/disconnects when you've finished, just use an instance of
this class as a context manager; for example:

```python
url = broker_url('guest', 'guest', 'localhost')
with Emitter(url, 'ex_name', 'key', logger_name=logger.name) as emitter:
    emitter.send_message({'status': 'success'}, serializer='json')
```

Note the lack of any explicit `connect()` call here:
`send_message()` calls it if it hasn't been called already, so you
never need to call it explicitly (though you can).

Note also the lack of any queue_name parameter in the Emitter
constructor call: the message will be sent to the given exchange
with the given routing key, but no queue declaration/binding takes
place here (compare with the next example).

### Long-lived connections

Alternatively, you can just create an instance of this class and
have it live for as long as you need it, in which case the
connection will remain open until the instance falls out of scope
or you call its `release()` method.

To have an instance live for the lifetime of an application, a
helpful pattern is to put it in a module-level global and lazily
declare it on first use, e.g.::

```python
# Module-level global for the instance; initially nothing here.
_emitter = None

def send_message(msg, serializer):
    global _emitter
    if _emitter is None:
        url = broker_url('guest', 'guest', 'localhost')
        _emitter = Emitter(url, 'ex_name', 'key', 'q_name', logger_name=logger.name)
    _emitter.send_message(msg, serializer)
```

Then you can just call this module's `send_message()` function
whenever you need to send a message: it will connect to the broker
on the first call, keep the connection open between calls, and
attempt to reconnect on the next call if the connection ever goes
away.

Note also that in this example we do pass a queue_name to the
Emitter constructor: upon connection, the Emitter will ensure that
the named queue is declared, and bound to the given exchange with
the given routing key.

### Customisation via subclassing

You may wish to use a custom subclass to override certain
behaviour around error handling.  In particular, you can override:

* Various `CONN_*` and `SEND_*` class variables, which control
    connection/send retry behaviour.  E.g. `SEND_MAX_RETRIES`
    defines how many times to retry sending a message before giving
    up.

* The `conn_errback()` callback method, which is called upon an
    error while connecting.

* The `send_errback()` callback method, which is called upon an
    error while sending a message.

* The `errback()` callback method, which (by default) is just what
    `conn_errback()` and `send_errback()` call - so override just this
    to change them both.  The default version just logs the error.

For example, here's a class which retries connecting forever;
waits up to 10 seconds between retries; only retries message sends
once; and logs the full traceback for connect errors:

```python
class CustomEmitter(Emitter):

    CONN_MAX_RETRIES = None
    CONN_INTERVAL_MAX = 10
    SEND_MAX_RETRIES = 1

    def conn_errback(self, ex, interval):
        logger.error('Error: %r', ex, exc_info=1)
        logger.info('Retry in %s seconds.', interval)
```

Note the following edge cases for `*_MAX_RETRIES`:

* `None` - for both `CONN_` and `SEND_` this means "try forever".

* 0 - for `CONN_` this means "try forever" but for `SEND_` this
    means "retry once", i.e. the same as `SEND_MAX_RETRIES = 1`;
    this seems to be a kombu bug.

* Negative values - for `CONN_` this means "just try once/no
    retries", but for `SEND_` it just breaks things (it will never
    recover the connection); this seems to be a kombu bug.


## Developing

If you're developing changes to this extension, you can mount it's local directory into an application so that changes
to the extension trigger an automatic reload.

Using the flask-skeleton-api as an example, and assuming you've checked the extension out to `<dev-env>/apps/lr-python-rabbitmq-utils`,
add the volume to your `docker-compose-fragment.yml` (be careful not to commit this change):

```diff
     volumes:
       - ./flask-skeleton-api:/src
+      - ./lr-python-rabbitmq-utils/landregistry/rabbitmq_utilities:/opt/app-root/lib/python3.9/site-packages/landregistry/rabbitmq_utilities

```

To run MyPy, run the command `mypy . --explicit-package-bases` from within the package root source directory (`blah/lr-python-rabbitmq-utilities`).


## Build and publication

The build and publish process is defined in the project's Jenkinsfile.

TAny commit to the master branch will be built and published by [Jenkins](https://jenkins-common-components.apps.development.test.stp.hmlr.zone/job/lr-flask-trace-id/). **As there is no concept of snapshots, and versions cannot be overwritten, every commit to master must have a unique version specified in `setup.py`!** It is therefore recommended that any development work is done in a branch until it is ready, then the version bumped and the branch merged into master.

The package is published to [Nexus' pypi-internal respository](https://nexus.devops.stp.hmlr.zone/#browse/browse:pypi-internal:landregistry-rabbitmq-utilities)