kanta/demo/main.py

"""Kanta feature demo.

Run from the project root:  python demo/main.py

Demonstrates bootstrap, colored transaction diffs, logfmt value formatting,
custom log headers, logging toggles, rollback, migrations, and a custom clock.
The database is recreated with fixed timestamps on every run; everything else
lives in this file.
"""

import asyncio
import logging
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import ModuleType

import msgspec

from kanta import Kanta
from kanta.logging import _ACTION, _ACTOR, _RESET, _SESSION, _TARGET
from kanta.logging import configure_logging

DB = Path(__file__).with_name("demo.kantadb")

# Fake directory: user id -> display name, resolved by the logfmt callbacks.
USERS = {"u1": "Alice", "u2": "Bob", "u3": "Carol"}

_now = datetime(2026, 8, 6, tzinfo=UTC)


def fake_now() -> datetime:
    """Deterministic clock: starts at midnight, +1h on every read."""
    global _now
    ts = _now
    _now += timedelta(hours=1)
    return ts


class DataV1(msgspec.Struct):
    """Original schema (version 0)."""

    users: dict[str, dict] = {}
    counter: int = 0


class Data(msgspec.Struct):
    """Current schema: migration v1 adds the settings section."""

    users: dict[str, dict] = {}
    counter: int = 0
    settings: dict[str, str] = {}


def migrate_v1(d: dict) -> None:
    """Add settings section"""
    d["settings"] = {"theme": "dark"}


migrations = ModuleType("demo_migrations")
migrations.migrate_v1 = migrate_v1


def resolve_actor(value: str) -> str | None:
    """Resolve the transaction user id to a display name."""
    return USERS.get(value)


def resolve_user_key(value: str, path: str) -> str | None:
    """Resolve user ids in diff paths to display names."""
    if path.startswith("users."):
        return USERS.get(value)
    return None


def header(action: str, user: str | None, extra: dict | None) -> str:
    """Aligned rich header: actor, session id, action, target."""
    actor = f"{_ACTOR}{user or '-':<8}{_RESET}"
    session = f"{_SESSION}{extra.get('session_id', '-'):>2}{_RESET}"
    target = f"{_TARGET}{extra['target']}{_RESET}"
    return f"{actor} {session}  {_ACTION}{action}{_RESET}  {target}"


def seed(data: DataV1) -> None:
    """Bootstrap: create the initial admin user."""
    data.users["u1"] = {"name": "Alice", "role": "admin"}


# Phase 1 instance: default logging, original schema.
kanta_v0 = Kanta(DB, DataV1())
# Phase 2 instance: migrations and a custom log header.
kanta_v1 = Kanta(DB, Data(), migrations=migrations)

for k in (kanta_v0, kanta_v1):
    k.clock(fake_now)
    k.logfmt(resolve_user_key)
    k.logfmt(resolve_actor, path="$user")

kanta_v0.bootstrap(seed)
kanta_v1.logheader(header)


def section(title: str) -> None:
    print(f"\n# {title}", flush=True)


async def main() -> None:
    DB.unlink(missing_ok=True)

    section("Standard logging: bootstrap, diffs, toggles, rollback")
    await kanta_v0.open()

    with kanta_v0.transaction(action="create", user="u2") as data:
        data.users["u2"] = {"name": "Bob", "role": "user"}

    with kanta_v0.transaction(action="update", user="u1") as data:
        data.users["u2"]["role"] = "editor"
        data.counter = 1

    with kanta_v0.transaction(action="delete", user="u1") as data:
        del data.users["u2"]

    # Display-only extra string, appended after the action.
    with kanta_v0.transaction(action="export", user="u1", extra=DB.name) as data:
        data.counter = 2

    # Compact logging, diff only: a system fix stamped by the clock, but the
    # modification time (m) is not updated.
    with kanta_v0.transaction(
        action="repair", mtime=False, log={"header": False, "diff": True}
    ) as data:
        data.users["u3"] = {"name": "Carol", "role": "user"}

    # A failing transaction rolls back and logs a warning.
    try:
        with kanta_v0.transaction(action="reset", user="u1") as data:
            data.counter = 99
            raise ValueError("simulated failure")
    except ValueError:
        pass

    # Compact logging, header only.
    with kanta_v0.transaction(
        action="import", user="u1", log={"header": True, "diff": False}
    ) as data:
        data.counter = 3

    await kanta_v0.close()

    section("Reopen with migrations and a custom log header")
    await kanta_v1.open()

    # No target given: defaults to the database filename.
    with kanta_v1.transaction(
        action="update", user="u1", extra={"session_id": 3}
    ) as data:
        data.settings["theme"] = "light"

    with kanta_v1.transaction(
        action="update",
        user="u2",
        extra={"session_id": 7, "target": "settings (demo)"},
    ) as data:
        data.settings["lang"] = "en"

    await kanta_v1.close()

    # The pretty names only exist in the logs; the database stores raw ids.
    section("Raw database records (user ids and timestamps, not pretty names)")
    print(DB.read_text(), end="", flush=True)


if __name__ == "__main__":
    configure_logging()
    logging.getLogger("kanta").setLevel(logging.DEBUG)  # show migration diffs
    asyncio.run(main())