feat(dbt): Add DBT integration framework

ibis_typing/dbt/README.md

@@ -0,0 +1,139 @@
+# `ibis_typing.dbt` — dbt Integration
+
+Bridges typed `ibis_typing` [Expression](../expression.py) classes with
+[DBT](https://docs.getdbt.com/) by compiling them to DBT SQL (Jinja templates).
+
+---
+
+## Overview
+
+| Class / function       | Purpose                                                      |
+|------------------------|--------------------------------------------------------------|
+| `DbtModel`             | Wraps an `Expression` as a DBT model with its `ModelConfig`  |
+| `DbtSource`            | Wraps an `IbisDbSchema` as a DBT source table                |
+| `DbtSnapshot`          | Wraps an `IbisSchema` as a DBT snapshot                      |
+| `ModelConfig`          | DBT model config (materialization, incremental strategy, …)  |
+| `SnapshotConfig`       | DBT snapshot config (strategy, unique key, updated-at, …)    |
+| `DbtModelResolver`     | Auto-discovers `Expression` implementations inside a package |
+| `dbt_model_to_dbt_sql` | Compiles a `DbtModel` / `DbtSnapshot` to a DBT SQL string    |
+
+---
+
+## Quick Start
+
+The canonical example lives in
+[`ibis_typing/samples/dbt_models.py`](../samples/dbt_models.py).
+It shows how to build model and source lookups that can be fed directly into the
+SQL compiler.
+
+```python
+# ibis_typing/samples/dbt_models.py
+from collections.abc import Mapping
+
+from ibis_typing import Expression, IbisDbSchema, IbisSchema, naming, samples
+from ibis_typing.dbt import (
+    DbtModel,
+    DbtSnapshot,
+    DbtSource,
+    ModelConfig,
+    SnapshotConfig,
+    dbt_model_resolver,
+)
+from ibis_typing.dbt.dbt_model import IncrementalStrategy, Materialized
+from ibis_typing.dbt.dbt_snapshot import 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
+
+my_namespace = ("my_database", "my_schema")
+
+
+def get_dbt_model_lookup() -> Mapping[type[Expression], DbtModel]:
+    resolver = dbt_model_resolver.DbtModelResolver(samples)
+
+    models = {
+        # Provide timestamp as a table so every dbt run uses a fixed timestamp.
+        DbtModel(
+            TimestampNow,
+            config=ModelConfig(
+                materialized=Materialized.table,
+            ).with_namespace(*my_namespace),
+        )
+    }
+    incremental_models = [
+        DbtModel(
+            expr,
+            config=ModelConfig(
+                materialized=Materialized.incremental,
+                incremental_strategy=IncrementalStrategy.merge,
+                unique_key=expr.incremental_params.group_by,
+            ).with_namespace(*my_namespace),
+        )
+        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),
+        )
+    ]
+    all_models = (*models, *incremental_models, *snapshots)
+    return {model.expr: model for model in all_models}
+
+
+def get_dbt_source_lookup() -> Mapping[type[IbisSchema], DbtSource]:
+    sources = [Calendar, CircleParameters]
+    return {schema: DbtSource(as_db_schema(schema)) for schema in sources}
+
+
+def as_db_schema(schema: type[IbisSchema]) -> type[IbisDbSchema]:
+    kwargs = {
+        "table_name": naming.snake_case(schema.__name__),
+        "table_namespace": my_namespace,
+    }
+    return type(schema.__name__, (IbisDbSchema, schema), kwargs)
+```
+
+---
+
+## Compiling to dbt SQL
+
+Use `DbtRefTableProvider` to wire the model and source lookups together, then
+call `dbt_model_to_dbt_sql` for each model.
+See [`ibis_typing/samples/tests/test_generate_ibis_dbt_sql.py`](../samples/tests/test_generate_ibis_dbt_sql.py)
+for the full test that also snapshots the generated SQL to disk.
+
+```python
+from ibis_typing.dbt import dbt_sql_compiler
+from ibis_typing.dbt.dbt_ibis_constructor import DbtRefTableProvider
+from ibis_typing.samples import dbt_models
+
+model_lookup = dbt_models.get_dbt_model_lookup()
+source_lookup = dbt_models.get_dbt_source_lookup()
+ref_provider = DbtRefTableProvider(model_lookup, source_lookup)
+
+for model in model_lookup.values():
+    sql = dbt_sql_compiler.dbt_model_to_dbt_sql(
+        model,
+        dialect="duckdb",
+        ref_provider=ref_provider,
+    )
+    print(sql)
+```
+
+The compiler automatically:
+
+- Injects a `{{ config(...) }}` Jinja header from the model's config.
+- Wraps incremental models in `{% if is_incremental() %} … {% else %} … {% endif %}`.
+- Wraps snapshots in `{% snapshot <name> %} … {% endsnapshot %}`.
+- Replaces internal placeholder tokens with `{{ ref("…") }}`, `{{ source("…", "…") }}`,
+  and `{{ this }}`.
+
+## Auto-Discovery with `DbtModelResolver`
+
+`DbtModelResolver` scans a Python package and returns all `Expression`
+subclasses grouped by category:

ibis_typing/dbt/__init__.py

@@ -0,0 +1,14 @@
+from .dbt_model import DbtModel, DbtSource, ModelConfig
+from .dbt_model_resolver import DbtModelResolver
+from .dbt_snapshot import DbtSnapshot, SnapshotConfig
+from .dbt_sql_compiler import dbt_model_to_dbt_sql
+
+__all__ = [
+    "DbtModel",
+    "DbtModelResolver",
+    "DbtSnapshot",
+    "DbtSource",
+    "ModelConfig",
+    "SnapshotConfig",
+    "dbt_model_to_dbt_sql",
+]

ibis_typing/dbt/dbt_ibis_constructor.py

@@ -0,0 +1,117 @@
+"""Construct ibis Tables for compiling DBT SQL models."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping
+
+import ibis
+from attrs import frozen
+
+from ibis_typing import Expression, IbisSchema, IbisTable, evaluator
+from ibis_typing.checksum_buckets import (
+    AsIncrementedBuckets,
+    ChecksumBuckets,
+    IncrementalExpression,
+)
+from ibis_typing.table_provider import TableProvider
+
+from .dbt_model import DbtModel, DbtSource
+
+logger = logging.getLogger(__name__)
+
+
+def construct_dbt_model[E: Expression](
+    schema: type[E],
+    ref_provider: DbtRefTableProvider,
+    *,
+    self: type[Expression] | None = None,
+    buckets_update: bool = False,
+) -> IbisTable[E]:
+    logger.info(
+        f"Constructing {schema.__name__}"
+        if not self
+        else f"Constructing {schema.__name__} for {self.__name__}"
+    )
+    self = self or schema
+    providers = [
+        DbtSelfRefProvider(self),
+        *([DbtBucketProvider(schema, ref_provider)] if buckets_update else ()),
+        ref_provider,
+    ]
+    return evaluator.from_expression(schema, table_providers=providers)
+
+
+def construct_dbt_bucket_increment[M: IncrementalExpression](
+    expr: type[M],
+    ref_provider: DbtRefTableProvider,
+) -> IbisTable[M]:
+    logger.info(f"Constructing bucket increment for {expr.__name__}")
+    return evaluator.construct_increment(
+        expr,
+        prior_providers=[DbtSelfRefProvider(expr)],
+        input_providers=[ref_provider],
+    )
+
+
+@frozen
+class DbtBucketProvider(TableProvider):
+    expr: type[Expression]
+    table_provider: DbtRefTableProvider
+
+    def __call__(self, schema):
+        if not issubclass(schema, ChecksumBuckets):
+            return None
+
+        assert issubclass(self.expr, IncrementalExpression)
+        return construct_dbt_model(
+            schema @ AsIncrementedBuckets(self.expr),
+            self.table_provider,
+            self=self.expr,
+        )
+
+
+@frozen
+class DbtSelfRefProvider(TableProvider):
+    self_expr: type[Expression]
+
+    SELF_TOKEN = "__dbt_this__"
+
+    def __call__(self, schema):
+        if not issubclass(schema, self.self_expr):
+            return None
+
+        table = ibis.table(schema.table_schema, name=self.SELF_TOKEN)
+        return schema.of(table)
+
+
+@frozen
+class DbtRefTableProvider(TableProvider):
+    """Provide tables with DBT token names for compiling DBT SQL."""
+
+    model_lookup: Mapping[type[Expression], DbtModel]
+    source_lookup: Mapping[type[IbisSchema], DbtSource]
+
+    REF_TOKEN = "__dbt_ref__"
+    SRC_TOKEN = "__dbt_src__"
+    SEP_TOKEN = "__dbt_sep__"
+
+    def __call__(self, schema):
+        if model := self.model_lookup.get(schema):
+            schema = model.db_schema
+            name = schema.table_name
+
+            ref_name = f"{self.REF_TOKEN}{name}{self.REF_TOKEN}"
+            table = ibis.table(schema.table_schema, name=ref_name)
+            return schema.of(table)
+
+        if source := self.source_lookup.get(schema):
+            schema = source.db_schema
+            name = schema.table_name
+            src = source.dbt_source_name
+
+            ref_name = f"{self.SRC_TOKEN}{name}{self.SEP_TOKEN}{src}{self.SRC_TOKEN}"
+            table = ibis.table(schema.table_schema, name=ref_name)
+            return schema.of(table)
+
+        return None