ruffed

mapping_provider/__init__.py

@@ -12,7 +12,7 @@ from fastapi import FastAPI
 
 from mapping_provider import config, environment
 from mapping_provider.api import common, map
-from mapping_provider.backends import cache, correlator, inventory, brian
+from mapping_provider.backends import brian, cache, correlator, inventory
 
 logger = logging.getLogger(__name__)
 

mapping_provider/backends/brian.py

 import contextlib
-from influxdb import InfluxDBClient
-from threading import Event
+import logging
 import time
-from typing import Generator, Any
+from collections.abc import Generator
+from threading import Event
+from typing import Any
 
-import logging
+from influxdb import InfluxDBClient
 
 from mapping_provider import config
 from mapping_provider.backends import cache
@@ -59,7 +60,8 @@ CACHED_SCID_RATES_SCHEMA = {
 }
 
 @contextlib.contextmanager
-def influx_client(influx_params: config.InfluxConnectionParams) -> Generator[InfluxDBClient]:
+def influx_client(
+    influx_params: config.InfluxConnectionParams) -> Generator[InfluxDBClient]:
     """
     context manager for creating an influx db connection
 
@@ -89,7 +91,7 @@ def influx_client(influx_params: config.InfluxConnectionParams) -> Generator[Inf
 
 def _load_scid_rates_rows(
         influx_params: config.InfluxConnectionParams,
-        window: str = '24h') -> Generator[dict[str, str|float|None], None, None]:
+        window: str = '24h') -> Generator[dict[str, str | float | None]]:
     """
     Return the count of all fields, grouped by
     hostname & interface_name.
@@ -124,7 +126,7 @@ def _load_scid_rates_rows(
 
             for row in values:
                 rate = {'scid': tags['scid']}
-                for field_name, field_value in zip(columns, row):
+                for field_name, field_value in zip(columns, row, strict=False):
                     rate[field_name] = field_value
                 yield rate
 

mapping_provider/backends/services.py

@@ -3,7 +3,7 @@ from collections.abc import Generator
 
 from pydantic import BaseModel
 
-from . import cache, correlator, inventory, brian
+from . import brian, cache, correlator, inventory
 
 logger = logging.getLogger(__name__)
 

test/test_utilization.py

 import tempfile
-from unittest.mock import patch, MagicMock
+from unittest.mock import MagicMock, patch
 
 import jsonschema
 
-from .common import load_test_data
 from mapping_provider.backends import brian, cache
 from mapping_provider.config import InfluxConnectionParams
 
+from .common import load_test_data
+
+
 def test_utilization():
 
     with tempfile.TemporaryDirectory() as tmpdir: