Formatting and linting

Author: angelosantos4

nodestream/metrics.py

@@ -27,16 +27,16 @@ def decrement_on(self, handler: "MetricHandler", value: Number = 1):
     def register(self, handler: "MetricHandler"):
         handler.increment(self, 0)
         return self
-    
+
     def __str__(self):
         return f"{self.name}: {self.value}"
-    
+
     def __hash__(self):
         return hash(self.name)
-    
+
     def __eq__(self, other):
         return self.name == other.name
-    
+
     def __ne__(self, other):
         return self.name != other.name
 
@@ -88,17 +88,27 @@ def tick(self):
 
 # Core metrics
 RECORDS = Metric("records", "Number of records processed", accumulate=True)
-NON_FATAL_ERRORS = Metric("non_fatal_errors", "Number of non-fatal errors", accumulate=True)
+NON_FATAL_ERRORS = Metric(
+    "non_fatal_errors", "Number of non-fatal errors", accumulate=True
+)
 FATAL_ERRORS = Metric("fatal_errors", "Number of fatal errors", accumulate=True)
-NODES_UPSERTED = Metric("nodes_upserted", "Number of nodes upserted to the graph", accumulate=True)
+NODES_UPSERTED = Metric(
+    "nodes_upserted", "Number of nodes upserted to the graph", accumulate=True
+)
 RELATIONSHIPS_UPSERTED = Metric(
-    "relationships_upserted", "Number of relationships upserted to the graph", accumulate=True
+    "relationships_upserted",
+    "Number of relationships upserted to the graph",
+    accumulate=True,
 )
 TIME_TO_LIVE_OPERATIONS = Metric(
-    "time_to_live_operations", "Number of time-to-live operations executed", accumulate=True
+    "time_to_live_operations",
+    "Number of time-to-live operations executed",
+    accumulate=True,
 )
 INGEST_HOOKS_EXECUTED = Metric(
-    "ingest_hooks_executed", "Number of ingest hooks executed to the graph", accumulate=True
+    "ingest_hooks_executed",
+    "Number of ingest hooks executed to the graph",
+    accumulate=True,
 )
 STEPS_RUNNING = Metric(
     "steps_running", "Number of steps currently running in the pipeline"
@@ -204,7 +214,7 @@ def increment(self, metric: Metric, value: Number):
         self.metrics[metric] = self.metrics.get(metric, 0) + value
 
     def decrement(self, metric: Metric, value: Number):
-        self.metrics[metric] = self.metrics.get(metric, 0) - value 
+        self.metrics[metric] = self.metrics.get(metric, 0) - value
 
     def discharge(self) -> dict[Metric, Number]:
         metrics = {}

nodestream/pipeline/progress_reporter.py

@@ -1,12 +1,12 @@
 import os
+import time
 from dataclasses import dataclass, field
 from logging import Logger, getLogger
 from typing import Any, Callable, Optional
 
 from psutil import Process
 
 from ..metrics import Metrics
-import time
 
 
 def no_op(*_, **__):

tests/unit/cli/operations/test_run_pipeline.py

@@ -1,6 +1,5 @@
 import pytest
 from hamcrest import assert_that, equal_to
-from unittest.mock import Mock
 
 from nodestream.cli.operations.run_pipeline import (
     WARNING_NO_TARGETS_PROVIDED,
@@ -40,7 +39,7 @@ def test_make_run_request(run_pipeline_operation, mocker):
         "step-outbox-size": "10001",
         "target": targets,
         "time-interval-seconds": None,  # No time interval provided
-        "reporting-frequency": "10000"
+        "reporting-frequency": "10000",
     }
     command.option.side_effect = lambda opt: option_responses.get(opt)
     command.has_json_logging_set = False  # Required for create_progress_reporter
@@ -55,41 +54,49 @@ def test_make_run_request(run_pipeline_operation, mocker):
     assert_that(result.progress_reporter.reporting_frequency, equal_to(10000))
 
 
-def test_create_progress_reporter_with_time_interval_seconds(run_pipeline_operation, mocker):
+def test_create_progress_reporter_with_time_interval_seconds(
+    run_pipeline_operation, mocker
+):
     """Test that time_interval_seconds gets properly converted to float and passed to PipelineProgressReporter"""
     command = mocker.Mock()
     command.option.side_effect = lambda opt: {
         "time-interval-seconds": "30.5",
-        "reporting-frequency": "1000"
+        "reporting-frequency": "1000",
     }.get(opt)
     command.has_json_logging_set = False
-    
+
     # Mock PipelineProgressReporter to capture arguments
-    mock_progress_reporter = mocker.patch("nodestream.cli.operations.run_pipeline.PipelineProgressReporter")
-    
-    result = run_pipeline_operation.create_progress_reporter(command, "test_pipeline")
-    
+    mock_progress_reporter = mocker.patch(
+        "nodestream.cli.operations.run_pipeline.PipelineProgressReporter"
+    )
+
+    _ = run_pipeline_operation.create_progress_reporter(command, "test_pipeline")
+
     # Verify PipelineProgressReporter was called with correct time_interval_seconds
     mock_progress_reporter.assert_called_once()
     call_args = mock_progress_reporter.call_args
     assert_that(call_args.kwargs["time_interval_seconds"], equal_to(30.5))
     assert_that(call_args.kwargs["reporting_frequency"], equal_to(1000))
 
 
-def test_create_progress_reporter_without_time_interval_seconds(run_pipeline_operation, mocker):
+def test_create_progress_reporter_without_time_interval_seconds(
+    run_pipeline_operation, mocker
+):
     """Test that time_interval_seconds is None when not provided"""
     command = mocker.Mock()
     command.option.side_effect = lambda opt: {
         "time-interval-seconds": None,
-        "reporting-frequency": "2000"
+        "reporting-frequency": "2000",
     }.get(opt)
     command.has_json_logging_set = False
-    
+
     # Mock PipelineProgressReporter to capture arguments
-    mock_progress_reporter = mocker.patch("nodestream.cli.operations.run_pipeline.PipelineProgressReporter")
-    
-    result = run_pipeline_operation.create_progress_reporter(command, "test_pipeline")
-    
+    mock_progress_reporter = mocker.patch(
+        "nodestream.cli.operations.run_pipeline.PipelineProgressReporter"
+    )
+
+    _ = run_pipeline_operation.create_progress_reporter(command, "test_pipeline")
+
     # Verify PipelineProgressReporter was called with None for time_interval_seconds
     mock_progress_reporter.assert_called_once()
     call_args = mock_progress_reporter.call_args
@@ -102,62 +109,68 @@ def test_create_progress_reporter_with_json_indicator(run_pipeline_operation, mo
     command = mocker.Mock()
     command.option.side_effect = lambda opt: {
         "time-interval-seconds": "15.0",
-        "reporting-frequency": "500"
+        "reporting-frequency": "500",
     }.get(opt)
     command.has_json_logging_set = True
-    
+
     # Mock PipelineProgressReporter to capture arguments
-    mock_progress_reporter = mocker.patch("nodestream.cli.operations.run_pipeline.PipelineProgressReporter")
-    
-    result = run_pipeline_operation.create_progress_reporter(command, "test_pipeline")
-    
+    mock_progress_reporter = mocker.patch(
+        "nodestream.cli.operations.run_pipeline.PipelineProgressReporter"
+    )
+
+    _ = run_pipeline_operation.create_progress_reporter(command, "test_pipeline")
+
     # Verify PipelineProgressReporter was called with correct arguments
     mock_progress_reporter.assert_called_once()
     call_args = mock_progress_reporter.call_args
     assert_that(call_args.kwargs["time_interval_seconds"], equal_to(15.0))
     assert_that(call_args.kwargs["reporting_frequency"], equal_to(500))
 
 
-def test_make_run_request_with_time_interval_seconds_integration(run_pipeline_operation, mocker):
+def test_make_run_request_with_time_interval_seconds_integration(
+    run_pipeline_operation, mocker
+):
     """Integration test to ensure make_run_request properly handles time_interval_seconds through create_progress_reporter"""
     annotations = ["annotation1"]
     targets = ["t1"]
     pipeline_name = "my_pipeline"
     command = mocker.Mock()
-    
+
     # Setup command.option to handle all the different option calls made by make_run_request
     option_responses = {
         "storage-backend": "my-storage",
         "annotations": annotations,
         "step-outbox-size": "10001",
         "target": targets,
         "time-interval-seconds": "45.0",
-        "reporting-frequency": "5000"
+        "reporting-frequency": "5000",
     }
     command.option.side_effect = lambda opt: option_responses.get(opt)
     command.has_json_logging_set = False
     command.is_very_verbose = False
     command.argument.return_value = [pipeline_name]
-    
+
     pipeline = mocker.Mock()
     pipeline.name = pipeline_name
     pipeline.configuration = PipelineConfiguration()
-    
+
     # Mock the project's get_object_storage_by_name method
     run_pipeline_operation.project.get_object_storage_by_name.return_value = None
     run_pipeline_operation.project.get_target_by_name.return_value = None
-    
+
     # Mock PipelineProgressReporter to capture its arguments
-    mock_progress_reporter = mocker.patch("nodestream.cli.operations.run_pipeline.PipelineProgressReporter")
-    
+    mock_progress_reporter = mocker.patch(
+        "nodestream.cli.operations.run_pipeline.PipelineProgressReporter"
+    )
+
     result = run_pipeline_operation.make_run_request(command, pipeline)
-    
+
     # Verify the progress reporter was created with correct time_interval_seconds
     mock_progress_reporter.assert_called_once()
     call_args = mock_progress_reporter.call_args
     assert_that(call_args.kwargs["time_interval_seconds"], equal_to(45.0))
     assert_that(call_args.kwargs["reporting_frequency"], equal_to(5000))
-    
+
     # Verify other parts of the request are still correct
     assert_that(result.pipeline_name, equal_to(pipeline_name))
     assert_that(result.initialization_arguments.annotations, equal_to(annotations))

tests/unit/pipeline/test_pipeline_progress_reporter.py

@@ -1,9 +1,10 @@
+from unittest.mock import Mock
+
 import pytest
 from hamcrest import assert_that, equal_to
-from unittest.mock import Mock
 
-from nodestream.pipeline import IterableExtractor, Pipeline, PipelineProgressReporter
 from nodestream.metrics import Metrics
+from nodestream.pipeline import IterableExtractor, Pipeline, PipelineProgressReporter
 
 
 @pytest.mark.asyncio
@@ -25,32 +26,29 @@ def test_pipeline_progress_reporter_with_time_interval_seconds(mocker):
     """Test that time_interval_seconds works correctly"""
     mock_callback = Mock()
     reporter = PipelineProgressReporter(
-        time_interval_seconds=0.1,
-        callback=mock_callback
+        time_interval_seconds=0.1, callback=mock_callback
     )
-    
-    mock_time = mocker.patch('nodestream.pipeline.progress_reporter.time.time')
+
+    mock_time = mocker.patch("nodestream.pipeline.progress_reporter.time.time")
     mock_time.side_effect = [0.15, 0.2]  # 150ms, 200ms
-    
+
     metrics = Metrics()
     reporter.report(1, metrics)  # Should report (150ms >= 100ms from 0)
     reporter.report(2, metrics)  # Should not report (200ms - 150ms = 50ms < 100ms)
-    
+
     assert_that(mock_callback.call_count, equal_to(1))
 
 
 def test_pipeline_progress_reporter_without_time_interval_uses_frequency():
     """Test that None time_interval_seconds falls back to frequency-based reporting"""
     mock_callback = Mock()
     reporter = PipelineProgressReporter(
-        time_interval_seconds=None,
-        reporting_frequency=3,
-        callback=mock_callback
+        time_interval_seconds=None, reporting_frequency=3, callback=mock_callback
     )
-    
+
     metrics = Metrics()
     for i in range(1, 7):  # 1,2,3,4,5,6
         reporter.report(i, metrics)
-    
+
     # Should report on multiples of 3: indices 3, 6
     assert_that(mock_callback.call_count, equal_to(2))

tests/unit/test_metrics.py

@@ -147,11 +147,11 @@ def test_metric_equality_and_hash():
     metric1 = Metric("test_metric", "Test description")
     metric2 = Metric("test_metric", "Different description")
     metric3 = Metric("different_metric", "Test description")
-    
+
     # Test __eq__ and __ne__
     assert metric1 == metric2  # Same name
     assert metric1 != metric3  # Different name
-    
+
     # Test __hash__
     assert hash(metric1) == hash(metric2)  # Same name should have same hash
 
@@ -160,19 +160,19 @@ def test_console_metric_handler_discharge_with_accumulate(mocker):
     """Test that ConsoleMetricHandler discharge resets accumulating metrics"""
     mock_command = mocker.Mock()
     handler = ConsoleMetricHandler(mock_command)
-    
+
     accumulating_metric = Metric("test_accumulate", accumulate=True)
     non_accumulating_metric = Metric("test_no_accumulate", accumulate=False)
-    
+
     handler.increment(accumulating_metric, 5)
     handler.increment(non_accumulating_metric, 3)
-    
+
     result = handler.discharge()
-    
+
     # Should return metric names as keys
     assert result["test_accumulate"] == 5
     assert result["test_no_accumulate"] == 3
-    
+
     # Accumulating metric should be reset to 0, non-accumulating should remain
     assert handler.metrics[accumulating_metric] == 0
     assert handler.metrics[non_accumulating_metric] == 3
@@ -181,19 +181,19 @@ def test_console_metric_handler_discharge_with_accumulate(mocker):
 def test_json_log_metric_handler_discharge_with_accumulate(mocker):
     """Test that JsonLogMetricHandler discharge resets accumulating metrics"""
     handler = JsonLogMetricHandler()
-    
+
     accumulating_metric = Metric("test_accumulate", accumulate=True)
     non_accumulating_metric = Metric("test_no_accumulate", accumulate=False)
-    
+
     handler.increment(accumulating_metric, 10)
     handler.increment(non_accumulating_metric, 7)
-    
+
     result = handler.discharge()
-    
+
     # Should return metric names as keys
     assert result["test_accumulate"] == 10
     assert result["test_no_accumulate"] == 7
-    
+
     # Accumulating metric should be reset to 0, non-accumulating should remain
     assert handler.metrics[accumulating_metric] == 0
     assert handler.metrics[non_accumulating_metric] == 7