Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions sqlmesh/integrations/dlt.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,29 @@
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
from sqlmesh.core.context import Context
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,
Expand Down Expand Up @@ -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 = {
Expand Down
51 changes: 50 additions & 1 deletion tests/integrations/test_dlt.py
Original file line number Diff line number Diff line change
@@ -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:
Expand Down
Loading