fix: service principal auth support for synapse copy job

dlt/destinations/impl/synapse/synapse.py

@@ -1,11 +1,9 @@
 import os
-from typing import ClassVar, Sequence, List, Dict, Any, Optional, cast
+from typing import ClassVar, Sequence, List, Dict, Any, Optional, cast, Union
 from copy import deepcopy
 from textwrap import dedent
 from urllib.parse import urlparse, urlunparse
 
-from dlt import current
-
 from dlt.common.destination import DestinationCapabilitiesContext
 from dlt.common.destination.reference import (
     SupportsStagingDestination,
@@ -16,13 +14,15 @@
 from dlt.common.schema.utils import (
     table_schema_has_type,
     get_inherited_table_hint,
-    is_complete_column,
 )
 
-from dlt.common.configuration.specs import AzureCredentialsWithoutDefaults
+from dlt.common.configuration.exceptions import ConfigurationException
+from dlt.common.configuration.specs import (
+    AzureCredentialsWithoutDefaults,
+    AzureServicePrincipalCredentialsWithoutDefaults,
+)
 
 from dlt.destinations.job_impl import NewReferenceJob
-from dlt.destinations.sql_jobs import SqlStagingCopyJob, SqlJobParams
 from dlt.destinations.sql_client import SqlClientBase
 from dlt.destinations.job_client_impl import SqlJobClientBase, LoadJob, CopyRemoteFileLoadJob
 from dlt.destinations.exceptions import LoadJobTerminalException
@@ -163,7 +163,7 @@ def start_file_load(self, table: TTableSchema, file_path: str, load_id: str) ->
                 table,
                 file_path,
                 self.sql_client,
-                cast(AzureCredentialsWithoutDefaults, self.config.staging_config.credentials),
+                self.config.staging_config.credentials,  # type: ignore[arg-type]
                 self.config.staging_use_msi,
             )
         return job
@@ -175,7 +175,9 @@ def __init__(
         table: TTableSchema,
         file_path: str,
         sql_client: SqlClientBase[Any],
-        staging_credentials: Optional[AzureCredentialsWithoutDefaults] = None,
+        staging_credentials: Optional[
+            Union[AzureCredentialsWithoutDefaults, AzureServicePrincipalCredentialsWithoutDefaults]
+        ] = None,
         staging_use_msi: bool = False,
     ) -> None:
         self.staging_use_msi = staging_use_msi
@@ -204,16 +206,32 @@ def execute(self, table: TTableSchema, bucket_path: str) -> None:
 
         staging_credentials = self._staging_credentials
         assert staging_credentials is not None
-        assert isinstance(staging_credentials, AzureCredentialsWithoutDefaults)
+        assert isinstance(
+            staging_credentials,
+            (AzureCredentialsWithoutDefaults, AzureServicePrincipalCredentialsWithoutDefaults),
+        )
         azure_storage_account_name = staging_credentials.azure_storage_account_name
         https_path = self._get_https_path(bucket_path, azure_storage_account_name)
         table_name = table["name"]
 
         if self.staging_use_msi:
             credential = "IDENTITY = 'Managed Identity'"
         else:
-            sas_token = staging_credentials.azure_storage_sas_token
-            credential = f"IDENTITY = 'Shared Access Signature', SECRET = '{sas_token}'"
+            # re-use staging credentials for copy into Synapse
+            if isinstance(staging_credentials, AzureCredentialsWithoutDefaults):
+                sas_token = staging_credentials.azure_storage_sas_token
+                credential = f"IDENTITY = 'Shared Access Signature', SECRET = '{sas_token}'"
+            elif isinstance(staging_credentials, AzureServicePrincipalCredentialsWithoutDefaults):
+                tenant_id = staging_credentials.azure_tenant_id
+                endpoint = f"https://login.microsoftonline.com/{tenant_id}/oauth2/token"
+                identity = f"{staging_credentials.azure_client_id}@{endpoint}"
+                secret = staging_credentials.azure_client_secret
+                credential = f"IDENTITY = '{identity}', SECRET = '{secret}'"
+            else:
+                raise ConfigurationException(
+                    f"Credentials of type `{type(staging_credentials)}` not supported"
+                    " when loading data from staging into Synapse using `COPY INTO`."
+                )
 
         # Copy data from staging file into Synapse table.
         with self._sql_client.begin_transaction():

dlt/extract/incremental/transform.py

@@ -43,6 +43,7 @@ class IncrementalTransform:
     Subclasses must implement the `__call__` method which will be called
     for each data item in the extracted data.
     """
+
     def __init__(
         self,
         resource_name: str,
@@ -110,6 +111,7 @@ def deduplication_disabled(self) -> bool:
 
 class JsonIncremental(IncrementalTransform):
     """Extracts incremental data from JSON data items."""
+
     def find_cursor_value(self, row: TDataItem) -> Any:
         """Finds value in row at cursor defined by self.cursor_path.
 

tests/load/pipeline/test_synapse.py

@@ -0,0 +1,44 @@
+from typing import Dict, Any, Union
+
+import pytest
+
+import dlt
+from dlt.destinations import filesystem, synapse
+from dlt.common.configuration.specs.azure_credentials import (
+    AzureCredentialsWithoutDefaults,
+    AzureServicePrincipalCredentialsWithoutDefaults,
+)
+
+from tests.utils import skip_if_not_active
+from tests.pipeline.utils import assert_load_info
+from tests.load.utils import AZ_BUCKET
+
+
+skip_if_not_active("synapse")
+
+
+@pytest.mark.parametrize("credentials_type", ("sas", "service_principal", "managed_identity"))
+def test_copy_file_load_job_credentials(credentials_type: str) -> None:
+    staging_creds: Union[
+        AzureCredentialsWithoutDefaults, AzureServicePrincipalCredentialsWithoutDefaults
+    ]
+    if credentials_type == "service_principal":
+        staging_creds = AzureServicePrincipalCredentialsWithoutDefaults(
+            **dlt.secrets.get("destination.fsazureprincipal.credentials")
+        )
+    else:
+        FS_CREDS: Dict[str, Any] = dlt.secrets.get("destination.filesystem.credentials")
+        staging_creds = AzureCredentialsWithoutDefaults(
+            azure_storage_account_name=FS_CREDS["azure_storage_account_name"],
+            azure_storage_account_key=FS_CREDS["azure_storage_account_key"],
+        )
+
+    pipeline = dlt.pipeline(
+        staging=filesystem(bucket_url=AZ_BUCKET, credentials=staging_creds),
+        destination=synapse(
+            staging_use_msi=(True if credentials_type == "managed_identity" else False)
+        ),
+    )
+
+    info = pipeline.run([{"foo": "bar"}], table_name="abstract")
+    assert_load_info(info)