Synapse

dlt/destinations/synapse/README.md

@@ -0,0 +1,5 @@
+# loader account setup
+
+1. Create new database `CREATE DATABASE dlt_data`
+2. Create new user, set password `CREATE USER loader WITH PASSWORD 'loader';`
+3. Set as database owner (we could set lower permission) `ALTER DATABASE dlt_data OWNER TO loader`

dlt/destinations/synapse/__init__.py

@@ -0,0 +1,51 @@
+from typing import Type
+
+from dlt.common.schema.schema import Schema
+from dlt.common.configuration import with_config, known_sections
+from dlt.common.configuration.accessors import config
+from dlt.common.data_writers.escape import escape_postgres_identifier, escape_mssql_literal
+from dlt.common.destination import DestinationCapabilitiesContext
+from dlt.common.destination.reference import JobClientBase, DestinationClientConfiguration
+from dlt.common.arithmetics import DEFAULT_NUMERIC_PRECISION, DEFAULT_NUMERIC_SCALE
+from dlt.common.wei import EVM_DECIMAL_PRECISION
+
+from dlt.destinations.synapse.configuration import SynapseClientConfiguration
+
+
+@with_config(spec=SynapseClientConfiguration, sections=(known_sections.DESTINATION, "synapse",))
+def _configure(config: SynapseClientConfiguration = config.value) -> SynapseClientConfiguration:
+    return config
+
+
+def capabilities() -> DestinationCapabilitiesContext:
+    # https://learn.microsoft.com/en-us/azure/synapse-analytics/sql-data-warehouse/sql-data-warehouse-service-capacity-limits
+    caps = DestinationCapabilitiesContext()
+    caps.preferred_loader_file_format = "insert_values"
+    caps.supported_loader_file_formats = ["insert_values"]
+    caps.preferred_staging_file_format = "parquet"
+    caps.supported_staging_file_formats = ["parquet"]
+    caps.escape_identifier = escape_postgres_identifier
+    caps.escape_literal = escape_mssql_literal
+    caps.decimal_precision = (DEFAULT_NUMERIC_PRECISION, DEFAULT_NUMERIC_SCALE)
+    caps.wei_precision = (DEFAULT_NUMERIC_PRECISION, 0)
+    caps.max_identifier_length = 128
+    caps.max_column_identifier_length = 128
+    caps.max_query_length = 4 * 1024 * 64 * 1024
+    caps.is_max_query_length_in_bytes = True
+    caps.max_text_data_type_length = 4000
+    caps.is_max_text_data_type_length_in_bytes = False
+    caps.supports_ddl_transactions = False
+    caps.max_rows_per_insert = 1000
+
+    return caps
+
+
+def client(schema: Schema, initial_config: DestinationClientConfiguration = config.value) -> JobClientBase:
+    # import client when creating instance so capabilities and config specs can be accessed without dependencies installed
+    from dlt.destinations.synapse.synapse import SynapseClient
+
+    return SynapseClient(schema, _configure(initial_config))  # type: ignore[arg-type]
+
+
+def spec() -> Type[DestinationClientConfiguration]:
+    return SynapseClientConfiguration

dlt/destinations/synapse/configuration.py

@@ -0,0 +1,89 @@
+from typing import Final, ClassVar, Any, List, Optional
+from sqlalchemy.engine import URL
+
+from dlt.common.configuration import configspec
+from dlt.common.configuration.specs import ConnectionStringCredentials
+from dlt.common.utils import digest128
+from dlt.common.typing import TSecretValue
+from dlt.common.exceptions import SystemConfigurationException
+
+from dlt.common.destination.reference import DestinationClientDwhWithStagingConfiguration
+
+
+@configspec
+class SynapseCredentials(ConnectionStringCredentials):
+    drivername: Final[str] = "sqlserver"  # type: ignore
+    user:  Final[str] = "dltadmin"
+    password: TSecretValue
+    host: str
+    port: int = 1433
+    connect_timeout: int = 15
+    odbc_driver: str = "ODBC Driver 17 for SQL Server"
+
+    __config_gen_annotations__: ClassVar[List[str]] = ["port", "connect_timeout"]
+
+    def parse_native_representation(self, native_value: Any) -> None:
+        # TODO: Support ODBC connection string or sqlalchemy URL
+        super().parse_native_representation(native_value)
+        self.connect_timeout = int(self.query.get("connect_timeout", self.connect_timeout))
+        if not self.is_partial():
+            self.resolve()
+
+    def on_resolved(self) -> None:
+        self.database = self.database.lower()
+
+    def to_url(self) -> URL:
+        url = super().to_url()
+        url.update_query_pairs([("connect_timeout", str(self.connect_timeout))])
+        return url
+
+    def on_partial(self) -> None:
+        self.odbc_driver = self._get_odbc_driver()
+        if not self.is_partial():
+            self.resolve()
+
+    def _get_odbc_driver(self) -> str:
+        if self.odbc_driver:
+            return self.odbc_driver
+        # Pick a default driver if available
+        supported_drivers = ['ODBC Driver 18 for SQL Server', 'ODBC Driver 17 for SQL Server']
+        import pyodbc
+        available_drivers = pyodbc.drivers()
+        for driver in supported_drivers:
+            if driver in available_drivers:
+                return driver
+        docs_url = "https://learn.microsoft.com/en-us/sql/connect/odbc/download-odbc-driver-for-sql-server?view=sql-server-ver16"
+        raise SystemConfigurationException(
+            f"No supported ODBC driver found for MS SQL Server.  "
+            f"See {docs_url} for information on how to install the '{supported_drivers[0]}' on your platform."
+        )
+
+    def to_odbc_dsn(self) -> str:
+        params = {
+            "DRIVER": self.odbc_driver,
+            "SERVER": self.host,
+            "PORT": self.port,
+            "DATABASE": self.database,
+            "UID": self.username,
+            "PWD": self.password,
+            "LongAsMax": "yes",
+            "MARS_Connection": "yes"
+        }
+        if self.query:
+            params.update(self.query)
+        return ";".join([f"{k}={v}" for k, v in params.items()])
+
+
+
+@configspec
+class SynapseClientConfiguration(DestinationClientDwhWithStagingConfiguration):
+    destination_name: Final[str] = "synapse"  # type: ignore
+    credentials: SynapseCredentials
+
+    create_indexes: bool = False
+
+    def fingerprint(self) -> str:
+        """Returns a fingerprint of host part of a connection string"""
+        if self.credentials and self.credentials.host:
+            return digest128(self.credentials.host)
+        return ""