korm 2.0.0a10

KORM Protocol v2 — JSON request/response protocol for dynamic database operations (reference implementation)

korm

Reference Python implementation of the KORM Protocol v2 — a JSON request/response protocol for dynamic database operations against PostgreSQL, MySQL/MariaDB, SQLite, and DuckDB.

A KORM server receives a JSON request naming a model and an action, executes it via SQLAlchemy Core 2.x, and returns a uniform response envelope. The protocol is fully describable by JSON Schema, so MCP tools, OpenAPI docs, and typed clients can be generated from one source.

import korm

db = korm.initialize_korm(url="postgresql+asyncpg://...", schema="korm.schema.json")

result = await db.process({          # async, v2-native
    "korm": 2,
    "action": "list",
    "model": "users",
    "where": {
        "is_active": True,
        "age": {"gte": 18, "lt": 65},
        "email": {"endsWith": "@example.com"},
    },
    "include": {"profile": True, "posts": {"limit": 5, "orderBy": "-created_at"}},
    "orderBy": "-created_at",
    "limit": 20,
})

result = db.process_sync(request_dict)   # sync engines

Every action returns the same envelope:

{
  "ok": true, "action": "list", "model": "users",
  "data": [ ... ],
  "meta": { "protocol": 2, "count": 20, "limit": 20, "offset": 0, "durationMs": 4 },
  "error": null
}

New in v2.0.0a8: Full-text search, subqueries, optimistic locking, database views, data seeding, OpenTelemetry tracing, Prometheus metrics, health checks, and audit logging.
New in v2.0.0a9: Admin panel, data classification & attribute-based access control (ABAC).
New in v2.0.0a10: MySQL, SQL Server, and CockroachDB backend support.

New in v2.0.0a4: attach read-only sub-requests via other_requests:

New in v2.0.0a7: Result caching, streaming responses, and slow query logging.

result = await db.process({
    "korm": 2, "action": "show", "model": "users",
    "where": {"id": 1},
    "other_requests": {
        "posts": {
            "action": "list",
            "where": {"author_id": 1},
            "limit": 10,
        },
        "stats": {
            "action": "aggregate",
            "aggregates": [{"fn": "count", "as": "post_count"}],
            "model": "posts",
        },
    },
})
# result["other_responses"]["posts"]["data"]  — list of posts
# result["other_responses"]["stats"]["data"]  — aggregate result

Features (spec v2.0)

• Actions: create, list, show, update, delete, restore, upsert, replace, sync, aggregate, batch (optionally atomic, with $ref chaining).
• Structured where grammar — no stringly-typed operator DSL: eq ne gt gte lt lte in notIn between notBetween like ilike contains startsWith endsWith is not, plus nestable and / or / not boolean trees with fully parenthesized SQL.
• Safe by construction — no raw SQL fragments anywhere; all identifiers validated against the schema; contains/startsWith/endsWith escape % and _.
• Pagination — offset mode and opaque keyset cursors (send "cursor": null for the first page, follow meta.nextCursor); optional HMAC-signed cursors via cursor_secret.
• Relations — hasOne / hasMany / belongsTo includes with per-relation select/where/orderBy/limit, nested dotted paths, and validated explicit joins. manyToMany is accepted in schemas (syntax v2.0) and executes from v2.1.
• Aggregates — count sum avg min max, groupBy, having, and JSON expression trees ({"add": ["salary", "bonus"]}) replacing v1's sumFormula strings.
• Soft delete — automatic deleted_at IS NULL filtering, withDeleted / onlyDeleted / hardDelete options, first-class restore.
• Hooks — beforeValidate, afterValidate, before<Action>, after<Action>, onError; sync or async; may mutate the request, short-circuit, or raise.
• Policy layer — per-role model/action allowlists, default_deny production mode, mandatory where-scopes for tenant isolation ({"org_id": "$ctx.orgId"}).
• Data Classification & ABAC (v2.0.0a9+) — column-level "classification": "pii" | "phi" | "public" tags on schema columns. Automatic redaction of classified data based on policy rules: redactClassifications (auto-redact columns by classification), unredactedClassifications (override for specific roles), redact (column-specific replacement patterns), and mask (partial-reveal patterns with {last4} support). See the Data Classification section below.
• v1 compatibility — a deterministic up-converter (on by default, compat_v1) accepts v1 operator strings (">=18", "><18,65", "[]a,b", "!x", "%like%", Or: keys, count/sum actions, string join conditions) and down-converts responses (bare count numbers, bare show objects) for v1 clients. strict_v2: true turns it off.
• New in v2.0.0a4: other_requests — attach read-only sub-requests (list, show, aggregate) to any primary action. Sub-requests execute after the primary action within the same transaction (for write actions) and return results under response.other_responses.<label>. Errors in a sub-request are captured per-label — a failing sub-request never fails the primary.
• Errors are codes, not prose: VALIDATION_FAILED, UNKNOWN_MODEL, UNKNOWN_COLUMN, UNKNOWN_RELATION, UNKNOWN_ACTION, NOT_FOUND, CONFLICT, FK_VIOLATION, POLICY_DENIED, LIMIT_EXCEEDED, TRANSACTION_FAILED, BATCH_ROLLED_BACK, UNSUPPORTED_ON_ENGINE, INTERNAL — with structured per-field details.

