Set up ruff/lefthook

.github/workflows/testrunner.yaml

@@ -0,0 +1,25 @@
+name: Testrunner
+
+on:
+  pull_request:
+
+jobs:
+  lint:
+    uses: zeitonline/gh-action-workflows/.github/workflows/lefthook.yaml@a7aaf1e4ad49e558e917f8d283ee751c38f30c16 # 4.5.0
+    with:
+      python-version: "3.14"
+      setup-command: bin/build lefthook
+      baseproject-name: pypi-oss
+      baseproject-environment: production
+
+  test:
+    needs:
+      - lint
+    runs-on: zon-ubuntu-general-dind
+    permissions:
+      contents: read
+
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
+      - run: bin/test

.lefthook.yaml

@@ -0,0 +1,11 @@
+pre-commit:
+  jobs:
+    - name: ruff
+      glob: "*.py"
+      group:
+        piped: true
+        jobs:
+          - name: format
+            run: .venv/bin/ruff format --force-exclude --exit-non-zero-on-format {staged_files}
+          - name: check
+            run: .venv/bin/ruff check --force-exclude --fix --exit-non-zero-on-fix {staged_files}

bin/build

@@ -0,0 +1,31 @@
+#!/bin/bash
+
+set -e
+if [[ -n "$CI" ]]; then set -x; fi
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+
+
+function main() {
+    cd "$DIR/.."
+
+    local command=$1
+    case $command in
+        lefthook)
+            build_lefthook
+        ;;
+        *)
+            build_lefthook
+        ;;
+    esac
+}
+
+function build_lefthook() {
+    uv sync --frozen --only-group lint
+
+    if [[ -z "$CI" && ! -f .git/hooks/pre-commit ]]; then
+        lefthook install
+    fi
+}
+
+
+main "$@"

pyproject.toml

@@ -21,6 +21,7 @@ prometheus = "celery_redis_prometheus.exporter:main"
 
 [dependency-groups]
 dev = ["pytest"]
+lint = ["lefthook", "ruff"]
 release = ["towncrier"]
 
 
@@ -51,3 +52,18 @@ underlines = ["", ""]
 directory = "change"
 name = ""
 showcontent = true
+
+
+[tool.ruff]
+line-length = 100
+
+[tool.ruff.lint]
+select = ["E", "F", "I"]
+fixable = ["ALL"]
+
+[tool.ruff.format]
+quote-style = "single"
+
+[tool.ruff.lint.isort]
+from-first = true
+lines-after-imports = 2

src/celery_redis_prometheus/exporter.py

@@ -1,14 +1,15 @@
 from functools import wraps
-import celery.bin.base
-import click
+import _thread
 import collections
 import json
 import logging
-import prometheus_client
-import _thread
 import threading
 import time
 
+import celery.bin.base
+import click
+import prometheus_client
+
 
 log = logging.getLogger(__name__)
 
@@ -21,23 +22,23 @@
 
 
 STATS = {
-    'tasks': prometheus_client.Counter(
-        'celery_tasks_total', 'Number of tasks', ['queue', 'state']),
+    'tasks': prometheus_client.Counter('celery_tasks_total', 'Number of tasks', ['queue', 'state']),
     'queuetime': prometheus_client.Histogram(
-        'celery_task_queuetime_seconds', 'Task queue wait time', ['queue']),
+        'celery_task_queuetime_seconds', 'Task queue wait time', ['queue']
+    ),
     'runtime': prometheus_client.Histogram(
-        'celery_task_runtime_seconds', 'Task runtime', ['queue']),
-    'queues': prometheus_client.Gauge(
-        'celery_queue_length', 'Queue length', ['queue'])
+        'celery_task_runtime_seconds', 'Task runtime', ['queue']
+    ),
+    'queues': prometheus_client.Gauge('celery_queue_length', 'Queue length', ['queue']),
 }
 
 
 @click.command(name='prometheus', cls=celery.bin.base.CeleryCommand)
 @click.option('--host', default='0.0.0.0', help='Listen host')
 @click.option('--port', default=9691, help='Listen port')
 @click.option(
-    '--queuelength-interval', default=0,
-    help='Check queue lengths every x seconds (0=disabled)')
+    '--queuelength-interval', default=0, help='Check queue lengths every x seconds (0=disabled)'
+)
 @click.option('--verbose', is_flag=True, help='Enable debug logging')
 @click.pass_context
 def main(ctx, host, port, queuelength_interval, verbose):
