Python SDK (Core)

Validate, plan, diff, and reconcile Kanoniv identity specs from Python. Built on the same Rust engine as the CLI — native speed, zero reimplementation.

Installation

bash

pip install kanoniv

Requirements: Python 3.9+

The package ships prebuilt wheels for Linux (x86_64) and macOS (x86_64, ARM). If no wheel is available for your platform, the build requires Rust 1.70+.

Some adapters use libraries you may already have installed:

Source.from_pandas() / result.to_pandas() need pandas
Source.from_polars() needs polars
Source.from_arrow() needs pyarrow
Source.from_duckdb() needs duckdb
Source.from_warehouse() / Source.from_dbt() need sqlalchemy (Databricks: databricks-sql-connector includes the dialect)

Quick Start

python

from kanoniv import Spec, Source, validate, reconcile

# Load and validate a spec
spec = Spec.from_file("customer-spec.yaml")
validate(spec).raise_on_error()

# Load data
sources = [
    Source.from_csv("crm", "crm.csv"),
    Source.from_csv("billing", "stripe.csv"),
]

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

Discover + Bootstrap Pipeline (Cloud)

Cloud-only in v0.3.0+

discover_entity_sources, bootstrap_spec, autotune, and FeedbackLabel were removed from the local SDK in v0.3.0. These capabilities are now available through the Cloud API and CLI:

bash

pip install kanoniv[cloud]
kanoniv login

# Ingest data, auto-discover schema, and bootstrap a spec
kanoniv ingest ./data/ --entity-type person
kanoniv autodetect --bootstrap --entity-type person

# Optimize thresholds
kanoniv autotune --entity-type person

# Reconcile
kanoniv reconcile --wait --entity-type person

See the CLI Reference for full details.

The fastest way to get started without writing YAML. The Cloud API profiles your data, detects identity-relevant columns (email, phone, name, domain, etc.), infers the entity type, and generates a complete identity spec with matching rules, blocking keys, and thresholds - all calibrated to your data.

The kanoniv autodetect --bootstrap command handles both discovery and spec generation in a single step. The generated spec can then be downloaded, inspected, and version-controlled as a YAML file.

Core API

Spec

Load and parse identity specification files.

python

from kanoniv import Spec

# From a file
spec = Spec.from_file("identity-spec.yaml")

# From a YAML string
spec = Spec.from_string("""
api_version: kanoniv/v1
identity_version: "1.0"
entity:
  name: customer
sources:
  crm:
    adapter: csv
    location: contacts.csv
    primary_key: crm_id
    attributes:
      email: email_address
      phone: phone_number
rules:
  - name: email_exact
    type: exact
    field: email
    weight: 1.0
decision:
  thresholds:
    match: 0.9
    review: 0.7
""")

Properties:

Property	Type	Description
`spec.entity`	`str`	Entity name from spec
`spec.version`	`str`	Spec version (default: `"0.0.0"`)
`spec.sources`	`list[dict]`	Source definitions
`spec.rules`	`list[dict]`	Matching rules
`spec.raw`	`str`	Original YAML string
`spec.parsed`	`dict`	Fully parsed spec as dict

validate

python

validate(spec: Spec) -> ValidationResult

Validate a spec for schema and semantic correctness.

python

from kanoniv import Spec, validate

spec = Spec.from_file("my-spec.yaml")
result = validate(spec)

print(result.valid)    # True / False
print(result.errors)   # ["Missing required field: ...", ...]

# Boolean conversion
if result:
    print("Valid!")

# Raise on error
result.raise_on_error()  # raises ValueError if invalid

Parameters:

Parameter	Type	Description
`spec`	`Spec`	The identity spec to validate

Returns: ValidationResult

Field	Type	Description
`valid`	`bool`	`True` if no errors found
`errors`	`list[str]`	Validation error messages

Method	Description
`raise_on_error()`	Raises `ValueError` if spec is invalid
`__bool__()`	Returns `self.valid`

plan

python

plan(spec: Spec) -> PlanResult

Generate an execution plan from a validated spec. The plan shows how the reconciliation engine will process data — what stages run, which rules fire, how blocking reduces comparisons, and what risks exist.

python

from kanoniv import Spec, plan

spec = Spec.from_file("my-spec.yaml")
p = plan(spec)

print(p.summary())           # Human-readable plan overview
print(p.entity)              # "customer"
print(p.plan_hash)           # "sha256:a1b2c3d4..."

# Serialize for storage or comparison
data = p.to_dict()

Returns: PlanResult

Property	Type	Description
`entity`	`str`	Entity name targeted by plan
`plan_hash`	`str`	SHA-256 hash of the execution plan (deterministic)
`execution_stages`	`list[dict]`	Ordered stages of execution
`match_strategies`	`list[dict]`	Matching strategies compiled from rules
`survivorship`	`list[dict]`	Survivorship rules summary
`blocking`	`dict`	Blocking analysis results
`risk_flags`	`list[dict]`	Static analysis risk flags

Method	Returns	Description
`summary()`	`str`	Human-readable plan summary
`to_dict()`	`dict`	Full plan as serializable dictionary

Execution Stages

Each entry in execution_stages:

Field	Type	Description
`stage`	`int`	Stage number (1-8)
`name`	`str`	Stage name
`description`	`str`	What this stage does with your spec
`inputs`	`list[str]`	Data flowing into this stage
`outputs`	`list[str]`	Data produced by this stage

The 8 stages are: Normalize sources → Generate blocking keys → Exact matches → Fuzzy matches → Score & decide → Cluster entities → Apply survivorship → Emit outputs.

Match Strategies

Each entry in match_strategies:

Field	Type	Description
`rule_name`	`str`	Name from spec
`match_type`	`str`	`"exact"`, `"similarity"`, `"range"`, `"composite"`
`field`	`str`	Field being matched
`algorithm`	`str \| None`	Algorithm for fuzzy rules (`"jaro_winkler"`, `"levenshtein"`, etc.)
`threshold`	`float \| None`	Minimum score to count as a match
`weight`	`float`	Rule weight in scoring
`evaluation_order`	`int`	Execution priority (exact=3, fuzzy=4)

Blocking Analysis

The blocking dict:

Field	Type	Description
`strategy`	`str`	Blocking strategy (`"composite"`, `"lsh"`, `"none"`)
`keys`	`list[dict]`	Blocking keys, each with `name` and `transformation`
`estimated_reduction`	`str`	`"none"`, `"low"`, `"medium"`, or `"high"`
`warnings`	`list[str]`	Warnings (e.g., "No blocking keys — O(n²) pairwise comparisons")

Survivorship Summary

Each entry in survivorship:

Field	Type	Description
`field`	`str`	Field name
`strategy`	`str`	`"most_recent"`, `"most_complete"`, `"source_priority"`, `"aggregate"`
`source_priority`	`list[str] \| None`	Source priority order (when strategy is `source_priority`)

Risk Flags

Each entry in risk_flags:

Field	Type	Description
`severity`	`str`	`"critical"`, `"high"`, `"medium"`, or `"low"`
`code`	`str`	Machine-readable risk code
`message`	`str`	Human-readable description
`recommendation`	`str`	Suggested fix

The engine checks 9 risk conditions:

Code	Severity	Trigger
`NO_BLOCKING`	critical	No blocking keys — O(n²) comparisons
`SINGLE_SIGNAL`	high	Only one match rule
`LOW_THRESHOLD`	high	Fuzzy threshold below 0.8
`HIGH_WEIGHT_FUZZY`	medium	Fuzzy rule weight above 0.9
`NO_SURVIVORSHIP`	medium	No survivorship rules defined
`PHONE_WITHOUT_BLOCKING`	high	Phone rule without phone blocking key
`NO_REVIEW_THRESHOLD`	medium	No review threshold configured
`SINGLE_SOURCE`	low	Only one data source
`MISSING_TEMPORAL`	low	No temporal configuration

diff

python

diff(spec_a: Spec, spec_b: Spec) -> DiffResult

