kanoniv 0.3.3

Identity resolution as code

kanoniv

Identity resolution as code. Define matching rules in YAML, reconcile locally in Python.

Installation

pip install kanoniv

Quick Start

import kanoniv

# 1. Load your spec
spec = kanoniv.Spec.from_file("kanoniv.yml")

# 2. Validate it
result = kanoniv.validate(spec)
result.raise_on_error()

# 3. Load sources
sources = [
    kanoniv.Source.from_csv("crm", "data/crm_contacts.csv", primary_key="id"),
    kanoniv.Source.from_csv("billing", "data/billing_accounts.csv", primary_key="id"),
]

# 4. Reconcile
result = kanoniv.reconcile(sources, spec)

# 5. Golden records as a DataFrame
df = result.to_pandas()
print(f"{result.cluster_count} entities, {result.merge_rate:.0%} merge rate")

Every record in the output DataFrame gets a kanoniv_id — a stable identifier that groups duplicate records across sources into a single entity.

What the Spec Covers

The YAML spec is the single source of truth for your identity resolution pipeline:

• Sources — canonical field mappings from each system
• Blocking — composite keys to reduce O(n²) comparisons
• Scoring — Fellegi-Sunter probabilistic matching with EM training
• Normalizers — email, phone, name, nickname, domain (built-in)
• Survivorship — golden record assembly rules (source priority, most complete)
• Governance — freshness checks, schema validation, shadow-mode deploys

See the spec reference for the full schema.

Source Adapters

# Pandas DataFrame
source = kanoniv.Source.from_pandas("crm", df, primary_key="contact_id")

# CSV file
source = kanoniv.Source.from_csv("billing", "data/billing.csv", primary_key="account_id")

# Warehouse table (requires sqlalchemy)
source = kanoniv.Source.from_warehouse(
    "erp", table="raw.erp_customers", connection_string="postgresql://..."
)

# dbt model (requires sqlalchemy)
source = kanoniv.Source.from_dbt("staging", model="stg_customers")

Validation & Planning

# Validate spec for errors
result = kanoniv.validate(spec)
if not result:
    print(result.errors)

# Preview the execution plan
plan = kanoniv.plan(spec)
print(plan.summary())

Diffing Specs

# Compare two spec versions
diff = kanoniv.diff(spec_v1, spec_v2)
print(diff.summary)

Cloud API (Optional)

For managed reconciliation, monitoring, and collaboration, install with the cloud extra:

pip install kanoniv[cloud]

client = kanoniv.Client(api_key="kn_...")

result = client.resolve(system="crm", external_id="003xxx")
entities = client.entities.search(q="[e-mail]")

See the cloud API docs for the full reference.

Links

• Documentation
• Spec Reference
• GitHub
• Examples

License

Apache-2.0