Build lineage-aware canonical models with agents

import dlt

dataset = dlt.attach(pipeline_name="hubspot").dataset()
columns = dataset["contacts"].columns

dataset = dlt.attach(pipeline_name="gtm_pipeline").dataset()

contacts = dataset["contacts"].to_ibis().select("email")
new_leads = dataset["leads"].to_ibis().select("email")

overlap = contacts.join(new_leads, "email").distinct().count().execute()
print(f"{overlap} people appear in both sources — stitched by email, HubSpot wins on conflict")

dataset = dlt.attach(pipeline_name="gtm_pipeline").dataset()

stage_history = dataset["hubspot__deal_stage_history"].to_ibis().select("deal_id", "stage")
grain = stage_history.group_by(["deal_id", "stage"]).agg(n=stage_history.count())
violations = grain.filter(grain.n > 1).count().execute()

print(f"grain violated — {violations} deals logged the same stage twice" if violations else "grain confirmed")

@dlt.hub.transformation(write_disposition="replace")
def customer_orders(dataset: dlt.Dataset) -> typing.Iterator[ir.Table]:
    orders = dataset.table("orders").to_ibis()
    yield orders.group_by("customer_id").aggregate(
        first_order=orders.ordered_at.min(),
        most_recent_order=orders.ordered_at.max(),
        number_of_orders=orders.id.count(),
    )

@dlt.source
def customers_metrics(raw_dataset: dlt.Dataset) -> list:
    return [customer_orders(raw_dataset)]

@dlt.hub.transformation(write_disposition="append")
def orders_from_new_loads(
    dataset: dlt.Dataset,
    loaded_at: dlt.sources.incremental[pendulum.DateTime] = dlt.sources.incremental(
        "_dlt_loads.inserted_at", range_start="open"
    ),
):
    # dlt auto-joins `_dlt_loads` — no manual JOIN needed
    yield dataset.table("orders").incremental(loaded_at).select("id", "customer_id", "amount")

import dlt
from dlt.hub import data_quality as dq

pipeline = dlt.attach(pipeline_name="person_interactions_to_cdm")
dataset = pipeline.dataset()  # same dataset object the transformation just wrote to

checks = {
    "fact_event_attendance": [
        dq.checks.is_unique("attendance_sk"),
        dq.checks.is_not_null("person_sk"),
        dq.checks.is_in("status", ["registered", "attended", "no_show"]),
    ]
}
dq.CheckSuite(dataset, checks=checks).checks