Compare two spec versions and return a comprehensive report of what changed. Compares every spec section: rules, sources, entity, blocking, thresholds, survivorship, scoring, metadata, and version.

python

from kanoniv import Spec, diff

old = Spec.from_file("v1.yaml")
new = Spec.from_file("v2.yaml")
result = diff(old, new)

print(result.summary)              # "2 rules changed, 1 source added, thresholds changed"
print(result.has_changes)          # True

# Rule changes
print(result.rules_added)          # ["fuzzy_phone"]
print(result.rules_removed)        # ["legacy_name"]
print(result.rules_modified)       # [{"name": "email_exact", "field": "weight", ...}]

# Source changes
print(result.sources_added)        # ["support"]
print(result.sources_removed)      # []
print(result.sources_modified)     # [{"name": "crm", "field": "table", ...}]

# Section-level change flags (bool)
result.entity_changed
result.blocking_changed
result.thresholds_changed
result.survivorship_changed
result.scoring_changed
result.metadata_changed
result.version_changed

# Section-level change details (field-level diffs)
result.entity_changes          # [{"path": "entity.name", "old_value": "...", "new_value": "..."}]
result.blocking_changes        # [{"path": "blocking.strategy", ...}]
result.decision_changes        # [{"path": "decision.thresholds.match", ...}]
result.survivorship_changes    # [{"path": "survivorship.rules[0].strategy", ...}]
result.scoring_changes         # [{"path": "decision.scoring.method", ...}]
result.metadata_changes        # [{"path": "metadata.owner", ...}]

Returns: DiffResult

Property	Type	Description
`rules_added`	`list[str]`	Rule names added in spec_b
`rules_removed`	`list[str]`	Rule names removed from spec_a
`rules_modified`	`list[dict]`	Changed rules — `name`, `field`, `old_value`, `new_value`
`sources_added`	`list[str]`	Source names added in spec_b
`sources_removed`	`list[str]`	Source names removed from spec_a
`sources_modified`	`list[dict]`	Changed sources — `name`, `field`, `old_value`, `new_value`
`entity_changed`	`bool`	Whether the entity definition changed
`entity_changes`	`list[dict]`	Field-level entity diffs — `path`, `old_value`, `new_value`
`blocking_changed`	`bool`	Whether blocking strategy or keys changed
`blocking_changes`	`list[dict]`	Field-level blocking diffs
`thresholds_changed`	`bool`	Whether decision thresholds changed
`decision_changes`	`list[dict]`	Field-level decision/threshold diffs
`survivorship_changed`	`bool`	Whether survivorship rules changed
`survivorship_changes`	`list[dict]`	Field-level survivorship diffs
`scoring_changed`	`bool`	Whether scoring configuration changed
`scoring_changes`	`list[dict]`	Field-level scoring diffs
`metadata_changed`	`bool`	Whether spec metadata changed
`metadata_changes`	`list[dict]`	Field-level metadata diffs
`version_changed`	`bool`	Whether `identity_version` changed
`summary`	`str`	Human-readable diff summary
`has_changes`	`bool`	`True` if any part of the spec changed

rules_modified and sources_modified return one entry per changed field — if a rule's weight and type both changed, you get two entries with the same name but different field values. All *_changes lists use path to identify the exact key that changed (e.g., "decision.thresholds.match").

Source

Source adapters load your data for local reconciliation. The name must match the source name in your spec.

python

from kanoniv import Source

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

Available adapters: from_csv, from_json, from_pandas, from_polars, from_arrow, from_duckdb, from_warehouse, from_dbt. See Source Adapters for the full reference.

reconcile

python

reconcile(sources: list[Source], spec: Spec) -> ReconcileResult

Run local identity resolution. Validates the spec, collects entities from all sources, applies attribute mappings, runs the Rust engine (blocking -> pairwise matching -> scoring -> clustering -> survivorship), and returns structured results.

python

from kanoniv import Source, Spec, reconcile

