korm 2.0.0a8

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, and SQLite.

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, and health checks.

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"}).
• 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

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().

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

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: 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, and health checks. meta.prevCursor is currently always null (forward-only keyset).