Add HMAC credentials and update Clickhouse configuration

Signed-off-by: Marcel Coetzee <[email protected]>

Author: Marcel Coetzee

dlt/common/configuration/specs/__init__.py

@@ -21,6 +21,7 @@
 from .api_credentials import OAuth2Credentials
 from .aws_credentials import AwsCredentials, AwsCredentialsWithoutDefaults
 from .azure_credentials import AzureCredentials, AzureCredentialsWithoutDefaults
+from .hmac_credentials import HMACCredentials
 
 
 # backward compatibility for service account credentials
@@ -53,4 +54,5 @@
     "AzureCredentialsWithoutDefaults",
     "GcpClientCredentials",
     "GcpClientCredentialsWithDefault",
-]
+    "HMACCredentials",
+]

dlt/common/configuration/specs/hmac_credentials.py

@@ -0,0 +1,41 @@
+from typing import Optional, Dict
+
+from dlt.common.configuration.specs import (
+    CredentialsConfiguration,
+    configspec,
+)
+from dlt.common.typing import TSecretStrValue, DictStrAny
+
+
+@configspec
+class HMACCredentials(CredentialsConfiguration):
+    aws_access_key_id: str = None
+    aws_secret_access_key: TSecretStrValue = None
+    aws_session_token: Optional[TSecretStrValue] = None
+    profile_name: Optional[str] = None
+    region_name: Optional[str] = None
+    endpoint_url: Optional[str] = None
+
+    def to_s3fs_credentials(self) -> Dict[str, Optional[str]]:
+        """Dict of keyword arguments that can be passed to s3fs"""
+        credentials: DictStrAny = dict(
+            key=self.aws_access_key_id,
+            secret=self.aws_secret_access_key,
+            token=self.aws_session_token,
+            profile=self.profile_name,
+            endpoint_url=self.endpoint_url,
+        )
+        if self.region_name:
+            credentials["client_kwargs"] = {"region_name": self.region_name}
+        return credentials
+
+    def to_native_representation(self) -> Dict[str, Optional[str]]:
+        """Return a dict that can be passed as kwargs to boto3 session"""
+        return dict(self)
+
+    def to_session_credentials(self) -> Dict[str, str]:
+        return dict(
+            aws_access_key_id=self.aws_access_key_id,
+            aws_secret_access_key=self.aws_secret_access_key,
+            aws_session_token=self.aws_session_token,
+        )

dlt/destinations/impl/clickhouse/clickhouse.py

@@ -7,15 +7,14 @@
 import clickhouse_connect
 from clickhouse_connect.driver.tools import insert_file
 
-import dlt
 from dlt import config
 from dlt.common.configuration.specs import (
     CredentialsConfiguration,
     AzureCredentialsWithoutDefaults,
     GcpCredentials,
     AwsCredentialsWithoutDefaults,
+    HMACCredentials,
 )
-from dlt.destinations.exceptions import DestinationTransientException
 from dlt.common.destination import DestinationCapabilitiesContext
 from dlt.common.destination.reference import (
     SupportsStagingDestination,
@@ -34,7 +33,7 @@
     TColumnSchemaBase,
 )
 from dlt.common.storages import FileStorage
-from dlt.destinations.exceptions import LoadJobTerminalException
+from dlt.destinations.exceptions import DestinationTransientException, LoadJobTerminalException
 from dlt.destinations.impl.clickhouse import capabilities
 from dlt.destinations.impl.clickhouse.clickhouse_adapter import (
     TTableEngineType,
@@ -209,13 +208,15 @@ def __init__(
                 access_key_id = staging_credentials.aws_access_key_id
                 secret_access_key = staging_credentials.aws_secret_access_key
             elif isinstance(staging_credentials, GcpCredentials):
-                access_key_id = client.credentials.gcp_access_key_id
-                secret_access_key = client.credentials.gcp_secret_access_key
+                # Defer to the provided HMAC Credentials.
+                access_key_id = client.credentials.gcp_access_key_id  # type: ignore
+                secret_access_key = client.credentials.gcp_secret_access_key  # type: ignore
                 if not access_key_id or not secret_access_key:
                     raise DestinationTransientException(
-                        "You have tried loading from gcs with clickhouse. Please provide valid"
-                        " 'gcp_access_key_id' and 'gcp_secret_access_key' to connect to gcs as"
-                        " outlined in the dlthub docs."
+                        "You have tried loading from gcs with clickhouse. "
+                        "Please provide valid HMAC credentials as outlined in "
+                        "the dlthub docs:\n"
+                        "https://dlthub.com/devel/dlt-ecosystem/destinations/clickhouse#using-google-cloud-storage-as-a-staging-area."
                     )
 
             auth = "NOSIGN"

dlt/destinations/impl/clickhouse/configuration.py

@@ -38,10 +38,6 @@ class ClickHouseCredentials(ConnectionStringCredentials):
     """Separator for dataset table names, defaults to '___', i.e. 'database.dataset___table'."""
     dataset_sentinel_table_name: str = "dlt_sentinel_table"
     """Special table to mark dataset as existing"""
-    gcp_access_key_id: Optional[str] = None
-    """When loading from a gcp bucket, you need to provide gcp interoperable keys"""
-    gcp_secret_access_key: Optional[str] = None
-    """When loading from a gcp bucket, you need to provide gcp interoperable keys"""
 
     __config_gen_annotations__: ClassVar[List[str]] = [
         "host",