snowflake_loader: Snowpipe streaming support

Author: fordN

docs/snowflake_loader.md

@@ -12,21 +12,17 @@ dependencies = [
     "pandas>=2.3.1",
     "pyarrow>=20.0.0",
     "typer>=0.15.2",
-
     # Flight SQL support
     "adbc-driver-manager>=1.5.0",
     "adbc-driver-postgresql>=1.5.0",
     "protobuf>=4.21.0",
-
     # Ethereum/blockchain utilities
     "base58>=2.1.1",
     "eth-hash[pysha3]>=0.7.1",
     "eth-utils>=5.2.0",
-
     # Google Cloud support
     "google-cloud-bigquery>=3.30.0",
     "google-cloud-storage>=3.1.0",
-
     # Arro3 for enhanced PyArrow operations
     "arro3-core>=0.5.1",
     "arro3-compute>=0.5.1",
@@ -59,6 +55,7 @@ iceberg = [
 
 snowflake = [
     "snowflake-connector-python>=3.5.0",
+    "snowpipe-streaming>=1.0.0",  # Snowpipe Streaming API
 ]
 
 lmdb = [
@@ -72,6 +69,7 @@ all_loaders = [
     "pyiceberg[sql-sqlite]>=0.10.0", # Apache Iceberg
     "pydantic>=2.0,<2.12",          # PyIceberg 0.10.0 compatibility
     "snowflake-connector-python>=3.5.0",  # Snowflake
+    "snowpipe-streaming>=1.0.0",        # Snowpipe Streaming API
     "lmdb>=1.4.0",                  # LMDB
 ]
 

pyproject.toml

@@ -211,6 +211,13 @@ def query_and_load(
             **{k: v for k, v in kwargs.items() if k in ['max_retries', 'retry_delay']},
         )
 
+        # Remove known LoadConfig params from kwargs, leaving loader-specific params
+        for key in ['max_retries', 'retry_delay']:
+            kwargs.pop(key, None)
+
+        # Remaining kwargs are loader-specific (e.g., channel_suffix for Snowflake)
+        loader_specific_kwargs = kwargs
+
         if read_all:
             self.logger.info(f'Loading entire query result to {loader_type}:{destination}')
         else:
@@ -221,20 +228,20 @@ def query_and_load(
         # Get the data and load
         if read_all:
             table = self.get_sql(query, read_all=True)
-            return self._load_table(table, loader_type, destination, loader_config, load_config)
+            return self._load_table(table, loader_type, destination, loader_config, load_config, **loader_specific_kwargs)
         else:
             batch_stream = self.get_sql(query, read_all=False)
-            return self._load_stream(batch_stream, loader_type, destination, loader_config, load_config)
+            return self._load_stream(batch_stream, loader_type, destination, loader_config, load_config, **loader_specific_kwargs)
 
     def _load_table(
-        self, table: pa.Table, loader: str, table_name: str, config: Dict[str, Any], load_config: LoadConfig
+        self, table: pa.Table, loader: str, table_name: str, config: Dict[str, Any], load_config: LoadConfig, **kwargs
     ) -> LoadResult:
         """Load a complete Arrow Table"""
         try:
             loader_instance = create_loader(loader, config)
 
             with loader_instance:
-                return loader_instance.load_table(table, table_name, **load_config.__dict__)
+                return loader_instance.load_table(table, table_name, **load_config.__dict__, **kwargs)
         except Exception as e:
             self.logger.error(f'Failed to load table: {e}')
             return LoadResult(
@@ -254,13 +261,14 @@ def _load_stream(
         table_name: str,
         config: Dict[str, Any],
         load_config: LoadConfig,
+        **kwargs,
     ) -> Iterator[LoadResult]:
         """Load from a stream of batches"""
         try:
             loader_instance = create_loader(loader, config)
 
             with loader_instance:
-                yield from loader_instance.load_stream(batch_stream, table_name, **load_config.__dict__)
+                yield from loader_instance.load_stream(batch_stream, table_name, **load_config.__dict__, **kwargs)
         except Exception as e:
             self.logger.error(f'Failed to load stream: {e}')
             yield LoadResult(