Install

pip install korm                 # SQLite via stdlib driver
pip install korm[postgres]       # psycopg + asyncpg
pip install korm[mysql]          # PyMySQL + aiomysql
pip install korm[fastapi]        # HTTP router
pip install korm[mcp]            # MCP server
pip install korm[opentelemetry]  # OpenTelemetry distributed tracing
pip install korm[prometheus]     # Prometheus metrics
pip install korm[admin]          # Admin panel (FastAPI + Jinja2)
pip install korm[audit]          # Audit log (built-in)
pip install korm[duckdb]         # DuckDB backend

Schema (korm.schema.json)

One file drives DDL sync, request validation, and generated types:

{
  "kormSchema": 2,
  "models": {
    "users": {
      "table": "users",
      "primaryKey": ["id"],
      "softDelete": { "column": "deleted_at" },
      "timestamps": { "createdAt": "created_at", "updatedAt": "updated_at" },
      "columns": {
        "id":       { "type": "increments" },
        "username": { "type": "string", "length": 80, "unique": true, "required": true },
        "email":    { "type": "string", "format": "email", "required": true }
      },
      "relations": {
        "profile": { "type": "hasOne", "model": "user_profiles", "foreignKey": "user_id" },
        "posts":   { "type": "hasMany", "model": "posts", "foreignKey": "author_id" }
      }
    }
  }
}

Fluent Query Builder (v2.0.0a6+)

Chainable Python API — no raw JSON dicts required:

# Sync
result = db.query("users").where(age__gte=18).limit(20).order_by("-created_at").list()

# Async
result = await db.query("users").where(age__gte=18).a.list()

# Build raw request dict for manual processing
req = db.query("users").where(age__gte=18).build("list")

Supported methods: .where(), .select(), .include(), .order_by(), .limit(), .offset(), .cursor(), .options(), .other() (sub-requests), .build(action), .run(action), plus terminal shorthands .list(), .show(), .create(data), .update(data), .delete(), .restore(), .upsert(data, conflict), .replace(data), .sync(data, conflict), and .aggregates(...).

WebSocket Subscriptions (v2.0.0a6+)

Subscribe to real-time mutation events:

async for event in db.subscribe("users", where={"org_id": 1}):
    print(event["event"], event["data"])  # "create" / "update" / "delete"

FastAPI WebSocket endpoint: ws://host/api/{model}/subscribe?where={"org_id":1}

Result Caching (v2.0.0a7+)

Opt-in per-request caching with auto-invalidation on mutations:

# Configure with default TTL
db = initialize_korm(url="sqlite:///app.db", schema=schema,
                     default_cache_ttl=30)  # 30s default

# Per-request override
result = db.process_sync({
    "korm": 2, "action": "list", "model": "users",
    "options": {"cache_ttl": 60},
})

# Bypass cache for this call
result = db.process_sync({
    "korm": 2, "action": "list", "model": "users",
    "options": {"cache_ttl": 0},
})

Built-in MemoryCache backend; bring your own via CacheBackend protocol.

Streaming Responses (v2.0.0a7+)

Stream rows one at a time using server-side cursors — no full-result buffering:

# Sync
for row in db.stream_sync("users", where={"is_active": True}):
    process(row)

# Async
async for row in db.stream("users"):
    process(row)

# QueryBuilder
for row in db.query("users").where(age__gte=18).stream():
    process(row)

# FastAPI endpoint: POST /{model}/stream returns NDJSON

Supports select column trimming, where, orderBy, and limit.

Slow Query Log (v2.0.0a7+)

Automatically log queries exceeding a duration threshold:

db = initialize_korm(url="postgresql://...", schema=schema,
                     slow_query_threshold_ms=500)  # log queries >500ms

On any request slower than the threshold, a structured dict is logged to korm.slow_query (Python logging module) containing: action, model, duration, compiled SQL, row count, and response metadata. Cache hits are skipped (no SQL executed).

Full-Text Search (v2.0.0a8+)

Three FTS operators backed by native SQL on each dialect:

result = db.process_sync({
    "korm": 2, "action": "list", "model": "posts",
    "where": {"title": {"search": "database optimization"}},
})
# PostgreSQL: title @@ plainto_tsquery('english', 'database optimization')
# MySQL:      MATCH(title) AGAINST('database optimization')
# SQLite:     title LIKE '%database%' AND title LIKE '%optimization%'

Other modes:

Operator	PostgreSQL	SQLite fallback	Use case
search	plainto_tsquery	LIKE on each word	Natural language
phrase	phraseto_tsquery	LIKE '%exact phrase%'	Exact phrase
websearch	websearch_to_tsquery	LIKE with % wildcards	Advanced search

When no orderBy is specified and a FTS operator is used, results are auto-ranked by ts_rank (PostgreSQL) / frequency (SQLite) descending.

Subqueries (v2.0.0a8+)

Reference another model's data in a where clause:

{"id": {"inSubquery": {
    "model": "posts", "select": ["author_id"],
    "where": {"views": {"gt": 10}},
}}}

Operators: inSubquery, notInSubquery, existsSubquery, notExistsSubquery.

Column-to-column comparisons use {"col": "table.column"} syntax:

{"existsSubquery": {
    "model": "posts",
    "where": {"author_id": {"eq": {"col": "users.id"}}},
}}

Optimistic Locking (v2.0.0a8+)

Add "version": true to any column in your schema:

{"version": { "type": "integer", "required": true, "version": true }}

Then include _version in update data or deletes:

# Update — version must match
db.process_sync({"korm": 2, "action": "update", "model": "posts",
    "where": {"id": 1}, "data": {"title": "new", "_version": 3}})

# Delete — version must match
db.process_sync({"korm": 2, "action": "delete", "model": "posts",
    "where": {"id": 1}, "_version": 3})

On mismatch, the server returns a CONFLICT error. The version column auto-increments on every write.

Database Views (v2.0.0a8+)

Define a read-only view in your schema:

{"active_users": {
    "table": "active_users_v",
    "modelType": "view",
    "columns": {
        "id": {"type": "integer"},
        "username": {"type": "string"}
    }
}}

• Write actions (create, update, delete, etc.) are rejected with UNSUPPORTED_ON_ENGINE
• sync_database skips view tables (manage the SQL view externally)
• primaryKey is optional — views use id column for default ordering
• diff reports view column mismatches

Data Seeding (v2.0.0a8+)

Generate realistic test data from the CLI:

korm seed --url sqlite:///dev.db --schema korm.schema.json --count 25
korm seed --url sqlite:///dev.db users --count 10 --overrides '{"role": "admin"}'

Or from Python:

from korm.seed import SeedFactory