spec = Spec.from_file("customer-spec.yaml")
sources = [
    Source.from_csv("crm", "crm.csv"),
    Source.from_csv("billing", "billing.csv"),
]

result = reconcile(sources, spec)

Active learning feedback

In v0.3.0+, feedback labels and active learning are managed through the Cloud API. Use kanoniv feedback via the CLI or client.feedback via the Cloud client. See Active Learning for the full workflow.

Parameters:

Parameter	Type	Description
`sources`	`list[Source]`	Data sources to reconcile
`spec`	`Spec`	Identity spec defining rules and thresholds
`feedback`	`list[FeedbackLabel] \| None`	Optional labeled pairs for active learning. With FS scoring, refines m/u weights. With rules-based scoring, acts as force-merge/force-split overrides.

Returns: ReconcileResult

Field	Type	Description
`clusters`	`list[list[str]]`	Groups of matched entity IDs
`golden_records`	`list[dict[str, str]]`	Merged canonical records (one per cluster)
`decisions`	`list[dict]`	Match decision trace for every evaluated pair
`telemetry`	`dict`	Engine performance metrics

Property	Type	Description
`cluster_count`	`int`	Number of clusters (`len(clusters)`)
`merge_rate`	`float`	Fraction of input entities that were merged (0.0 – 1.0)

Method	Returns	Description
`to_pandas()`	`pd.DataFrame`	Golden records as a DataFrame (requires `pandas`)

Decisions

Each entry in decisions is the full trace of one pairwise comparison:

Field	Type	Description
`entity_a_id`	`str`	First entity UUID
`entity_b_id`	`str`	Second entity UUID
`decision`	`str`	`"Merge"`, `"Review"`, or `"NoMerge"`
`confidence`	`float`	Weighted aggregate score (0.0 – 1.0)
`rule_results`	`list[dict]`	Per-rule scoring breakdown
`matched_on`	`list[str]`	Rule names that matched
`blocking_key`	`str \| None`	Which blocking key brought this pair together
`temporal_adjustment`	`float \| None`	Temporal adjustment factor applied to confidence

Each entry in rule_results:

Field	Type	Description
`rule_name`	`str`	Rule name from spec
`score`	`float`	Similarity score (0.0 – 1.0)
`weight`	`float`	Rule weight
`matched`	`bool`	Whether this rule's threshold was met
`explanation`	`str`	Human-readable explanation (e.g., `"Fuzzy match on field email (score: 0.92)"`)

Telemetry

The telemetry dict contains engine performance metrics:

Field	Type	Description
`pairs_evaluated`	`int`	Total pairwise comparisons performed
`blocking_groups`	`int`	Number of blocking groups generated
`decisions_by_type`	`dict[str, int]`	Count per decision type: `{"Merge": 42, "Review": 5, "NoMerge": 118}`
`rule_telemetry`	`list[dict]`	Per-rule aggregate metrics

Each entry in rule_telemetry:

Field	Type	Description
`rule_name`	`str`	Rule name
`evaluated`	`int`	Times this rule was evaluated
`matched`	`int`	Times this rule matched
`skipped`	`int`	Times this rule was skipped
`avg_score`	`float`	Average score across all evaluations

Engine Capabilities

The reconciliation engine supports:

Matching rules — 4 types:

Type	Description
`exact`	Field equality (with optional case-insensitive / normalize flags)
`similarity`	String similarity (Jaro-Winkler, Levenshtein, Soundex, etc.)
`range`	Numeric tolerance (absolute difference)
`composite`	Boolean AND/OR over child rules

Blocking — reduces O(n²) to manageable pair counts:

Strategy	Description
Plan-based	11 transformations: Lowercase, FirstInitial, Soundex, DoubleMetaphone, AreaCode, Last4, NormalizePhone, NormalizeEmail, NormalizeName, NormalizeDomain, Trigrams
LSH	Locality-sensitive hashing via MinHash for approximate nearest-neighbor blocking

Survivorship — 4 strategies for golden record construction:

