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"}


class Clock:
    """Deterministic clock: manually advanced, so every run is identical."""

    def __init__(self) -> None:
        self.current = datetime(2026, 8, 6, 12, 0, tzinfo=UTC)

    def advance(self, **kwargs) -> None:
        self.current += timedelta(**kwargs)


clock = Clock()


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 add_clock(kanta: Kanta) -> None:
    """Use the shared deterministic clock for all record timestamps."""

    @kanta.clock
    def fake_now() -> datetime:
        return clock.current


def add_logfmts(kanta: Kanta) -> None:
    """Resolve user ids to display names in headers and diff paths."""

    @kanta.logfmt(path="$user")
    def resolve_actor(value: str) -> str | None:
        return USERS.get(value)

    @kanta.logfmt
    def resolve_user_key(value: str, path: str) -> str | None:
        if path.startswith("users."):
            return USERS.get(value)
        return None


def add_header(kanta: Kanta) -> None:
    """Aligned rich header: actor, session id, action, target."""

    @kanta.logheader
    def header(action: str, user: str | None, extra: dict | None) -> str:
        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 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")

    kanta = Kanta(DB, DataV1())
    add_clock(kanta)
    add_logfmts(kanta)

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

    await kanta.open()

    clock.advance(minutes=2)
    with kanta.transaction(action="create", user="u2") as data:
        data.users["u2"] = {"name": "Bob", "role": "user"}

    clock.advance(minutes=5)
    with kanta.transaction(action="update", user="u1") as data:
        data.users["u2"]["role"] = "editor"
        data.counter = 1

    clock.advance(seconds=30)
    with kanta.transaction(action="delete", user="u1") as data:
        del data.users["u2"]

    # Display-only extra string, appended after the action.
    clock.advance(hours=1)
    with kanta.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.
    clock.advance(minutes=10)
    with kanta.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.transaction(action="reset", user="u1") as data:
            data.counter = 99
            raise ValueError("simulated failure")
    except ValueError:
        pass

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

    await kanta.close()

    section("Reopen with migrations and a custom log header")

    clock.advance(days=1)
    kanta = Kanta(DB, Data(), migrations=migrations)
    add_clock(kanta)
    add_logfmts(kanta)
    add_header(kanta)
    await kanta.open()

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

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

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