factory = SeedFactory(schema)
row = factory.row("users", overrides={"role": "admin"})
# row = {"username": "users_1", "email": "[e-mail]", "age": 34, ...}

Supports per-model counts and overrides via YAML profile files (--profile seeding.yaml).

OpenTelemetry Tracing (v2.0.0a8+)

Distributed tracing with no-op fallback when packages are absent:

pip install korm[opentelemetry]

from korm.tracing import setup_tracing

provider = setup_tracing(service_name="my-app")
db = initialize_korm(url="postgresql://...", schema="...",
                     tracer_provider=provider)

Creates a korm.request span per action with attributes: korm.action, korm.model, korm.duration_ms, korm.row_count, korm.error_code. Auto-instruments SQLAlchemy if opentelemetry-instrumentation-sqlalchemy is available (DB call spans as children).

Prometheus Metrics (v2.0.0a8+)

Zero-cost metrics when prometheus_client is not installed:

pip install korm[prometheus]

from korm.metrics import MetricsRegistry

metrics = MetricsRegistry(service_name="my-app")
db = initialize_korm(url="sqlite:///app.db", schema="...", metrics=metrics)

Metrics:

• korm_requests_total{action, model, status} — request counter
• korm_request_duration_seconds{action, model} — latency histogram
• korm_db_errors_total{action, model, code} — error counter
• korm_cache_operations_total{operation} — cache hit/miss/invalidate

Scrape endpoint: GET /metrics (auto-added to the FastAPI router).

Health Checks (v2.0.0a8+)

from korm.health import HealthRegistry, db_connectivity_check

health = HealthRegistry()
health.add_check("database", db_connectivity_check(engine))
status = health.check()
# {"status": "healthy", "checks": [{"name": "database", "status": "ok"}]}

FastAPI endpoints: GET /health and GET /ready (503 when degraded), added automatically by crud_router().

Data Classification & Attribute-Based Access Control (v2.0.0a9+)

Column-level data classification tags drive automatic redaction and masking in responses:

{
  "models": {
    "users": {
      "columns": {
        "id":       { "type": "integer" },
        "username": { "type": "string" },
        "email":    { "type": "string", "classification": "pii" },
        "ssn":      { "type": "string", "classification": "phi" },
        "phone":    { "type": "string", "classification": "pii" }
      }
    }
  }
}

Then define redaction rules in the policy:

db = initialize_korm(url="sqlite:///app.db", schema=schema, policy={
    "mode": "default_deny",
    "roles": {
        "billing": {"models": {"users": {
            "actions": ["list", "show"],
            "readableColumns": ["id", "username", "email", "ssn"],
            "redactClassifications": ["phi"],              # redact phi columns → "***"
        }}},
        "compliance": {"models": {"users": {
            "actions": ["list", "show"],
            "readableColumns": ["id", "username", "email", "ssn"],
            "redactClassifications": ["pii", "phi"],
            "unredactedClassifications": ["pii"],           # override: see pii but not phi
        }}},
        "support": {"models": {"users": {
            "actions": ["list"],
            "readableColumns": ["id", "email"],
            "redact": {"email": "***@***.***"},            # static replacement
            "mask": {"phone": "***-***-{last4}"},          # partial reveal
        }}},
    }
})

Policy key	Description
redactClassifications	List of classifications whose values are replaced with "***"
unredactedClassifications	Override: classifications the role may see
redact	Dict of {column: replacement_string} for static redaction
mask	Dict of {column: pattern} with {last4} placeholder for last 4 chars

Classification metadata is also exposed in the generated OpenAPI spec as x-classification on each property.

OpenAPI 3.1 Spec (v2.0.0a6+)

Auto-generate an OpenAPI spec from your schema:

from korm import generate_openapi
spec = generate_openapi(schema)

CLI: korm serve-openapi --url sqlite:///app.db --schema korm.schema.json

DuckDB Backend (v2.0.0a10+)

KORM supports additional database backends beyond the built-in SQLite:

pip install korm[mysql,mssql,cockroachdb,duckdb]  # install all backends
# or individually:
pip install korm[mysql]          # MySQL
pip install korm[mssql]          # SQL Server
pip install korm[cockroachdb]    # CockroachDB
pip install korm[duckdb]         # DuckDB

MySQL — works out-of-the-box via mysql+pymysql://. No patches needed.

db = initialize_korm(url="mysql+pymysql://user:pass@localhost/mydb", schema=...)

SQL Server — uses mssql+pyodbc://. SQLAlchemy's dialect handles TOP/LIMIT, OUTPUT INSERTED.*/RETURNING, and OFFSET FETCH natively.

db = initialize_korm(url="mssql+pyodbc://user:pass@localhost:1433/mydb?driver=ODBC+Driver+18+for+SQL+Server", schema=...)

CockroachDB — PostgreSQL-wire-compatible. Automatic retry (max 3 attempts, exponential backoff) on serialization conflicts (SQLSTATE 40001).

db = initialize_korm(url="cockroachdb://root@localhost:26257/defaultdb?sslmode=disable", schema=...)
# or using standard PostgreSQL URL:
db = initialize_korm(url="postgresql://root@localhost:26257/defaultdb?sslmode=disable", schema=...)

DuckDB — embedded analytical engine for OLAP workloads:

db = initialize_korm(url="duckdb:///:memory:", schema=...)
sync_database(db.engine, db.schema)

result = db.process_sync({"korm": 2, "action": "list", "model": "users",
                          "where": {"age": {"gte": 18}}})

Also works with file-based databases: duckdb:///path/to/data.duckdb.

Compatibility notes:

• MySQL: jsonContains operator uses PostgreSQL @> syntax; MySQL JSON_CONTAINS can be registered as a custom operator
• SQL Server: cursor pagination uses OFFSET FETCH NEXT ROWS ONLY
• CockroachDB: increments/bigIncrements work but UUID primary keys are recommended for distributed workloads; retry logic applies to CockroachDB dialect (cockroachdb://) — PostgreSQL URL users get no retry
• DuckDB: increments/bigIncrements use plain INTEGER (no SERIAL support; auto-patched); full-text search falls back to LIKE-based operators; async not supported; JSON column storage works

All core actions (create, list, show, update, delete, upsert, replace, sync, aggregate, batch) and where operators work across all backends.

FastAPI

from fastapi import FastAPI
from korm.fastapi import crud_router

app = FastAPI()
app.include_router(crud_router(db), prefix="/api")
# POST /api/{model}/crud

CLI

korm init                    # scaffold korm.schema.json
korm generate-schema --url sqlite:///app.db      # introspect a live database
korm sync  --url ... --schema korm.schema.json   # dev-time DDL sync
korm diff  --url ... --schema korm.schema.json   # migration skeleton
korm serve-mcp --url ... --schema ...            # per-model MCP tools over stdio
korm serve-openapi --url ... --schema ...        # OpenAPI 3.1 spec + Swagger UI
korm serve-openapi --url ... --schema ... --no-serve  # print spec as JSON

sync is a dev/prototyping tool; for production use migration tools (Alembic/Knex) generated from korm diff output.

Development

pip install -e ".[dev]"
pytest

Status

Implements all 15 tier-0 through tier-3 features of KORM Protocol v2, plus all four tier-4 backends (MySQL, SQL Server, CockroachDB, DuckDB) and tier-5 audit logging, admin panel, and data classification: CRUD, structured where grammar, pagination (offset + keyset cursors), relations (hasOne/hasMany/belongsTo/manyToMany), aggregates (count/sum/avg/min/max with groupBy/having), soft delete, hooks, policy layer, v1 compatibility, other_requests sub-requests, fluent query builder, OpenAPI spec generation, WebSocket subscriptions, result caching, streaming responses, slow query logging, full-text search, subqueries, optimistic locking, database views, data seeding, OpenTelemetry tracing, Prometheus metrics, health checks, audit logging, admin panel, data classification/ABAC, and all four database backends. meta.prevCursor is currently always null (forward-only keyset).