fixup(dbt): Fix DBT snapshot integration

ibis_typing/dbt/dbt_snapshot.py

@@ -18,7 +18,7 @@
 
 
 @frozen(kw_only=True)
-class DbtSnapshot[E: IbisSchema](DbtModel):
+class DbtSnapshot[E: DbtSnapshotAbstract](DbtModel):
     """Integration mark-up for DBT snapshot function."""
 
     expr: type[E]

ibis_typing/dbt/dbt_sql_compiler.py

@@ -55,7 +55,7 @@ def dbt_model_to_dbt_sql(
 def dbt_snapshot_to_dbt_sql(
     snapshot: DbtSnapshot, dialect: str, *, ref_provider: DbtRefTableProvider
 ) -> str:
-    exported = snapshot.expr @ AsRevertible(ExpressionExport)
+    exported = snapshot.expr.origin @ AsRevertible(ExpressionExport)
     snapshot_table = dbt_ibis_constructor.construct_dbt_model(exported, ref_provider)
 
     table_sql = snapshot_table.table.to_sql(dialect)

ibis_typing/samples/dbt_models.py

@@ -10,7 +10,10 @@
     dbt_model_resolver,
 )
 from ibis_typing.dbt.dbt_model import IncrementalStrategy, Materialized
-from ibis_typing.dbt.dbt_snapshot import SnapshotStrategy
+from ibis_typing.dbt.dbt_snapshot import (
+    DbtSnapshotAbstract,
+    SnapshotStrategy,
+)
 from ibis_typing.ibis_time import TimestampNow
 from ibis_typing.samples.sample_incremental_calendar import Calendar, CalendarWidth
 from ibis_typing.samples.sample_transforms import CircleParameters
@@ -28,7 +31,7 @@ def get_dbt_model_lookup() -> Mapping[type[Expression], DbtModel]:
             config=ModelConfig(
                 materialized=Materialized.table,
             ).with_namespace(*my_namespace),
-        )
+        ),
     }
     incremental_models = [
         DbtModel(
@@ -42,14 +45,7 @@ def get_dbt_model_lookup() -> Mapping[type[Expression], DbtModel]:
         for expr in resolver.incremental_models
     ]
     snapshots = [
-        DbtSnapshot(
-            expr=CalendarWidth,
-            config=SnapshotConfig(
-                strategy=SnapshotStrategy.timestamp,
-                unique_key=CalendarWidth.incremental_params.group_by,
-                updated_at=CalendarWidth.incremental_params.updated_at_col,
-            ).with_namespace(*my_namespace),
-        )
+        DbtSnapshot(expr=expr, config=expr.config) for expr in resolver.snapshots
     ]
     all_models = (*models, *incremental_models, *snapshots)
 
@@ -67,3 +63,12 @@ def as_db_schema(schema: type[IbisSchema]) -> type[IbisDbSchema]:
         "table_namespace": my_namespace,
     }
     return type(schema.__name__, (IbisDbSchema, schema), kwargs)
+
+
+class CalendarWidthSnapshot(DbtSnapshotAbstract):
+    origin = CalendarWidth
+    config = SnapshotConfig(
+        strategy=SnapshotStrategy.timestamp,
+        unique_key=CalendarWidth.incremental_params.group_by,
+        updated_at=CalendarWidth.incremental_params.updated_at_col,
+    ).with_namespace(*my_namespace)

ibis_typing/samples/generated/__ibis_sql/calendar_width.sql

@@ -1,6 +1,48 @@
-{% snapshot calendar_width %}
-{{ config(database='my_database', schema='my_schema', unique_key=[], strategy='timestamp', updated_at='checksum_updated_at') }}
+{{ config(database='my_database', schema='my_schema', materialized='incremental', incremental_strategy='merge', unique_key=[]) }}
+{% if is_incremental() %}
 SELECT
-  *
-FROM {{ ref("calendar_width") }}
-{% endsnapshot %}
+  ANY_VALUE("t8"."checksum_updated_at") AS "checksum_updated_at",
+  DATE_DIFF('DAY', MIN("t8"."day"), MAX("t8"."day")) AS "day_span"
+FROM (
+  SELECT
+    "t7"."checksum_updated_at",
+    "t7"."day"
+  FROM (
+    SELECT
+      "t6"."checksum_updated_at",
+      "t6"."checksum",
+      "t3"."day"
+    FROM (
+      SELECT
+        *
+      FROM {{ ref("calendar_checksum_bucket") }} AS "t0"
+      WHERE
+        "t0"."checksum_updated_at" > (
+          SELECT
+            MAX("t2"."checksum_updated_at") AS "Max(checksum_updated_at)"
+          FROM {{ this }} AS "t2"
+        )
+    ) AS "t6"
+    LEFT OUTER JOIN {{ source("my_schema", "calendar") }} AS "t3"
+      ON TRUE
+  ) AS "t7"
+) AS "t8"
+{% else %}
+SELECT
+  ANY_VALUE("t5"."checksum_updated_at") AS "checksum_updated_at",
+  DATE_DIFF('DAY', MIN("t5"."day"), MAX("t5"."day")) AS "day_span"
+FROM (
+  SELECT
+    "t4"."checksum_updated_at",
+    "t4"."day"
+  FROM (
+    SELECT
+      "t2"."checksum_updated_at",
+      "t2"."checksum",
+      "t3"."day"
+    FROM {{ ref("calendar_checksum_bucket") }} AS "t2"
+    LEFT OUTER JOIN {{ source("my_schema", "calendar") }} AS "t3"
+      ON TRUE
+  ) AS "t4"
+) AS "t5"
+{% endif %}

ibis_typing/samples/generated/__ibis_sql/calendar_width_snapshot.sql

@@ -0,0 +1,6 @@
+{% snapshot calendar_width_snapshot %}
+{{ config(database='my_database', schema='my_schema', unique_key=[], strategy='timestamp', updated_at='checksum_updated_at') }}
+SELECT
+  *
+FROM {{ ref("calendar_width") }}
+{% endsnapshot %}

ibis_typing/samples/generated/sample_schemas/__init__.py

@@ -1,4 +1,5 @@
 # noqa
 from .calendar_checksum_bucket import *
 from .calendar_width import *
+from .calendar_width_snapshot import *
 from .circle import *

ibis_typing/samples/generated/sample_schemas/calendar_width_snapshot.py

@@ -0,0 +1,26 @@
+from attrs import frozen
+from ibis_typing import IbisSchema
+from ibis_typing.ibis_types import *
+
+__all__ = [
+    "CalendarWidthSnapshot",
+]
+
+
+@frozen
+class CalendarWidthSnapshot(IbisSchema):
+    checksum_updated_at: Timestamp = None
+    day_span: Int64 = None
+    dbt_scd_id: String = None
+    dbt_updated_at: Timestamp = None
+    dbt_valid_from: Timestamp = None
+    dbt_valid_to: Timestamp = None
+
+    table_schema = {
+        "checksum_updated_at": "timestamp(6)",
+        "day_span": "int64",
+        "dbt_scd_id": "string",
+        "dbt_updated_at": "timestamp(6)",
+        "dbt_valid_from": "timestamp(6)",
+        "dbt_valid_to": "timestamp(6)",
+    }