Add linting with ruff

.github/workflows/build.yml

@@ -24,8 +24,12 @@ jobs:
         sudo rm -f /etc/boto.cfg
         sudo apt-get -qq update
         sudo apt-get install -y libicu-dev
-        pip install pytest pytest-env pytest-cov pytest-mock wheel
+        make dev
         pip install -e ".[dev]"
+    - name: Run the code format check
+      run: make format-check
+    - name: Run the linter
+      run: make lint
     - name: Run the tests
       run: |
         make test
@@ -36,4 +40,4 @@ jobs:
       if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags')
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
-        password: ${{ secrets.pypi_password }}
+        password: ${{ secrets.pypi_password }}

Makefile

@@ -8,9 +8,23 @@ install:
 	pip install -q -e .
 	pip install -q twine coverage nose moto boto3
 
+dev:
+	python3 -m pip install --upgrade pip
+	python3 -m pip install -q -r requirements.txt
+	python3 -m pip install -q -r requirements-dev.txt
+
 test:
 	docker-compose run --rm shell pytest --cov=servicelayer
 
+lint:
+	ruff check .
+
+format:
+	black .
+
+format-check:
+	black --check .
+
 build:
 	python setup.py sdist bdist_wheel
 

requirements-dev.txt

@@ -0,0 +1,6 @@
+black==23.3.0
+ruff==0.0.270
+pytest==7.3.1
+pytest-env==0.8.1
+pytest-cov==4.1.0
+pytest-mock==3.10.0

ruff.toml

@@ -0,0 +1,44 @@
+# Enable pycodestyle (`E`) and Pyflakes (`F`) codes by default.
+select = ["E", "F"]
+ignore = []
+
+# Allow autofix for all enabled rules (when `--fix`) is provided.
+fixable = ["A", "B", "C", "D", "E", "F", "G", "I", "N", "Q", "S", "T", "W", "ANN", "ARG", "BLE", "COM", "DJ", "DTZ", "EM", "ERA", "EXE", "FBT", "ICN", "INP", "ISC", "NPY", "PD", "PGH", "PIE", "PL", "PT", "PTH", "PYI", "RET", "RSE", "RUF", "SIM", "SLF", "TCH", "TID", "TRY", "UP", "YTT"]
+unfixable = []
+
+# Exclude a variety of commonly ignored directories.
+exclude = [
+    ".bzr",
+    ".direnv",
+    ".eggs",
+    ".git",
+    ".git-rewrite",
+    ".hg",
+    ".mypy_cache",
+    ".nox",
+    ".pants.d",
+    ".pytype",
+    ".ruff_cache",
+    ".svn",
+    ".tox",
+    ".venv",
+    "__pypackages__",
+    "_build",
+    "buck-out",
+    "build",
+    "dist",
+    "node_modules",
+    "venv",
+]
+
+# Same as Black.
+line-length = 88
+
+# Allow unused variables when underscore-prefixed.
+dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
+
+target-version = "py38"
+
+[mccabe]
+# Unlike Flake8, default to a complexity level of 10.
+max-complexity = 10

servicelayer/metrics.py

@@ -0,0 +1,45 @@
+from opencensus.ext.stackdriver import stats_exporter
+from opencensus.stats import aggregation
+from opencensus.stats import measure
+from opencensus.stats import stats
+from opencensus.stats import view
+
+
+# A measure that represents task latency in ms.
+LATENCY_MS = measure.MeasureFloat(
+    "task_latency", "The task latency in milliseconds", "ms"
+)
+
+# A view of the task latency measure that aggregates measurements according to
+# a histogram with predefined bucket boundaries. This aggregate is periodically
+# exported to Stackdriver Monitoring.
+LATENCY_VIEW = view.View(
+    "task_latency_distribution",
+    "The distribution of the task latencies",
+    [],
+    LATENCY_MS,
+    # Latency in buckets: [>=0ms, >=100ms, >=200ms, >=400ms, >=1s, >=2s, >=4s]
+    aggregation.DistributionAggregation([100.0, 200.0, 400.0, 1000.0, 2000.0, 4000.0]),
+)
+
+MEASURES = [
+    LATENCY_MS,
+]
+
+VIEWS = [
+    LATENCY_VIEW,
+]
+
+
+def init():
+    for view in VIEWS:
+        stats.stats.view_manager.register_view(view)
+    exporter = stats_exporter.new_stats_exporter()
+    # print('Exporting stats to project "{}"'.format(exporter.options.project_id))
+    stats.stats.view_manager.register_exporter(exporter)
+
+
+def record_value(measure, value):
+    mmap = stats.stats.stats_recorder.new_measurement_map()
+    mmap.measure_float_put(measure, value)
+    mmap.record()

servicelayer/tags.py

@@ -83,7 +83,10 @@ def _upsert_values(self, conn, row):
         istmt = upsert(self.table).values(row)
         stmt = istmt.on_conflict_do_update(
             index_elements=["key"],
-            set_=dict(value=istmt.excluded.value, timestamp=istmt.excluded.timestamp,),
+            set_=dict(
+                value=istmt.excluded.value,
+                timestamp=istmt.excluded.timestamp,
+            ),
         )
         conn.execute(stmt)
 

servicelayer/taskqueue.py

@@ -131,8 +131,9 @@ def cleanup_dataset_status(cls, conn):
     def should_execute(self, task_id):
         """Should a task be executed?
 
-        When a the processing of a task is cancelled, there is no way to tell RabbitMQ to drop it.
-        So we store that state in Redis and make the worker check with Redis before executing a task.
+        When a the processing of a task is cancelled, there is no way
+        to tell RabbitMQ to drop it. So we store that state in Redis
+        and make the worker check with Redis before executing a task.
         """
         attempt = 1
         while True:
@@ -369,7 +370,8 @@ def ack_message(self, task, channel):
         skip_ack = task.context.get("skip_ack")
         if skip_ack:
             log.info(
-                f"Skipping acknowledging message {task.delivery_tag} for task_id {task.task_id}"
+                f"Skipping acknowledging message {task.delivery_tag}"
+                "for task_id {task.task_id}"
             )
         else:
             log.info(
@@ -390,7 +392,9 @@ def run(self):
         signal.signal(signal.SIGTERM, self.on_signal)
 
         # worker threads
-        process = lambda: self.process(blocking=True)
+        def process():
+            return self.process(blocking=True)
+
         threads = []
         for _ in range(self.num_threads):
             thread = threading.Thread(target=process)

servicelayer/worker.py

@@ -75,7 +75,8 @@ def process(self, blocking=True, interval=INTERVAL):
             task = Stage.get_task(self.conn, stages, timeout=interval)
             if task is None:
                 if not blocking:
-                    # If we get a null task, retry to fetch a task a bunch of times before quitting
+                    # If we get a null task, retry to fetch a task
+                    # a bunch of times before quitting
                     if retries >= TASK_FETCH_RETRY:
                         log.info("Worker thread is exiting")
                         return self.exit_code
@@ -96,7 +97,10 @@ def run(self, blocking=True, interval=INTERVAL):
         signal.signal(signal.SIGINT, self._handle_signal)
         signal.signal(signal.SIGTERM, self._handle_signal)
         self.init_internal()
-        process = lambda: self.process(blocking=blocking, interval=interval)
+
+        def process():
+            return self.process(blocking=blocking, interval=interval)
+
         if not self.num_threads:
             return process()
         log.info("Worker has %d threads.", self.num_threads)