diff --git a/sqlmesh/integrations/dlt.py b/sqlmesh/integrations/dlt.py index e8c53e7efb..8cb51322e7 100644 --- a/sqlmesh/integrations/dlt.py +++ b/sqlmesh/integrations/dlt.py @@ -1,6 +1,7 @@ import typing as t import click from datetime import datetime, timedelta, timezone +from packaging import version from pydantic import ValidationError from sqlglot import exp, parse_one from sqlmesh.core.config.connection import parse_connection_config @@ -8,6 +9,21 @@ from sqlmesh.utils.date import yesterday_ds +def _get_destination_client(pipeline: t.Any, schema: t.Any) -> t.Any: + """Return the dlt destination client, handling the 1.10.0 API change. + + dlt 1.10.0 replaced the private ``_sql_job_client`` with the public + ``destination_client``. The version must be compared semantically rather + than lexicographically: a string comparison wrongly evaluates + ``"1.9.0" >= "1.10.0"`` as ``True``. + """ + import dlt + + if version.parse(dlt.__version__) >= version.parse("1.10.0"): + return pipeline.destination_client() + return pipeline._sql_job_client(schema) # type: ignore + + def generate_dlt_models_and_settings( pipeline_name: str, dialect: str, @@ -63,10 +79,7 @@ def generate_dlt_models_and_settings( if db_type == "filesystem": connection_config = None else: - if dlt.__version__ >= "1.10.0": - client = pipeline.destination_client() - else: - client = pipeline._sql_job_client(schema) # type: ignore + client = _get_destination_client(pipeline, schema) config = client.config credentials = config.credentials configs = { diff --git a/tests/integrations/test_dlt.py b/tests/integrations/test_dlt.py index ca4f374acb..27ed9bd7cc 100644 --- a/tests/integrations/test_dlt.py +++ b/tests/integrations/test_dlt.py @@ -1,4 +1,53 @@ -from sqlmesh.integrations.dlt import generate_incremental_model +import typing as t + +import pytest + +from sqlmesh.integrations.dlt import _get_destination_client, generate_incremental_model + + +class _FakePipeline: + """Records which destination-client accessor the version gate selects.""" + + def __init__(self) -> None: + self.called: t.Optional[str] = None + + def destination_client(self) -> str: + self.called = "destination_client" + return self.called + + def _sql_job_client(self, schema: t.Any) -> str: + self.called = "_sql_job_client" + return self.called + + +@pytest.mark.parametrize( + "dlt_version, expected", + [ + ("1.9.0", "_sql_job_client"), + ("1.2.0", "_sql_job_client"), + ("1.5.10", "_sql_job_client"), + ("1.10.0", "destination_client"), + ("1.28.1", "destination_client"), + ("2.0.0", "destination_client"), + ], +) +def test_get_destination_client_version_gate( + monkeypatch: pytest.MonkeyPatch, dlt_version: str, expected: str +) -> None: + # The dlt destination-client API changed in 1.10.0: the public + # destination_client() replaced the private _sql_job_client(). The version + # gate must compare semantically (packaging.version.parse), not + # lexicographically. A string compare wrongly treats "1.9.0" >= "1.10.0" as + # True, which would send every 1.x version down the >=1.10.0 branch and make + # the pre-1.10 fallback unreachable. + import dlt + + monkeypatch.setattr(dlt, "__version__", dlt_version) + pipeline = _FakePipeline() + + _get_destination_client(pipeline, schema=None) + + assert pipeline.called == expected def test_generate_incremental_model_filters_on_timestamp_macros() -> None: