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.callbacks import DictPost, DictPre
from kanta.logging import _ACTION, _ACTOR, _RESET, _SESSION, _TARGET
from kanta.logging import configure_logging

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


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

# 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)

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


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


@kanta_v0.logfmt
@kanta_v1.logfmt
def resolve_user(
    value: str, path: str, previous: DictPre, current: DictPost
) -> str | None:
    """Resolve user ids to names from the database state itself."""
    if path != "$user" and not path.startswith("users."):
        return None
    # Post-change state first, then pre-change (deleted users are only there).
    for state in (current, previous):
        name = state.get("users", {}).get(value, {}).get("name")
        if name:
            return name
    return None


@kanta_v1.logheader
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}"


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


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="u3",
        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())