New zocalo.util.rabbitmq.RabbitMQAPI

.pre-commit-config.yaml

@@ -37,3 +37,4 @@ repos:
   hooks:
   - id: mypy
     files: 'src/.*\.py$'
+    additional_dependencies: [types-requests]

HISTORY.rst

@@ -5,13 +5,15 @@ History
 Unreleased
 ----------
 * Add command line tools for handling dead-letter messages
-* ``zocalo.dlq_check`` checks dead-letter queues for messages 
-* ``zocalo.dlq_purge`` removes messages from specified DLQs and dumps them to a directory 
-  specified in the Zocalo configuration 
-* ``zocalo.dlq_reinject`` takes a serialised message produced by ``zocalo.dlq_purge`` and 
+* ``zocalo.dlq_check`` checks dead-letter queues for messages
+* ``zocalo.dlq_purge`` removes messages from specified DLQs and dumps them to a directory
+  specified in the Zocalo configuration
+* ``zocalo.dlq_reinject`` takes a serialised message produced by ``zocalo.dlq_purge`` and
   places it back on a queue
-* Use ``argparse`` for all command line tools and make use of ``workflows`` transport 
+* Use ``argparse`` for all command line tools and make use of ``workflows`` transport
   argument injection. Minimum ``workflows`` version is now 2.14
+* New ``zocalo.util.rabbitmq.RabbitMQAPI()`` providing a thin wrapper around the
+  RabbitMQ HTTP API
 
 0.10.0 (2021-10-04)
 -------------------

requirements_dev.txt

@@ -1,8 +1,11 @@
 PyYAML==6.0
 graypy==2.1.0
 marshmallow==3.14.0
+pydantic
 pytest-cov==3.0.0
 pytest-mock
 pytest==6.2.5
+requests
+requests_mock
 setuptools==58.3.0
 workflows==2.14

setup.cfg

@@ -30,6 +30,8 @@ install_requires =
     PyYAML
     graypy>=1.0
     marshmallow
+    requests
+    pydantic
     setuptools
     workflows>=2.14
 packages = find:

src/zocalo/cli/dlq_check.py

@@ -1,12 +1,10 @@
 import argparse
-import json
-import urllib
 
 import workflows.transport
 
 import zocalo.configuration
 from zocalo.util.jmxstats import JMXAPI
-from zocalo.util.rabbitmq import http_api_request
+from zocalo.util.rabbitmq import RabbitMQAPI
 
 #
 # zocalo.dlq_check
@@ -50,16 +48,14 @@ def extract_queue_name(namestring):
 def check_dlq_rabbitmq(
     zc: zocalo.configuration.Configuration, namespace: str = None
 ) -> dict:
-    _api_request = http_api_request(zc, "/queues")
-    with urllib.request.urlopen(_api_request) as response:
-        reply = response.read()
-    queue_info = json.loads(reply)
-    dlq_info = {}
-    for q in queue_info:
-        if q["name"].startswith("dlq."):
-            if (namespace is None or q["vhost"] == namespace) and int(q["messages"]):
-                dlq_info[q["name"]] = int(q["messages"])
-    return dlq_info
+    rmq = RabbitMQAPI.from_zocalo_configuration(zc)
+    return {
+        q.name: q.messages
+        for q in rmq.queues()
+        if q.name.startswith("dlq.")
+        and (namespace is None or q.vhost == namespace)
+        and q.messages
+    }
 
 
 def run() -> None:

src/zocalo/cli/dlq_reinject.py

@@ -15,7 +15,7 @@
 import workflows.transport
 
 import zocalo.configuration
-from zocalo.util.rabbitmq import http_api_request
+from zocalo.util.rabbitmq import RabbitMQAPI
 
 
 def run() -> None:
@@ -130,12 +130,8 @@ def run() -> None:
             header = dlqmsg["header"]
             exchange = header.get("headers", {}).get("x-death", {})[0].get("exchange")
             if exchange:
-                import urllib
-
-                _api_request = http_api_request(zc, "/queues")
-                with urllib.request.urlopen(_api_request) as response:
-                    reply = response.read()
-                exchange_info = json.loads(reply)
+                rmqapi = RabbitMQAPI.from_zocalo_configuration(zc)
+                exchange_info = rmqapi.get("queues").json()
                 for exch in exchange_info:
                     if exch["name"] == exchange:
                         if exch["type"] == "fanout":