Source Adapters

How Adapters Work

Kanoniv doesn't move your data. Your data stays where it already lives - in Snowflake, PostgreSQL, CSV files, pandas DataFrames, Polars DataFrames, DuckDB tables, dbt models. Adapters are how you tell Kanoniv to read from those sources for reconciliation.

There are two parts to every source:

1. Spec (YAML) - Declare what the source is and where it lives. This is metadata - the engine never connects to your database directly.

sources:
  crm:
    adapter: postgres
    location: public.contacts
    primary_key: contact_id
    schema:
      email: { type: string, pii: true }
      name: { type: string }

2. Python (SDK) - Load the actual data. Connection strings, credentials, and data access happen here - never in the YAML.

crm = Source.from_warehouse(
    "crm", "public.contacts",
    connection_string="postgresql://user:pass@host/db",
    primary_key="contact_id",
)

Then pass both to reconcile():

from kanoniv import Spec, Source, reconcile

spec = Spec.from_file("customer-spec.yaml")
sources = [crm]
result = reconcile(sources, spec)

Why This Design?

• Your data stays put. Kanoniv reads from your existing infrastructure. No ETL, no data movement, no storage liability.
• Credentials stay in code. Connection strings never touch your YAML spec, so specs can be committed to Git safely.
• Mix any source. CSV files, warehouse tables, Pandas/Polars DataFrames, Arrow tables, DuckDB queries, and dbt models in the same reconciliation.
• Test locally, run in production. Use Source.from_csv() during development, swap to Source.from_warehouse() for production - same spec, same rules.

---

Available Adapters

Adapter	SDK Method	Requires	Use Case
CSV	Source.from_csv()	nothing	Local files, prototyping
JSON	Source.from_json()	nothing	JSON array files
Pandas	Source.from_pandas()	pandas	In-memory DataFrames
Polars	Source.from_polars()	polars	In-memory Polars DataFrames
PyArrow	Source.from_arrow()	pyarrow	Arrow Tables from any source
DuckDB	Source.from_duckdb()	duckdb	SQL queries on DuckDB tables
Warehouse	Source.from_warehouse()	sqlalchemy + dialect	PostgreSQL, Snowflake, BigQuery, Redshift, MySQL, Databricks
dbt	Source.from_dbt()	sqlalchemy + dialect	dbt model outputs

One method for all warehouses

Source.from_warehouse() uses SQLAlchemy under the hood. You don't need a different adapter for each database - just install the right dialect package and pass the connection string.

---

CSV

# Spec
sources:
  crm:
    adapter: csv
    location: data/contacts.csv
    primary_key: id

# Python
source = Source.from_csv("crm", "data/contacts.csv", primary_key="id")

Reads with stdlib csv.DictReader. UTF-8, comma-delimited. All values are stringified.

JSON

# Spec
sources:
  events:
    adapter: json
    location: data/events.json
    primary_key: event_id

# Python
source = Source.from_json("events", "data/events.json", primary_key="event_id")

File must contain a top-level JSON array of objects. null values become empty strings.

Pandas

# Spec
sources:
  analytics:
    adapter: pandas
    location: dataframe://df_customers
    primary_key: user_id

# Python
source = Source.from_pandas("analytics", df, primary_key="user_id")

Pass any pandas DataFrame. NaN and None values become empty strings.

Polars

# Python
import polars as pl

df = pl.read_csv("data/contacts.csv")
source = Source.from_polars("crm", df, primary_key="id")

Pass any Polars DataFrame. null values become empty strings. Types are inferred from Polars dtypes - Int*/UInt*/Float* map to "number", Boolean to "boolean", Date/Datetime to "date", everything else to "string".

No extra dependencies beyond polars itself - if you already have it installed, the adapter just works.

PyArrow

# Python
import pyarrow.parquet as pq

table = pq.read_table("data/contacts.parquet")
source = Source.from_arrow("crm", table, primary_key="id")

Pass any pyarrow.Table. Useful for reading Parquet files, Arrow IPC, or any library that produces Arrow tables. null values become empty strings.

No extra dependencies beyond pyarrow itself.

DuckDB

# Python
import duckdb

con = duckdb.connect("analytics.duckdb")
source = Source.from_duckdb("crm", con, "SELECT * FROM contacts", primary_key="id")

Pass a DuckDB connection and a SQL query (or bare table name). Results are streamed in chunks of 1,000 rows.

Signature: Source.from_duckdb(name, connection, query, *, primary_key=None)

• query can be any valid SQL or a bare table name (auto-wrapped in SELECT * FROM)
• row_count() returns None to avoid expensive COUNT(*) scans
• Connection is not copied - the adapter uses your existing connection directly

# Bare table name works too
source = Source.from_duckdb("crm", con, "contacts", primary_key="id")

# Complex queries work
source = Source.from_duckdb("recent", con,
    "SELECT * FROM contacts WHERE updated_at > '2024-01-01'",
    primary_key="id",
)

No extra dependencies beyond duckdb itself.

Warehouse

One method handles all SQL databases via SQLAlchemy. Install the dialect for your database:

Database	Dialect Package	Connection String
PostgreSQL	psycopg2 (usually pre-installed)	postgresql://user:pass@host:5432/db
Snowflake	snowflake-sqlalchemy	snowflake://user:pass@account/db
BigQuery	sqlalchemy-bigquery	bigquery://project-id/dataset
Redshift	sqlalchemy-redshift	redshift://user:pass@cluster/db
MySQL	pymysql	mysql+pymysql://user:pass@host/db
Databricks	databricks-sql-connector (includes SQLAlchemy dialect)	databricks://token:dapi...@host?http_path=/sql/...

# Spec
sources:
  customers:
    adapter: snowflake
    location: ANALYTICS.CORE.CUSTOMERS
    primary_key: customer_id
    schema:
      email: { type: string, pii: true }
      phone: { type: string, pii: true }
      first_name: { type: string }
      last_name: { type: string }

# Python
source = Source.from_warehouse(
    "customers", "ANALYTICS.CORE.CUSTOMERS",
    connection_string="snowflake://user:pass@account/ANALYTICS",
    primary_key="customer_id",
)

Signature: Source.from_warehouse(name, table, connection_string, *, primary_key=None, batch_size=5000)

• table supports schema.table and database.schema.table dot notation
• Rows are streamed in batches of 5,000 by default (configurable via batch_size)
• Connection is lazy - not opened until data is actually read

Databricks Example

# Spec
sources:
  customers:
    adapter: databricks
    location: main.default.customers
    primary_key: customer_id

# Python
source = Source.from_warehouse(
    "customers", "main.default.customers",
    connection_string="databricks://token:[email protected]?http_path=/sql/1.0/warehouses/abc",
    primary_key="customer_id",
)

The databricks-sql-connector package includes a SQLAlchemy dialect, so Source.from_warehouse() works out of the box. Install with pip install databricks-sql-connector.

Arrow Fast Path (Cloud)

When using kanoniv.cloud.reconcile() with Databricks sources, install the databricks extra for the Arrow fast path:

pip install kanoniv[cloud,databricks]

This uses the native Databricks connector with fetchall_arrow() for columnar transfer, bypassing SQLAlchemy entirely. The staging layer auto-detects databricks:// URLs and routes through the fast path.

dbt

Load data from a dbt model by resolving its table location through the dbt manifest.

# Spec
sources:
  staging:
    adapter: dbt
    location: ref('stg_customers')
    primary_key: customer_id

# Python
source = Source.from_dbt(
    "staging", model="stg_customers",
    connection_string="postgresql://user:pass@host/db",
    primary_key="customer_id",
)

Signature: Source.from_dbt(name, model, *, connection_string=None, manifest_path="target/manifest.json", primary_key=None)

• Resolves the model name through manifest.json to database.schema.alias
• Accepts both "stg_customers" and ref('stg_customers') syntax
• Schema is inferred from manifest column metadata when available, falling back to warehouse reflection

---

Mixing Adapters

Adapters can be freely combined in the same reconciliation. The engine normalizes everything before matching:

from kanoniv import Source, Spec, reconcile

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

sources = [
    Source.from_csv("crm", "crm_export.csv", primary_key="id"),
    Source.from_polars("analytics", polars_df, primary_key="user_id"),
    Source.from_arrow("events", arrow_table, primary_key="event_id"),
    Source.from_duckdb("billing", con, "stripe_customers", primary_key="customer_id"),
    Source.from_warehouse("warehouse", "public.accounts",
                          connection_string="postgresql://...",
                          primary_key="account_id"),
]

result = reconcile(sources, spec)
print(f"Golden records: {len(result.golden_records)}")
print(f"Merge rate: {result.merge_rate:.1%}")

This is one of Kanoniv's strengths - you can reconcile a CSV export from your CRM against a Polars DataFrame, an Arrow table from Parquet, a DuckDB query, and live Snowflake tables, all in one call.

---

Primary Key

The primary_key parameter tells the adapter which column to use as the entity's external ID. If omitted, each row gets a random UUID.

Setting it is strongly recommended - it makes decisions and golden records traceable back to source rows. Without it, you can't tell which original record contributed to a golden record.

# Good - traceable back to CRM contact_id
Source.from_csv("crm", "contacts.csv", primary_key="contact_id")

# Works but not recommended - rows get random UUIDs
Source.from_csv("crm", "contacts.csv")

---

Schema Introspection

Every source exposes a schema() method for inspecting data before reconciliation:

source = Source.from_csv("crm", "crm.csv", primary_key="id")
schema = source.schema()

for col in schema.columns:
    print(f"{col.name}: {col.dtype} (nullable={col.nullable})")

SourceSchema field	Type	Description
columns	list[ColumnSchema]	Column definitions
primary_key	str | None	Primary key column name
row_count	int | None	Row count (when available without a full scan)

ColumnSchema field	Type	Description
name	str	Column name
dtype	str	Inferred type: "string", "number", "boolean", or "date"
nullable	bool	Whether null/empty values were observed
sample_values	list[str]	Up to 5 sample values

Types are inferred by sampling up to 100 rows (CSV, JSON), from native dtypes (Pandas, Polars, PyArrow), from query metadata (DuckDB), or from database column types (Warehouse, dbt).

---

Next Steps

• Spec Reference: Sources - Full YAML source configuration (attributes, freshness, temporal, schema, tags, sampling)
• Quickstart - Write a spec and reconcile in 5 minutes
• First Reconciliation - Full walkthrough with sample data