Project Setup & History¶
Initialize a Kontra project, track validation history, and compare runs over time.
Initialize a Project¶
Creates:
- .kontra/config.yml - Project configuration
- contracts/ - Directory for validation contracts
The config file contains documented defaults and examples:
version: "1"
defaults:
preplan: "on" # on | off
pushdown: "on" # on | off
projection: "on" # on | off
output_format: "rich" # rich | json
state_backend: "local"
datasources: {}
# prod_db:
# type: postgres
# host: ${PGHOST}
# ...
Named Datasources¶
Define datasources once in config, reference them everywhere:
# .kontra/config.yml
datasources:
prod_db:
type: postgres
host: ${PGHOST}
user: ${PGUSER}
password: ${PGPASSWORD}
database: ${PGDATABASE}
tables:
users: public.users
orders: public.orders
data_lake:
type: s3
bucket: ${S3_BUCKET}
prefix: warehouse/
tables:
events: events.parquet
Then use them:
Credentials stay in config (or environment variables). Contracts stay clean and portable.
Validation History¶
When you run kontra validate, results are automatically saved to a state backend. This enables comparing runs over time.
Contract identity is based on a fingerprint of the contract's name and rules fields. Renaming the name field creates a new fingerprint, so previous history and diff comparisons will not carry over. See Contracts Reference for details.
View History¶
# Show all runs for a contract
kontra history contract.yml
# Recent runs only
kontra history contract.yml --since 7d
# Failed runs only
kontra history contract.yml --failed-only
# JSON output
kontra history contract.yml -o json
Compare Runs (Diff)¶
# Compare latest to previous
kontra diff contract.yml
# Compare to 7 days ago
kontra diff contract.yml --since 7d
# Compare to specific date
kontra diff contract.yml --run 2024-01-15
# Output formats
kontra diff contract.yml -o json # CI/CD integration
kontra diff contract.yml -o llm # Token-optimized
Python API¶
import kontra
# Compare latest to previous run
diff = kontra.diff("my_contract")
if diff.regressed:
print("Quality regressed!")
for rule in diff.new_failures:
print(f" NEW: {rule.rule_id}")
if diff.improved:
print("Quality improved!")
for rule in diff.resolved:
print(f" RESOLVED: {rule.rule_id}")
# Diff properties
diff.has_changes # bool
diff.improved # bool
diff.regressed # bool
diff.before # run summary
diff.after # run summary
diff.new_failures # list of new failures
diff.resolved # list of resolved failures
diff.count_changes # list of count changes
# Serialize
diff.to_dict()
diff.to_json()
diff.to_llm()
History API¶
# List past runs
runs = kontra.list_runs("my_contract")
for run in runs:
print(f"{run['timestamp']}: {'PASS' if run['passed'] else 'FAIL'}")
# Get specific run
result = kontra.get_run("my_contract") # latest
result = kontra.get_run("my_contract", run_id="2024-01-15T10:30:00")
# Check if history exists
if kontra.has_runs("my_contract"):
diff = kontra.diff("my_contract")
Profile History¶
Validation history tracks contracts. Profile history tracks the shape of a source — its columns, dtypes, null rates, and distinct counts — over time. Profiles are only saved when you ask for it, either from the CLI or the API:
kontra profile users.parquet --save-profile
kontra profile-diff users.parquet
kontra profile-diff users.parquet --since 7d
kontra.profile("users.parquet", save=True)
kontra.get_profile("users.parquet") # latest saved profile
kontra.list_profiles("users.parquet") # saved runs, newest first
diff = kontra.profile_diff("users.parquet") # latest vs previous
diff = kontra.profile_diff("users.parquet", since="7d")
Both entry points write to the same store, so a profile saved by the CLI is
visible to the API and vice versa. A source is identified by a fingerprint of
its resolved URI; named datasources are resolved before fingerprinting, so
prod_db.users and its underlying URI refer to the same history. Inline
DataFrames have no stable identity and are not saved. profile_diff() returns
None when there is not enough history to compare.
Profile Store Backends¶
By default profiles are stored locally, under .kontra/profiles/ in the current
directory (one subdirectory per source, one JSON file per run). This is the same
local-vs-database split the validation-state backends use, and needs no setup.
For a shared, multi-host store, profiles can live in PostgreSQL instead — a
single kontra_profiles table, one row per run:
from kontra.scout.store import get_profile_store
store = get_profile_store("postgres", uri="postgres://user:pass@host/db")
# or rely on PGHOST / PGUSER / ... / DATABASE_URL:
store = get_profile_store("postgres")
The local store is the default; the Postgres backend is opt-in and requires the
kontra[postgres] extra. Creating a Postgres profile store does not replace the
public API's default local store: save and query it explicitly. The official MCP
server owns and injects its PostgreSQL profile store for this reason.
State Backends¶
Configure where validation history is stored:
# .kontra/config.yml
defaults:
state_backend: "local" # default
environments:
production:
state_backend: postgres://${PGHOST}/${PGDATABASE}
staging:
state_backend: s3://${S3_BUCKET}/kontra-state/
Local (Default)¶
State stored in .kontra/state/ directory. No setup required.
PostgreSQL¶
State stored in normalized database tables (kontra_runs,
kontra_rule_results, kontra_annotations). Profiles use the separate
kontra_profiles table.
S3¶
State stored as JSON files in S3.
Requires pip install kontra[s3] and AWS credentials.
SQL Server¶
State stored in database tables.
Annotations¶
Annotations provide "memory without authority"—agents and humans can record context about validation runs without affecting Kontra's behavior.
import kontra
# Annotate the latest run
kontra.annotate(
"users_contract.yml",
actor_type="agent",
actor_id="repair-agent-v2",
annotation_type="resolution",
summary="Fixed null emails by backfilling from user_profiles",
)
# Annotate a specific rule
kontra.annotate(
"users_contract.yml",
rule_id="COL:email:not_null",
actor_type="human",
actor_id="alice@example.com",
annotation_type="false_positive",
summary="Service accounts are expected to have null emails",
)
# Load run with annotations
result = kontra.get_run_with_annotations("users_contract.yml")
for ann in result.annotations or []:
print(f"[{ann['annotation_type']}] {ann['summary']}")
# Query annotations across runs (agent memory)
hints = kontra.get_annotations(
"users_contract.yml",
rule_id="COL:email:not_null",
)
for hint in hints:
print(f"[{hint['annotation_type']}] {hint['summary']}")
annotation_type is an open vocabulary: any non-empty string is accepted, so a workflow can define its own verdicts. The documented types are suggestions surfaced by tooling — resolution, root_cause, false_positive, acknowledged, suppressed, note, plus diagnosis (a first responder's assessment of a failure) and expected (an owner's verdict in an adjudication flow).
Key invariant: Kontra never reads annotations during validation or diff. They're purely for consumer use.
Annotations are stored in:
- kontra_annotations table (PostgreSQL, SQL Server)
- <run_id>.ann.jsonl files (local, S3)
Disabling State¶
Skip saving results to state backend: