Fix linting/formatting issues in loader test files

Author: fordN

docs/implementing_data_loaders.md

@@ -781,5 +781,3 @@ class KeyValueLoader(DataLoader[KeyValueConfig]):
             'database': self.config.database
         }
 ```
-
-This documentation provides everything needed to implement new data loaders efficiently and consistently!

tests/integration/loaders/backends/test_deltalake.py

@@ -91,7 +91,6 @@ class TestDeltaLakeSpecific:
 
     def test_partitioning(self, delta_partitioned_config, small_test_data):
         """Test DeltaLake partitioning functionality"""
-        import pyarrow as pa
 
         loader = DeltaLakeLoader(delta_partitioned_config)
 
@@ -148,7 +147,7 @@ def test_schema_evolution(self, delta_basic_config, small_test_data):
             # Load with new schema (if schema evolution enabled)
             from src.amp.loaders.base import LoadMode
 
-            result2 = loader.load_table(extended_table, 'test_table', mode=LoadMode.APPEND)
+            loader.load_table(extended_table, 'test_table', mode=LoadMode.APPEND)
 
             # Result depends on merge_schema configuration
             dt = DeltaTable(loader.config.table_path)

tests/integration/loaders/backends/test_iceberg.py

@@ -10,9 +10,6 @@
 import pytest
 
 try:
-    from pyiceberg.catalog import load_catalog
-    from pyiceberg.schema import Schema
-
     from src.amp.loaders.implementations.iceberg_loader import IcebergLoader
     from tests.integration.loaders.conftest import LoaderTestConfig
     from tests.integration.loaders.test_base_loader import BaseLoaderTests
@@ -102,7 +99,6 @@ def test_catalog_initialization(self, iceberg_basic_config):
 
     def test_partitioning(self, iceberg_basic_config, small_test_data):
         """Test Iceberg partitioning (partition spec)"""
-        import pyarrow as pa
 
         # Create config with partitioning
         config = {**iceberg_basic_config, 'partition_spec': [('year', 'identity'), ('month', 'identity')]}
@@ -147,7 +143,7 @@ def test_schema_evolution(self, iceberg_basic_config, small_test_data):
             extended_table = pa.Table.from_pydict(extended_data)
 
             # Load with new schema
-            result2 = loader.load_table(extended_table, table_name, mode=LoadMode.APPEND)
+            loader.load_table(extended_table, table_name, mode=LoadMode.APPEND)
 
             # Schema evolution depends on config
             catalog = loader._catalog
@@ -161,9 +157,10 @@ def test_schema_evolution(self, iceberg_basic_config, small_test_data):
 
     def test_timestamp_conversion(self, iceberg_basic_config):
         """Test timestamp conversion for Iceberg"""
-        import pyarrow as pa
         from datetime import datetime
 
+        import pyarrow as pa
+
         # Create data with timestamps
         data = {
             'id': [1, 2, 3],
@@ -193,7 +190,6 @@ def test_timestamp_conversion(self, iceberg_basic_config):
 
     def test_multiple_tables(self, iceberg_basic_config, small_test_data):
         """Test managing multiple tables with same loader"""
-        from src.amp.loaders.base import LoadMode
 
         loader = IcebergLoader(iceberg_basic_config)
 

tests/integration/loaders/backends/test_lmdb.py

@@ -11,8 +11,6 @@
 import pytest
 
 try:
-    import lmdb
-
     from src.amp.loaders.implementations.lmdb_loader import LMDBLoader
     from tests.integration.loaders.conftest import LoaderTestConfig
     from tests.integration.loaders.test_base_loader import BaseLoaderTests
@@ -37,7 +35,7 @@ def get_row_count(self, loader: LMDBLoader, table_name: str) -> int:
         count = 0
         with loader.env.begin(db=loader.db) as txn:
             cursor = txn.cursor()
-            for key, value in cursor:
+            for _key, _value in cursor:
                 count += 1
         return count
 
@@ -58,7 +56,7 @@ def query_rows(
                     row_data = json.loads(value.decode())
                     row_data['_key'] = key.decode()
                     rows.append(row_data)
-                except:
+                except Exception:
                     # Fallback to raw value
                     rows.append({'_key': key.decode(), '_value': value.decode()})
         return rows
@@ -79,20 +77,20 @@ def get_column_names(self, loader: LMDBLoader, table_name: str) -> List[str]:
 
         with loader.env.begin(db=loader.db) as txn:
             cursor = txn.cursor()
-            for key, value in cursor:
+            for _key, value in cursor:
                 try:
                     row_data = json.loads(value.decode())
                     return list(row_data.keys())
-                except:
+                except Exception:
                     return ['_value']  # Fallback
         return []
 
 
 @pytest.fixture
 def lmdb_test_env():
     """Create and cleanup temporary directory for LMDB databases"""
-    import tempfile
     import shutil
+    import tempfile
 
     temp_dir = tempfile.mkdtemp(prefix='lmdb_test_')
     yield temp_dir

tests/integration/loaders/backends/test_postgresql.py

@@ -70,7 +70,7 @@ def query_rows(
                 rows = cur.fetchall()
 
                 # Convert to list of dicts
-                return [dict(zip(columns, row)) for row in rows]
+                return [dict(zip(columns, row, strict=False)) for row in rows]
         finally:
             loader.pool.putconn(conn)
 
@@ -318,9 +318,10 @@ class TestPostgreSQLPerformance:
 
     def test_large_data_loading(self, postgresql_test_config, test_table_name, cleanup_tables):
         """Test loading large datasets"""
-        import pyarrow as pa
         from datetime import datetime
 
+        import pyarrow as pa
+
         cleanup_tables.append(test_table_name)
 
         # Create large dataset

tests/integration/loaders/backends/test_redis.py

@@ -54,7 +54,7 @@ def get_row_count(self, loader: RedisLoader, table_name: str) -> int:
             # Get stream length
             try:
                 return loader.redis_client.xlen(table_name)
-            except:
+            except Exception:
                 return 0
         else:
             # For other structures, scan for keys
@@ -166,7 +166,6 @@ class TestRedisSpecific:
 
     def test_hash_storage(self, redis_test_config, small_test_data, cleanup_redis):
         """Test Redis hash data structure storage"""
-        import pyarrow as pa
 
         keys_to_clean, patterns_to_clean = cleanup_redis
         table_name = 'test_hash'
@@ -343,7 +342,7 @@ def test_data_structure_comparison(self, redis_test_config, comprehensive_test_d
                 }
 
         # Verify all structures work
-        for structure, data in results.items():
+        for _structure, data in results.items():
             assert data['success'] == True
             assert data['rows_loaded'] == 1000
 

tests/integration/loaders/backends/test_snowflake.py

@@ -53,7 +53,7 @@ def query_rows(
             cur.execute(query)
             columns = [col[0] for col in cur.description]
             rows = cur.fetchall()
-            return [dict(zip(columns, row)) for row in rows]
+            return [dict(zip(columns, row, strict=False)) for row in rows]
 
     def cleanup_table(self, loader: SnowflakeLoader, table_name: str) -> None:
         """Drop Snowflake table"""
@@ -87,7 +87,6 @@ class TestSnowflakeSpecific:
 
     def test_stage_loading_method(self, snowflake_config, small_test_table, test_table_name, cleanup_tables):
         """Test Snowflake stage-based loading (Snowflake-specific optimization)"""
-        import pyarrow as pa
 
         cleanup_tables.append(test_table_name)
 

tests/integration/loaders/conftest.py

@@ -141,11 +141,7 @@ def loader(self, request):
             loader_config = loader_config.copy()
             # Only enable state if not explicitly configured
             if 'state' not in loader_config:
-                loader_config['state'] = {
-                    'enabled': True,
-                    'storage': 'memory',
-                    'store_batch_id': True
-                }
+                loader_config['state'] = {'enabled': True, 'storage': 'memory', 'store_batch_id': True}
 
         # Create and return the loader instance
         return self.config.loader_class(loader_config)

tests/integration/loaders/test_base_streaming.py

@@ -138,7 +138,9 @@ def test_reorg_deletion(self, loader, test_table_name, cleanup_tables):
             reorg_response = ResponseBatch.reorg_batch(
                 invalidation_ranges=[BlockRange(network='ethereum', start=104, end=108)]
             )
-            reorg_results = list(loader.load_stream_continuous(iter([reorg_response]), test_table_name, connection_name='test_conn'))
+            reorg_results = list(
+                loader.load_stream_continuous(iter([reorg_response]), test_table_name, connection_name='test_conn')
+            )
             assert len(reorg_results) == 1
             assert reorg_results[0].success
 
@@ -174,7 +176,9 @@ def test_reorg_overlapping_ranges(self, loader, test_table_name, cleanup_tables)
             )
 
             # Load via streaming API (with connection_name for state tracking)
-            results = list(loader.load_stream_continuous(iter([response]), test_table_name, connection_name='test_conn'))
+            results = list(
+                loader.load_stream_continuous(iter([response]), test_table_name, connection_name='test_conn')
+            )
             assert len(results) == 1
             assert results[0].success
 
@@ -187,7 +191,9 @@ def test_reorg_overlapping_ranges(self, loader, test_table_name, cleanup_tables)
             reorg_response = ResponseBatch.reorg_batch(
                 invalidation_ranges=[BlockRange(network='ethereum', start=160, end=180)]
             )
-            reorg_results = list(loader.load_stream_continuous(iter([reorg_response]), test_table_name, connection_name='test_conn'))
+            reorg_results = list(
+                loader.load_stream_continuous(iter([reorg_response]), test_table_name, connection_name='test_conn')
+            )
             assert len(reorg_results) == 1
             assert reorg_results[0].success
 
@@ -248,7 +254,9 @@ def test_reorg_multi_network(self, loader, test_table_name, cleanup_tables):
             reorg_response = ResponseBatch.reorg_batch(
                 invalidation_ranges=[BlockRange(network='ethereum', start=100, end=100)]
             )
-            reorg_results = list(loader.load_stream_continuous(iter([reorg_response]), test_table_name, connection_name='test_conn'))
+            reorg_results = list(
+                loader.load_stream_continuous(iter([reorg_response]), test_table_name, connection_name='test_conn')
+            )
             assert len(reorg_results) == 1
             assert reorg_results[0].success