@@ -68,7 +69,10 @@ def main(ctx, host, port, queuelength_interval, verbose):
         except Exception as e:
             log.error(
                 'Failed to capture events: "%s", trying again in %s seconds.',
-                e, try_interval, exc_info=True)
+                e,
+                try_interval,
+                exc_info=True,
+            )
             time.sleep(try_interval)
 
 
@@ -78,11 +82,11 @@ def wrapper(self, event):
         self.state.event(event)
         task = self.state.tasks.get(event['uuid'])
         return fn(self, event, task)
+
     return wrapper
 
 
 class CeleryEventReceiver:
-
     def __init__(self, app):
         self.app = app
 
@@ -91,8 +95,7 @@ def on_task_started(self, event, task):
         log.debug('Started %s', task)
         STATS['tasks'].labels(task.routing_key, 'started').inc()
         if task.sent:
-            STATS['queuetime'].labels(task.routing_key).observe(
-                time.time() - task.sent)
+            STATS['queuetime'].labels(task.routing_key).observe(time.time() - task.sent)
 
     @task_handler
     def on_task_succeeded(self, event, task):
@@ -125,18 +128,20 @@ def __call__(self, *args, **kw):
         kw.setdefault('wakeup', False)
 
         with self.app.connection() as connection:
-            recv = self.app.events.Receiver(connection, handlers={
-                'task-started': self.on_task_started,
-                'task-succeeded': self.on_task_succeeded,
-                'task-failed': self.on_task_failed,
-                'task-retried': self.on_task_retried,
-                '*': self.state.event,
-            })
+            recv = self.app.events.Receiver(
+                connection,
+                handlers={
+                    'task-started': self.on_task_started,
+                    'task-succeeded': self.on_task_succeeded,
+                    'task-failed': self.on_task_failed,
+                    'task-retried': self.on_task_retried,
+                    '*': self.state.event,
+                },
+            )
             recv.capture(*args, **kw)
 
 
 class QueueLengthMonitor(threading.Thread):
-
     def __init__(self, app, interval):
         super().__init__()
         self.app = app
@@ -149,8 +154,7 @@ def run(self):
                 lengths = collections.Counter()
 
                 with self.app.connection() as connection:
-                    pipe = connection.channel().client.pipeline(
-                        transaction=False)
+                    pipe = connection.channel().client.pipeline(transaction=False)
                     for queue in self.app.conf['task_queues']:
                         # Not claimed by any worker yet
                         pipe.llen(queue.name)
@@ -173,9 +177,7 @@ def run(self):
 
                 time.sleep(self.interval)
             except Exception:
-                log.error(
-                    'Uncaught exception, preventing thread from crashing.',
-                    exc_info=True)
+                log.error('Uncaught exception, preventing thread from crashing.', exc_info=True)
 
     def stop(self):
         self.running = False

src/celery_redis_prometheus/tests/test_exporter.py

@@ -1,33 +1,32 @@
-from . import conftest
-import celery_redis_prometheus.exporter
-import pytest
 import threading
 
+import pytest
+
+import celery_redis_prometheus.exporter
+
+from . import conftest
+
 
 def test_collects_task_events(celery_worker):
-    receiver = celery_redis_prometheus.exporter.CeleryEventReceiver(
-        conftest.CELERY)
+    receiver = celery_redis_prometheus.exporter.CeleryEventReceiver(conftest.CELERY)
     # 3 = recived + started + succeeded
     thread = threading.Thread(target=lambda: receiver(limit=3))
     thread.start()
     conftest.celery_ping.delay().get()
     thread.join()
     data = celery_redis_prometheus.exporter.STATS['tasks'].collect()
-    item = [x for x in data[0].samples
-            if x.labels.get('state', '') == 'succeeded']
+    item = [x for x in data[0].samples if x.labels.get('state', '') == 'succeeded']
     assert item[0].value == 1
 
 
 def test_sets_separate_state_for_retry_failed(celery_worker):
-    receiver = celery_redis_prometheus.exporter.CeleryEventReceiver(
-        conftest.CELERY)
+    receiver = celery_redis_prometheus.exporter.CeleryEventReceiver(conftest.CELERY)
     # 6 = received + started + retry + received + started + failed
     thread = threading.Thread(target=lambda: receiver(limit=6))
     thread.start()
     with pytest.raises(Exception):
         conftest.provoke_retry.delay().get()
     thread.join()
     data = celery_redis_prometheus.exporter.STATS['tasks'].collect()
-    item = [x for x in data[0].samples
-            if x.labels.get('state', '') == 'retries-exceeded']
+    item = [x for x in data[0].samples if x.labels.get('state', '') == 'retries-exceeded']
     assert item[0].value == 1