Formatting and linting

Author: fordN

src/amp/client.py

@@ -362,9 +362,7 @@ def query_and_load_streaming(
                         f'timestamp {checkpoint.timestamp}'
                     )
                     if checkpoint.is_reorg:
-                        resume_points = ', '.join(
-                            f'{r.network}:{r.start}' for r in checkpoint.ranges
-                        )
+                        resume_points = ', '.join(f'{r.network}:{r.start}' for r in checkpoint.ranges)
                         self.logger.info(f'Reorg resume points: {resume_points}')
             except Exception as e:
                 self.logger.warning(f'Failed to load checkpoint, starting from beginning: {e}')

src/amp/loaders/base.py

@@ -197,8 +197,8 @@ def load_batch(self, batch: pa.RecordBatch, table_name: str, **kwargs) -> LoadRe
                 )
                 self.logger.error(error_msg)
                 self.logger.error(
-                    f'Client will stop. On restart, streaming will resume from last checkpoint. '
-                    f'Fix the data/configuration issue before restarting.'
+                    'Client will stop. On restart, streaming will resume from last checkpoint. '
+                    'Fix the data/configuration issue before restarting.'
                 )
                 # Raise exception to stop the stream
                 raise RuntimeError(error_msg)
@@ -220,8 +220,8 @@ def load_batch(self, batch: pa.RecordBatch, table_name: str, **kwargs) -> LoadRe
                 )
                 self.logger.error(error_msg)
                 self.logger.error(
-                    f'Client will stop. On restart, streaming will resume from last checkpoint. '
-                    f'Fix the underlying issue before restarting.'
+                    'Client will stop. On restart, streaming will resume from last checkpoint. '
+                    'Fix the underlying issue before restarting.'
                 )
                 # Raise exception to stop the stream
                 raise RuntimeError(error_msg)
