Linting fixes

Author: fordN

src/amp/config/label_manager.py

@@ -123,7 +123,7 @@ def hex_to_binary(v):
             )
 
         except FileNotFoundError:
-            raise FileNotFoundError(f'Label CSV file not found: {csv_path}')
+            raise FileNotFoundError(f'Label CSV file not found: {csv_path}') from None
         except Exception as e:
             raise ValueError(f"Failed to load label CSV '{csv_path}': {e}") from e
 

src/amp/loaders/implementations/deltalake_loader.py

@@ -707,7 +707,8 @@ def _handle_reorg(self, invalidation_ranges: List[BlockRange], table_name: str,
                 # Overwrite the table with filtered data
                 self.logger.info(
                     f'Executing blockchain reorg deletion for {len(invalidation_ranges)} networks '
-                    f'in Delta Lake table. Deleting {deleted_count} rows affected by {len(all_affected_batch_ids)} batches.'
+                    f'in Delta Lake table. Deleting {deleted_count} rows affected by '
+                    f'{len(all_affected_batch_ids)} batches.'
                 )
 
                 # Use overwrite mode to replace table contents

src/amp/loaders/implementations/snowflake_loader.py

@@ -659,7 +659,7 @@ def acquire(self, timeout: Optional[float] = 30.0) -> SnowflakeConnection:
 
             return connection
         except Empty:
-            raise RuntimeError(f'Failed to acquire connection from pool within {timeout}s')
+            raise RuntimeError(f'Failed to acquire connection from pool within {timeout}s') from None
 
     def release(self, connection: SnowflakeConnection) -> None:
         """
@@ -772,13 +772,6 @@ def connect(self) -> None:
             else:
                 # Create dedicated connection (legacy behavior)
                 # Set defaults for connection parameters
-                default_params = {
-                    'login_timeout': 60,
-                    'network_timeout': 300,
-                    'socket_timeout': 300,
-                    'validate_default_parameters': True,
-                    'paramstyle': 'qmark',
-                }
 
                 conn_params = {
                     'account': self.config.account,
@@ -920,7 +913,7 @@ def _init_streaming_client(self, table_name: str) -> None:
             raise ImportError(
                 'snowpipe-streaming package required for Snowpipe Streaming. '
                 'Install with: pip install snowpipe-streaming'
-            )
+            ) from None
         except Exception as e:
             self.logger.error(f'Failed to initialize Snowpipe Streaming client for {table_name}: {e}')
             raise
@@ -971,7 +964,8 @@ def _create_streaming_pipe(self, pipe_name: str, table_name: str) -> None:
             """
             self.cursor.execute(create_pipe_sql)
             self.logger.info(
-                f"Created or verified Snowpipe Streaming pipe '{pipe_name}' for table {table_name} with {len(column_info)} columns"
+                f"Created or verified Snowpipe Streaming pipe '{pipe_name}' for table {table_name} "
+                f'with {len(column_info)} columns'
             )
         except Exception as e:
             # Pipe creation might fail if it already exists or if we don't have permissions
@@ -1237,7 +1231,8 @@ def _load_via_stage(self, batch: pa.RecordBatch, table_name: str) -> int:
 
         t_end = time.time()
         self.logger.info(
-            f'Total _load_via_stage took {t_end - t_start:.2f}s for {rows_loaded} rows ({rows_loaded / (t_end - t_start):.0f} rows/sec)'
+            f'Total _load_via_stage took {t_end - t_start:.2f}s for {rows_loaded} rows '
+            f'({rows_loaded / (t_end - t_start):.0f} rows/sec)'
         )
 
         return rows_loaded
@@ -1334,7 +1329,7 @@ def _load_via_pandas(self, batch: pa.RecordBatch, table_name: str) -> int:
             raise ImportError(
                 'pandas and snowflake.connector.pandas_tools are required for pandas loading. '
                 'Install with: pip install pandas'
-            )
+            ) from None
 
         t_start = time.time()
         max_retries = 3  # Retry on transient errors
@@ -1575,7 +1570,10 @@ def _append_with_retry(self, channel: Any, rows: List[Dict[str, Any]]) -> None:
                 import sys
 
                 append_time_ms = (t_append_end - t_append_start) * 1000
-                timing_msg = f'⏱️  Snowpipe append: {len(rows)} rows in {append_time_ms:.2f}ms ({len(rows) / append_time_ms * 1000:.0f} rows/sec)\n'
+                rows_per_sec = len(rows) / append_time_ms * 1000
+                timing_msg = (
+                    f'⏱️  Snowpipe append: {len(rows)} rows in {append_time_ms:.2f}ms ({rows_per_sec:.0f} rows/sec)\n'
+                )
                 sys.stderr.write(timing_msg)
                 sys.stderr.flush()
 
