Integrations

Kanoniv integrates with the tools your data team already uses. The Python SDK provides source adapters for reading data and standard output formats for writing results to any destination.

Data Sources

CSV Files

Read customer data directly from CSV files — no database or warehouse required.

from kanoniv import Source

source = Source.from_csv("crm", "contacts.csv")

Best for: local development, prototyping, small datasets, CI/CD testing.

Pandas DataFrames

Pass any DataFrame to Kanoniv. Use this when your data comes from a custom pipeline, API response, or in-memory transformation.

import pandas as pd
from kanoniv import Source

df = pd.read_csv("contacts.csv")
# Or: df = pd.read_sql("SELECT * FROM customers", conn)
# Or: df = pd.read_parquet("customers.parquet")

source = Source.from_pandas("crm", df)

Best for: existing Python pipelines, data transformations, notebook workflows.

SQL Databases (via SQLAlchemy)

Connect to any SQL database — PostgreSQL, Snowflake, BigQuery, Redshift, MySQL — using Source.from_warehouse(). It uses SQLAlchemy under the hood, so any database with a SQLAlchemy dialect works.

from kanoniv import Source

# PostgreSQL
source = Source.from_warehouse(
    "crm", "public.customers",
    connection_string="postgresql://user:pass@host:5432/db",
)

Best for: production databases, data warehouses, direct warehouse queries.

Snowflake

Install the Snowflake SQLAlchemy dialect (pip install snowflake-sqlalchemy), then use from_warehouse():

from kanoniv import Source

source = Source.from_warehouse(
    "crm", "ANALYTICS.PUBLIC.CUSTOMERS",
    connection_string="snowflake://user:pass@account/ANALYTICS",
)

BigQuery

Install the BigQuery SQLAlchemy dialect (pip install sqlalchemy-bigquery):

from kanoniv import Source

source = Source.from_warehouse(
    "crm", "project.dataset.customers",
    connection_string="bigquery://project-id/dataset",
)

dbt Models

Read data from dbt models. Kanoniv resolves the model reference through the manifest to the underlying warehouse table.

from kanoniv import Source

source = Source.from_dbt(
    "crm", model="stg_customers",
    connection_string="snowflake://user:pass@account/ANALYTICS",
)

Resolves the model name via manifest.json to database.schema.alias, then reads from the warehouse using SQLAlchemy.

Best for: teams already using dbt for data transformation who want identity resolution as a downstream step.

Orchestration

Airflow

Run Kanoniv reconciliation as an Airflow task:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def run_reconciliation():
    from kanoniv import Spec, Source, reconcile, validate

    spec = Spec.from_file("/opt/airflow/dags/specs/customer-spec.yaml")
    validate(spec).raise_on_error()

    sources = [
        Source.from_csv("crm", "/data/crm_export.csv"),
        Source.from_csv("billing", "/data/stripe_export.csv"),
    ]
    result = reconcile(sources, spec)

    # Write golden records
    result.to_dataframe().to_csv("/data/golden_customers.csv", index=False)
    print(f"Golden records: {len(result.golden_records)}")

dag = DAG(
    "customer_reconciliation",
    schedule_interval="@daily",
    start_date=datetime(2026, 1, 1),
)

reconcile_task = PythonOperator(
    task_id="reconcile",
    python_callable=run_reconciliation,
    dag=dag,
)

Prefect

from prefect import flow, task
from kanoniv import Spec, Source, reconcile, validate

@task
def load_spec():
    spec = Spec.from_file("customer-spec.yaml")
    validate(spec).raise_on_error()
    return spec

@task
def run_reconciliation(spec):
    sources = [
        Source.from_csv("crm", "contacts.csv"),
        Source.from_csv("billing", "stripe.csv"),
    ]
    return reconcile(sources, spec)

@flow
def reconciliation_flow():
    spec = load_spec()
    result = run_reconciliation(spec)
    print(f"Golden records: {len(result.golden_records)}")

reconciliation_flow()

Dagster

from dagster import op, job, Out
from kanoniv import Spec, Source, reconcile, validate

@op
def load_and_reconcile():
    spec = Spec.from_file("customer-spec.yaml")
    validate(spec).raise_on_error()

    sources = [
        Source.from_csv("crm", "contacts.csv"),
        Source.from_csv("billing", "stripe.csv"),
    ]
    result = reconcile(sources, spec)
    return result.to_dataframe()

@job
def reconciliation_job():
    load_and_reconcile()

Output Destinations

Kanoniv outputs golden records as Pandas DataFrames, which you can write to any destination:

CSV

result.to_dataframe().to_csv("golden_customers.csv", index=False)

Parquet

result.to_dataframe().to_parquet("golden_customers.parquet")

PostgreSQL / MySQL

import sqlalchemy

engine = sqlalchemy.create_engine("postgresql://user:pass@host/db")
result.to_dataframe().to_sql(
    "golden_customers",
    engine,
    if_exists="replace",
    index=False
)

Snowflake

from sqlalchemy import create_engine

engine = create_engine("snowflake://user:pass@account/ANALYTICS/PUBLIC?warehouse=COMPUTE_WH")

result.to_dataframe().to_sql(
    "golden_customers",
    engine,
    if_exists="replace",
    index=False
)

BigQuery

result.to_dataframe().to_gbq(
    "project.dataset.golden_customers",
    if_exists="replace"
)

S3

import boto3
import io

csv_buffer = io.StringIO()
result.to_dataframe().to_csv(csv_buffer, index=False)

s3 = boto3.client("s3")
s3.put_object(
    Bucket="my-bucket",
    Key="golden_customers.csv",
    Body=csv_buffer.getvalue()
)

Notebooks

Kanoniv works in Jupyter notebooks, Google Colab, and VS Code notebooks:

# In a notebook cell
from kanoniv import Spec, Source, reconcile, validate

spec = Spec.from_file("customer-spec.yaml")
validate(spec).raise_on_error()

sources = [
    Source.from_csv("crm", "contacts.csv"),
    Source.from_csv("billing", "stripe.csv"),
]
result = reconcile(sources, spec)

# Display golden records as a table
result.to_dataframe().head(20)

CI/CD

Run spec validation and reconciliation in CI/CD pipelines to catch breaking changes before deployment:

GitHub Actions

# .github/workflows/validate-spec.yml
name: Validate Spec
on: [push, pull_request]
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install kanoniv
      - run: |
          python -c "
          from kanoniv import Spec, validate
          spec = Spec.from_file('customer-spec.yaml')
          result = validate(spec)
          result.raise_on_error()
          print('Spec validation passed')
          "

Integration Matrix

Integration	Source Adapter	Output	Orchestration
CSV	Source.from_csv()	to_csv()	—
Pandas	Source.from_pandas()	to_dataframe()	—
PostgreSQL	Source.from_warehouse()	to_sql()	—
Snowflake	Source.from_warehouse()	to_sql()	—
BigQuery	Source.from_warehouse()	to_gbq()	—
dbt	Source.from_dbt()	—	dbt Core
Airflow	Via Python adapter	Via Python output	PythonOperator
Prefect	Via Python adapter	Via Python output	@task / @flow
Dagster	Via Python adapter	Via Python output	@op / @job
GitHub Actions	Via CLI/Python	—	Workflow step
Jupyter	Any adapter	to_dataframe()	—