@@ -547,8 +547,7 @@ def _process_reorg_event(
                 # Log reorg details
                 for range_obj in response.invalidation_ranges:
                     self.logger.warning(
-                        f'Reorg detected on {range_obj.network}: '
-                        f'blocks {range_obj.start}-{range_obj.end} invalidated'
+                        f'Reorg detected on {range_obj.network}: blocks {range_obj.start}-{range_obj.end} invalidated'
                     )
 
                 # Save reorg checkpoint (keeps old checkpoints for history)
@@ -674,9 +673,7 @@ def _process_batch_non_transactional(
 
             if is_duplicate:
                 # Skip this batch - already processed
-                self.logger.info(
-                    f'Skipping duplicate batch: {len(ranges)} ranges already processed for {table_name}'
-                )
+                self.logger.info(f'Skipping duplicate batch: {len(ranges)} ranges already processed for {table_name}')
                 return LoadResult(
                     rows_loaded=0,
                     duration=0.0,
@@ -693,9 +690,7 @@ def _process_batch_non_transactional(
         if result.success and ranges:
             # Mark batch as processed (for exactly-once semantics)
             try:
-                self.processed_ranges_store.mark_processed(
-                    connection_name, table_name, ranges, batch_hash
-                )
+                self.processed_ranges_store.mark_processed(connection_name, table_name, ranges, batch_hash)
             except Exception as e:
                 self.logger.error(f'Failed to mark ranges as processed: {e}')
                 # Continue anyway - checkpoint will provide resume capability
@@ -733,10 +728,7 @@ def _save_checkpoint_if_complete(
 
             try:
                 self.checkpoint_store.save(connection_name, table_name, checkpoint)
-                self.logger.info(
-                    f'Saved checkpoint at batch {batch_count} '
-                    f'({len(checkpoint.ranges)} ranges)'
-                )
+                self.logger.info(f'Saved checkpoint at batch {batch_count} ({len(checkpoint.ranges)} ranges)')
             except Exception as e:
                 # Log but don't fail the stream
                 self.logger.error(f'Failed to save checkpoint: {e}')
@@ -760,9 +752,7 @@ def _augment_streaming_result(
         result.metadata['batch_count'] = batch_count
         result.metadata['ranges_complete'] = ranges_complete
         if ranges:
-            result.metadata['block_ranges'] = [
-                {'network': r.network, 'start': r.start, 'end': r.end} for r in ranges
-            ]
+            result.metadata['block_ranges'] = [{'network': r.network, 'start': r.start, 'end': r.end} for r in ranges]
         return result
 
     def _compute_reorg_resume_point(self, invalidation_ranges: List[BlockRange]) -> List[BlockRange]:

src/amp/loaders/implementations/postgresql_loader.py

@@ -4,7 +4,7 @@
 import pyarrow as pa
 from psycopg2.pool import ThreadedConnectionPool
 
-from ...streaming.idempotency import DatabaseProcessedRangesStore, IdempotencyConfig
+from ...streaming.idempotency import DatabaseProcessedRangesStore
 from ...streaming.types import BlockRange
 from ..base import DataLoader, LoadMode
 from ._postgres_helpers import has_binary_columns, prepare_csv_data, prepare_insert_data

src/amp/streaming/checkpoint.py

@@ -9,7 +9,7 @@
 import json
 import logging
 from abc import ABC, abstractmethod
-from dataclasses import asdict, dataclass
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Dict, List, Optional
 
@@ -88,16 +88,12 @@ def __init__(self, config: CheckpointConfig):
         self.config = config
 
     @abstractmethod
-    def save(
-        self, connection_name: str, table_name: str, checkpoint: CheckpointState
-    ) -> None:
+    def save(self, connection_name: str, table_name: str, checkpoint: CheckpointState) -> None:
         """Save a checkpoint for a specific connection and table"""
         pass
 
     @abstractmethod
-    def load(
-        self, connection_name: str, table_name: str, worker_id: int = 0
-    ) -> Optional[CheckpointState]:
+    def load(self, connection_name: str, table_name: str, worker_id: int = 0) -> Optional[CheckpointState]:
         """Load the latest checkpoint for a connection and table"""
         pass
 
@@ -159,9 +155,7 @@ def _ensure_table_exists(self):
             logger.debug(f'Checkpoint table creation skipped: {e}')
             self.conn.rollback()
 
-    def save(
-        self, connection_name: str, table_name: str, checkpoint: CheckpointState
-    ) -> None:
+    def save(self, connection_name: str, table_name: str, checkpoint: CheckpointState) -> None:
         """Save checkpoint to database"""
         if not self.config.enabled:
             return
@@ -203,9 +197,7 @@ def save(
             self.conn.rollback()
             raise
 
-    def load(
-        self, connection_name: str, table_name: str, worker_id: int = 0
-    ) -> Optional[CheckpointState]:
+    def load(self, connection_name: str, table_name: str, worker_id: int = 0) -> Optional[CheckpointState]:
         """Load checkpoint from database"""
         if not self.config.enabled:
             return None
@@ -310,14 +302,10 @@ def delete(self, connection_name: str, table_name: str, worker_id: int = 0) -> N
 class NullCheckpointStore(CheckpointStore):
     """No-op checkpoint store for when checkpointing is disabled"""
 
-    def save(
-        self, connection_name: str, table_name: str, checkpoint: CheckpointState
-    ) -> None:
+    def save(self, connection_name: str, table_name: str, checkpoint: CheckpointState) -> None:
         pass
 
-    def load(
-        self, connection_name: str, table_name: str, worker_id: int = 0
-    ) -> Optional[CheckpointState]:
+    def load(self, connection_name: str, table_name: str, worker_id: int = 0) -> Optional[CheckpointState]:
         return None
 
     def delete_for_network(self, connection_name: str, table_name: str, network: str) -> None:

src/amp/streaming/idempotency.py

@@ -6,7 +6,6 @@
 """
 
 import hashlib
-import json
 import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
@@ -96,9 +95,7 @@ def __init__(self, config: IdempotencyConfig):
         self.config = config
 
     @abstractmethod
-    def is_processed(
-        self, connection_name: str, table_name: str, ranges: List[BlockRange]
-    ) -> bool:
+    def is_processed(self, connection_name: str, table_name: str, ranges: List[BlockRange]) -> bool:
         """
         Check if a set of block ranges has already been processed.
 
@@ -200,9 +197,7 @@ def _ensure_table_exists(self):
             logger.debug(f'Processed ranges table creation skipped: {e}')
             self.conn.rollback()
 
-    def is_processed(
-        self, connection_name: str, table_name: str, ranges: List[BlockRange]
-    ) -> bool:
+    def is_processed(self, connection_name: str, table_name: str, ranges: List[BlockRange]) -> bool:
         """Check if all ranges have been processed"""
         if not self.config.enabled or not ranges:
             return False
@@ -243,10 +238,7 @@ def is_processed(
                 return False
 
         # All ranges have been processed
-        logger.info(
-            f'Duplicate detected: {len(ranges)} ranges already processed for '
-            f'{connection_name}.{table_name}'
-        )
+        logger.info(f'Duplicate detected: {len(ranges)} ranges already processed for {connection_name}.{table_name}')
         return True
 
     def mark_processed(
@@ -290,9 +282,7 @@ def mark_processed(
                 )
 
             self.conn.commit()
-            logger.debug(
-                f'Marked {len(ranges)} ranges as processed for {connection_name}.{table_name}'
-            )
+            logger.debug(f'Marked {len(ranges)} ranges as processed for {connection_name}.{table_name}')
 
         except Exception as e:
             logger.error(f'Failed to mark ranges as processed: {e}')
@@ -336,9 +326,7 @@ def cleanup_old_ranges(self, connection_name: str, table_name: str, days: int) -
 class NullProcessedRangesStore(ProcessedRangesStore):
     """No-op processed ranges store when idempotency is disabled"""
 
-    def is_processed(
-        self, connection_name: str, table_name: str, ranges: List[BlockRange]
-    ) -> bool:
+    def is_processed(self, connection_name: str, table_name: str, ranges: List[BlockRange]) -> bool:
         return False
 
     def mark_processed(

src/amp/streaming/types.py

@@ -121,9 +121,7 @@ def from_flight_data(cls, metadata_bytes: bytes) -> 'BatchMetadata':
             # Store remaining fields in extra
             extra = {k: v for k, v in metadata_dict.items() if k not in ('ranges', 'ranges_complete')}
 
-            return cls(
-                ranges=ranges, ranges_complete=ranges_complete, extra=extra if extra else None
-            )
+            return cls(ranges=ranges, ranges_complete=ranges_complete, extra=extra if extra else None)
         except (json.JSONDecodeError, KeyError) as e:
             # Fallback to empty metadata if parsing fails
             return cls(ranges=[], ranges_complete=False, extra={'parse_error': str(e)})

tests/integration/test_checkpoint_resume.py

@@ -7,13 +7,10 @@
 
 import time
 from datetime import datetime
-from unittest.mock import MagicMock, Mock, patch
 
 import psycopg2
-import pyarrow as pa
 import pytest
 
-from src.amp.loaders.base import DataLoader
 from src.amp.streaming.checkpoint import (
     CheckpointConfig,
     CheckpointState,
@@ -368,7 +365,6 @@ class TestCheckpointDisabled:
 
     def test_checkpoint_disabled_by_default(self):
         """Test that checkpoints are disabled by default"""
-        from src.amp.streaming.checkpoint import NullCheckpointStore
 
         config = CheckpointConfig()
         assert config.enabled is False
@@ -633,7 +629,6 @@ def test_exactly_once_with_checkpoint_integration(self, checkpoint_db_connection
     def test_transactional_exactly_once_postgresql(self, postgresql_test_config, small_test_data):
         """Test transactional exactly-once semantics in PostgreSQL loader"""
         from src.amp.loaders.implementations.postgresql_loader import PostgreSQLLoader
-        from src.amp.streaming.idempotency import IdempotencyConfig
 
         # Configure loader with idempotency enabled
         config_dict = {
@@ -696,10 +691,13 @@ def test_transactional_exactly_once_postgresql(self, postgresql_test_config, sma
             conn = loader.pool.getconn()
             try:
                 cursor = conn.cursor()
-                cursor.execute("""
+                cursor.execute(
+                    """
                     SELECT COUNT(*) FROM test_amp_processed_ranges
                     WHERE connection_name = %s AND table_name = %s
-                """, (connection_name, test_table))
+                """,
+                    (connection_name, test_table),
+                )
                 processed_count = cursor.fetchone()[0]
                 assert processed_count == 1, 'Exactly one processed range entry should exist'
             finally:

tests/integration/test_resilient_streaming.py

@@ -8,17 +8,11 @@
 import time
 from dataclasses import dataclass
 from typing import Any, Dict
-from unittest.mock import MagicMock, patch
 
 import pyarrow as pa
 import pytest
 
 from amp.loaders.base import DataLoader
-from amp.loaders.types import LoadResult
-from amp.streaming.resilience import (
-    BackPressureConfig,
-    RetryConfig,
-)
 
 
 @dataclass

tests/unit/test_checkpoint.py

@@ -2,7 +2,6 @@
 Unit tests for checkpoint system.
 """
 
-import json
 from datetime import datetime
 
 import pytest

tests/unit/test_idempotency.py

@@ -205,9 +205,10 @@ class TestBatchHashing:
 
     def test_compute_batch_hash(self):
         """Test computing hash of batch data"""
-        from src.amp.streaming.idempotency import compute_batch_hash
         import pyarrow as pa
 
+        from src.amp.streaming.idempotency import compute_batch_hash
+
         # Create test batch
         schema = pa.schema([('id', pa.int64()), ('value', pa.string())])
         batch = pa.record_batch([[1, 2, 3], ['a', 'b', 'c']], schema=schema)
@@ -220,9 +221,10 @@ def test_compute_batch_hash(self):
 
     def test_same_data_same_hash(self):
         """Test that same data produces same hash"""
-        from src.amp.streaming.idempotency import compute_batch_hash
         import pyarrow as pa
 
+        from src.amp.streaming.idempotency import compute_batch_hash
+
         schema = pa.schema([('id', pa.int64()), ('value', pa.string())])
 
         batch1 = pa.record_batch([[1, 2, 3], ['a', 'b', 'c']], schema=schema)
@@ -235,9 +237,10 @@ def test_same_data_same_hash(self):
 
     def test_different_data_different_hash(self):
         """Test that different data produces different hash"""
-        from src.amp.streaming.idempotency import compute_batch_hash
         import pyarrow as pa
 
+        from src.amp.streaming.idempotency import compute_batch_hash
+
         schema = pa.schema([('id', pa.int64()), ('value', pa.string())])
 
         batch1 = pa.record_batch([[1, 2, 3], ['a', 'b', 'c']], schema=schema)