@@ -1649,7 +1647,11 @@ def _load_via_streaming(self, batch: pa.RecordBatch, table_name: str, **kwargs)
             t_batch_end = time.perf_counter()
             batch_time_ms = (t_batch_end - t_batch_start) * 1000
             num_chunks = (batch.num_rows + MAX_ROWS_PER_CHUNK - 1) // MAX_ROWS_PER_CHUNK
-            timing_msg = f'⏱️  Batch load complete: {total_loaded} rows in {batch_time_ms:.2f}ms ({total_loaded / batch_time_ms * 1000:.0f} rows/sec) [{num_chunks} chunks]\n'
+            rows_per_sec = total_loaded / batch_time_ms * 1000
+            timing_msg = (
+                f'⏱️  Batch load complete: {total_loaded} rows in {batch_time_ms:.2f}ms '
+                f'({rows_per_sec:.0f} rows/sec) [{num_chunks} chunks]\n'
+            )
             sys.stderr.write(timing_msg)
             sys.stderr.flush()
 
@@ -1661,7 +1663,10 @@ def _load_via_streaming(self, batch: pa.RecordBatch, table_name: str, **kwargs)
 
             t_batch_end = time.perf_counter()
             batch_time_ms = (t_batch_end - t_batch_start) * 1000
-            timing_msg = f'⏱️  Batch load complete: {len(rows)} rows in {batch_time_ms:.2f}ms ({len(rows) / batch_time_ms * 1000:.0f} rows/sec)\n'
+            rows_per_sec = len(rows) / batch_time_ms * 1000
+            timing_msg = (
+                f'⏱️  Batch load complete: {len(rows)} rows in {batch_time_ms:.2f}ms ({rows_per_sec:.0f} rows/sec)\n'
+            )
             sys.stderr.write(timing_msg)
             sys.stderr.flush()
 

src/amp/streaming/parallel.py

@@ -408,8 +408,9 @@ def _get_resume_adjusted_config(
                         total_gap_blocks = sum(br.end - br.start + 1 for br in gap_ranges)
 
                         log_message = (
-                            f'Resume optimization: Detected {len(gap_ranges)} gap(s) totaling {total_gap_blocks:,} blocks. '
-                            f'Will prioritize gap filling before processing remaining historical range.'
+                            f'Resume optimization: Detected {len(gap_ranges)} gap(s) totaling '
+                            f'{total_gap_blocks:,} blocks. Will prioritize gap filling before '
+                            f'processing remaining historical range.'
                         )
 
                         return config, resume_watermark, log_message

tests/integration/test_checkpoint_resume.py

@@ -406,7 +406,7 @@ def test_processed_ranges_table_creation(self, checkpoint_db_connection):
             enabled=True,
             table_prefix='test_amp_',
         )
-        store = DatabaseProcessedRangesStore(config, checkpoint_db_connection)
+        DatabaseProcessedRangesStore(config, checkpoint_db_connection)
 
         cursor = checkpoint_db_connection.cursor()
         cursor.execute("""

tests/integration/test_resilient_streaming.py

@@ -362,7 +362,7 @@ def test_all_resilience_features_together(self):
         batch = pa.record_batch([[1]], schema=schema)
 
         # Multiple successful loads with retries
-        for i in range(3):
+        for _i in range(3):
             # Reset failure mode for each iteration
             loader.current_attempt = 0
 

tests/integration/test_snowflake_loader.py

@@ -1075,7 +1075,8 @@ def test_streaming_performance(
             print(f'  Throughput: {rows_per_second:,.0f} rows/sec')
             print(f'  Loading method: {result.metadata.get("loading_method")}')
 
-            # Wait for Snowpipe streaming data to become queryable (eventual consistency, larger dataset may take longer)
+            # Wait for Snowpipe streaming data to become queryable
+            # (eventual consistency, larger dataset may take longer)
             count = wait_for_snowpipe_data(loader, test_table_name, performance_test_data.num_rows, max_wait=60)
             assert count == performance_test_data.num_rows
 

tests/unit/test_streaming_types.py

@@ -276,7 +276,7 @@ def test_from_flight_data_ranges_complete_default(self):
 
 @pytest.mark.unit
 class TestResponseBatch:
-    """Test ResponseBatch properties"""
+    """Test ResponseBatch factory methods and properties"""
 
     def test_num_rows_property(self):
         """Test num_rows property delegates to data"""
@@ -305,11 +305,6 @@ def test_networks_property(self):
         assert len(networks) == 2
         assert set(networks) == {'ethereum', 'polygon'}
 
-
-@pytest.mark.unit
-class TestResponseBatch:
-    """Test ResponseBatch factory methods and properties"""
-
     def test_data_batch_creation(self):
         """Test creating a data batch response"""
         data = pa.record_batch([pa.array([1])], names=['id'])