Strategy	Description
`most_recent`	Pick value from entity with latest timestamp
`most_complete`	Pick longest (most complete) value
`source_priority`	Pick value from highest-priority source
`aggregate`	Combine values: max, min, sum, avg, union, intersection

Temporal matching — 3 strategies:

Strategy	Description
`latest_only`	Filter out historical entities
`point_in_time`	Both entities must be valid at a specific timestamp
`bi_temporal`	Temporal interval overlap check

Normalization — applied during blocking and matching:

Normalizer	Description
Email	Gmail dot-stripping, `+` alias removal, lowercase
Phone	E.164 format, digit extraction
Name	Unicode NFC, lowercase, collapse whitespace
Nickname	Name normalization + canonical nickname resolution (Bob → Robert, Liz → Elizabeth)
Domain	Strip protocol/www/path, lowercase
Generic	Trim whitespace, lowercase

Clustering - Union-Find (transitive closure) by default, or graph-based clustering for weighted edge filtering and coherence checking. Configured via decision.clustering.method in the spec.

uncertain_pairs (Cloud-only in v0.3.0+)

Removed from local SDK in v0.3.0

uncertain_pairs is now available through the Cloud API. Use client.feedback or the CLI to manage active learning feedback.

Returns the n pairs closest to the decision boundary. These are the pairs where human labeling has the most impact on model improvement. In v0.3.0+, this is accessed via the Cloud API rather than the local ReconcileResult.

See Active Learning for the full feedback workflow.

save / load

python

result.save(path: str) -> None
ReconcileResult.load(path: str) -> ReconcileResult

Persist a reconciliation result (including feedback labels) to disk and reload it later. This allows accumulating labeled feedback across multiple runs.

python

result.save("customer-reconciliation.json")

# Later...
from kanoniv import ReconcileResult
previous = ReconcileResult.load("customer-reconciliation.json")

FeedbackLabel (Cloud-only in v0.3.0+)

Removed from local SDK in v0.3.0

FeedbackLabel is no longer available as a local Python import. Feedback labels are now managed through the Cloud API:

bash

# Via CLI
kanoniv feedback add --entity-a cust_123 --entity-b cust_456 --label match

# Via Cloud client
client.feedback.create(entity_a_id="cust_123", entity_b_id="cust_456", label="match")

Represents a human judgment about whether two records are the same entity. With Fellegi-Sunter scoring, labels refine the m/u probability estimates through supervised EM. With rules-based scoring, labels act as force-merge/force-split overrides. See Active Learning for details.

Validation Rules

The validator checks schema, semantic, and safety rules.

Schema Checks

Check	Description
Required fields	`api_version`, `identity_version`, `entity` must be present
API version format	Must match `kanoniv/v<N>` pattern
Entity structure	`entity.name` is required
Rule structure	Each rule must have `name` and `type`
Weight bounds	Rule weights must be between 0.0 and 1.0
Threshold bounds	Decision thresholds must be between 0.0 and 1.0
Threshold ordering	`match` >= `review` >= `reject`
Source structure	Each source requires `adapter`/`location`/`primary_key` or legacy `system`/`table`/`id`
Warehouse adapters	Snowflake, BigQuery, Redshift, Postgres require `location`
Schema field types	Must be one of: `string`, `integer`, `float`, `boolean`, `date`, `timestamp`, `json`
Sampling rate	Must be > 0.0 and <= 1.0
Tag format	Lowercase alphanumeric + hyphens only, max 20 per source

Semantic Checks

Check	Description
At least one source	Spec must define at least one source
At least one rule	Spec must define at least one rule
Field references	Rule fields must exist in source attributes
Duplicate names	No duplicate rule names or source names
Composite rules	Child rules validated recursively
Field suggestions	Typo detection with "Did you mean...?" hints
Governance tags	Sources must have all tags declared in `governance.required_tags`

Warnings

These produce warnings (not errors) — the spec is still valid:

Check	Description
Empty attributes	Source has no attribute mappings
PII without compliance	Source declares PII fields but no `entity.compliance` config

Safety Limits

Limit	Default
Max sources	10
Max rules	50
Max blocking keys	5
Max hierarchy depth	10

Usage Examples

Notebook Validation

python

from kanoniv import Spec, validate, plan

spec = Spec.from_file("specs/customer.yaml")
result = validate(spec)

if result:
    p = plan(spec)
    print(f"Plan hash: {p.plan_hash}")
    print(p.summary())

    # Check for risks before running
    for flag in p.risk_flags:
        print(f"[{flag['severity']}] {flag['code']}: {flag['message']}")

    # Inspect blocking analysis
    blocking = p.blocking
    print(f"Blocking strategy: {blocking['strategy']}")
    print(f"Estimated reduction: {blocking['estimated_reduction']}")
else:
    print(f"Found {len(result.errors)} error(s):")
    for e in result.errors:
        print(f"  - {e}")

Full Reconciliation Pipeline

python

from kanoniv import Spec, Source, validate, plan, reconcile

# 1. Load spec
spec = Spec.from_file("specs/customer.yaml")

# 2. Validate
validation = validate(spec)
if not validation:
    print(f"Invalid: {validation.errors}")
    raise SystemExit(1)

# 3. Plan
execution_plan = plan(spec)
print(execution_plan.summary())

# 4. Load data
sources = [
    Source.from_csv("crm", "data/crm.csv"),
    Source.from_csv("billing", "data/stripe.csv"),
    Source.from_csv("support", "data/zendesk.csv"),
]

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

# 6. Inspect decisions
for d in result.decisions:
    if d["decision"] == "Merge":
        print(f"  Merge: {d['entity_a_id'][:8]} ↔ {d['entity_b_id'][:8]} "
              f"(confidence: {d['confidence']:.2f})")

# 7. Telemetry
t = result.telemetry
print(f"Pairs evaluated: {t['pairs_evaluated']}")
print(f"Decisions: {t['decisions_by_type']}")

# 8. Export
df = result.to_pandas()
df.to_csv("golden_records.csv", index=False)

CI Pipeline Gate

python

import sys
import glob
from kanoniv import Spec, validate

failed = False
for path in glob.glob("specs/**/*.yaml", recursive=True):
    spec = Spec.from_file(path)
    result = validate(spec)
    if not result:
        print(f"FAIL {path}:")
        for e in result.errors:
            print(f"  - {e}")
        failed = True
    else:
        print(f"OK   {path}")

sys.exit(1 if failed else 0)

Change Detection in CI

python

from kanoniv import Spec, plan, diff

old = Spec.from_file("specs/customer.yaml.bak")
new = Spec.from_file("specs/customer.yaml")

old_plan = plan(old)
new_plan = plan(new)

if old_plan.plan_hash != new_plan.plan_hash:
    result = diff(old, new)
    print(f"Spec changed: {result.summary}")

    if result.thresholds_changed:
        print("WARNING: Decision thresholds changed — requires review")
        for c in result.decision_changes:
            print(f"  {c['path']}: {c['old_value']} → {c['new_value']}")
    if result.blocking_changed:
        print("WARNING: Blocking strategy changed — may affect performance")
    if result.entity_changed:
        print("WARNING: Entity definition changed — downstream consumers may break")
    if result.sources_added or result.sources_removed:
        print(f"Sources added: {result.sources_added}, removed: {result.sources_removed}")
    if result.survivorship_changed:
        print("WARNING: Survivorship rules changed — golden record output may differ")
    if result.scoring_changed:
        print("WARNING: Scoring configuration changed")
    if result.version_changed:
        print("INFO: Spec version bumped")

Pytest Integration

python

import pytest
from kanoniv import Spec, validate, plan

SPEC_FILES = [
    "specs/customer.yaml",
    "specs/vendor.yaml",
    "specs/employee.yaml",
]

@pytest.mark.parametrize("spec_path", SPEC_FILES)
def test_spec_is_valid(spec_path):
    spec = Spec.from_file(spec_path)
    result = validate(spec)
    assert result.valid, f"Validation errors: {result.errors}"

@pytest.mark.parametrize("spec_path", SPEC_FILES)
def test_spec_plans(spec_path):
    spec = Spec.from_file(spec_path)
    p = plan(spec)
    assert p.plan_hash.startswith("sha256:")
    assert len(p.execution_stages) > 0

Pre-commit Hook

yaml

# .pre-commit-config.yaml
repos:
  - repo: local
    hooks:
      - id: kanoniv-validate
        name: Validate Kanoniv specs
        entry: python -c "
          import sys, glob;
          from kanoniv import Spec, validate;
          files = [f for f in sys.argv[1:] if f.endswith('.yaml')];
          bad = [(f, validate(Spec.from_file(f))) for f in files];
          bad = [(f, r) for f, r in bad if not r];
          [print(f'FAIL {f}: {r.errors}') for f, r in bad];
          sys.exit(1 if bad else 0)
        "
        language: python
        additional_dependencies: ["kanoniv"]
        types: [yaml]

Architecture

The Python SDK is a thin native extension; all logic lives in the Rust kanoniv_core crate:

kanoniv (Python package)
├── __init__.py          Re-exports: Spec, Source, validate, plan, diff, reconcile
├── spec.py              Spec class (from_file, from_string)
├── source.py            Source class (from_csv, from_pandas, from_json, ...)
├── validate.py          validate() → ValidationResult
├── plan.py              plan() → PlanResult
├── diff.py              diff() → DiffResult
├── reconcile.py         reconcile() → ReconcileResult
├── adapters/
│   ├── file.py          CsvAdapter, JsonAdapter (stdlib only)
│   ├── pandas.py        PandasAdapter (requires pandas)
│   ├── polars.py        PolarsAdapter (requires polars)
│   ├── arrow.py         ArrowAdapter (requires pyarrow)
│   ├── duckdb.py        DuckDBAdapter (requires duckdb)
│   ├── warehouse.py     WarehouseAdapter (requires sqlalchemy)
│   └── dbt.py           DbtAdapter (requires sqlalchemy)
└── _native.so           Compiled Rust extension (PyO3)
    └── kanoniv_core     Shared Rust validation + reconciliation engine

This means:

Same validation as the CLI, so there is no drift between tools
Native speed: validation and reconciliation run in compiled Rust
Minimal Python dependencies: CSV and JSON adapters use stdlib only; pandas and sqlalchemy are optional
Local execution: no API calls required for validate, plan, diff, or reconcile

Python SDK (Core) ​

Installation ​

Quick Start ​

Discover + Bootstrap Pipeline (Cloud) ​

Core API ​

Spec ​

validate ​

plan ​

Execution Stages ​

Match Strategies ​

Blocking Analysis ​

Survivorship Summary ​

Risk Flags ​

diff ​

Source ​

reconcile ​

Decisions ​

Telemetry ​

Engine Capabilities ​

uncertain_pairs (Cloud-only in v0.3.0+) ​

save / load ​

FeedbackLabel (Cloud-only in v0.3.0+) ​

Validation Rules ​

Schema Checks ​

Semantic Checks ​

Warnings ​

Safety Limits ​

Usage Examples ​

Notebook Validation ​

Full Reconciliation Pipeline ​

CI Pipeline Gate ​

Change Detection in CI ​

Pytest Integration ​

Pre-commit Hook ​

Architecture ​

Python SDK (Core)

Installation

Quick Start

Discover + Bootstrap Pipeline (Cloud)

Core API

Spec

validate

plan

Execution Stages

Match Strategies

Blocking Analysis

Survivorship Summary

Risk Flags

diff

Source

reconcile

Decisions

Telemetry

Engine Capabilities

uncertain_pairs (Cloud-only in v0.3.0+)

save / load

FeedbackLabel (Cloud-only in v0.3.0+)

Validation Rules

Schema Checks

Semantic Checks

Warnings

Safety Limits

Usage Examples

Notebook Validation

Full Reconciliation Pipeline

CI Pipeline Gate

Change Detection in CI

Pytest Integration

Pre-commit Hook

Architecture