Source code for promptdb.cli

"""Rich-powered CLI for local prompt operations.

The CLI provides six commands, all rendering output with Rich tables, panels,
and syntax-highlighted JSON. It uses the same :class:`~promptdb.client.PromptClient`
and :class:`~promptdb.service.PromptService` as the API.

Commands::

    promptdb init                 # scaffold a workspace with sample files
    promptdb list                 # list all registered prompt versions
    promptdb register-file <path> <namespace> <name>  # register from file
    promptdb resolve <ref>        # resolve a prompt reference to JSON
    promptdb render <ref> --var key=value  # render with variables
    promptdb export-file <ref> <path>     # write version bundle to disk

Prompt references use ``namespace/name:selector`` format::

    promptdb resolve support/triage:production
    promptdb resolve support/triage:rev:2
    promptdb resolve support/triage:latest

The ``register-file`` command supports YAML, JSON, and plain text files.
For plain text, specify ``--kind string`` or ``--kind chat``.

Entry point: ``promptdb = promptdb.cli:main`` (configured in pyproject.toml).
"""

from __future__ import annotations

import argparse
import json
from collections.abc import Sequence
from pathlib import Path

from rich.columns import Columns
from rich.console import Console
from rich.json import JSON
from rich.panel import Panel
from rich.syntax import Syntax
from rich.table import Table

from promptdb import PromptClient, PromptKind, PromptMetadata, PromptSpec
from promptdb.console import get_console
from promptdb.domain import MessageRole, TemplateFormat
from promptdb.files import save_prompt_spec

_SAMPLE_SPEC = PromptSpec(
    kind=PromptKind.CHAT,
    template_format=TemplateFormat.FSTRING,
    messages=[
        {
            "role": "system",
            "template": "You are a {persona} assistant for {company}.",
        },
        {
            "role": "human",
            "template": "{question}",
        },
    ],
    partial_variables={"persona": "helpful", "company": "OOAI"},
    metadata=PromptMetadata(
        title="Demo support assistant",
        description="Generated by `promptdb init`.",
        tags=["demo", "support"],
        user_version="v1",
    ),
)


def _build_parser() -> argparse.ArgumentParser:
    """Build the top-level CLI parser.

    Args:
        None.

    Returns:
        argparse.ArgumentParser: Configured parser.

    Raises:
        None.

    Examples:
        >>> _build_parser().prog
        'promptdb'
    """
    parser = argparse.ArgumentParser(prog="promptdb", description="Prompt registry CLI")
    subparsers = parser.add_subparsers(dest="command", required=True)

    init_parser = subparsers.add_parser("init", help="Create a local promptdb workspace.")
    init_parser.add_argument("--root", default=".", help="Workspace root directory.")
    init_parser.add_argument(
        "--force",
        action="store_true",
        help="Overwrite generated files if they already exist.",
    )

    subparsers.add_parser("list", help="List stored prompt versions.")

    register_parser = subparsers.add_parser("register-file", help="Register a prompt from a file.")
    register_parser.add_argument("path")
    register_parser.add_argument("namespace")
    register_parser.add_argument("name")
    register_parser.add_argument("--kind", choices=["string", "chat"], default=None)
    register_parser.add_argument("--alias", default="latest")
    register_parser.add_argument("--created-by", default=None)
    register_parser.add_argument("--user-version", default=None)
    register_parser.add_argument(
        "--message-role",
        choices=[role.value for role in MessageRole],
        default=MessageRole.HUMAN.value,
    )

    resolve_parser = subparsers.add_parser("resolve", help="Resolve a prompt reference.")
    resolve_parser.add_argument("ref")

    render_parser = subparsers.add_parser("render", help="Render a prompt reference.")
    render_parser.add_argument("ref")
    render_parser.add_argument("--var", action="append", default=[])

    export_parser = subparsers.add_parser(
        "export-file",
        help="Write a resolved version bundle to disk.",
    )
    export_parser.add_argument("ref")
    export_parser.add_argument("path")

    return parser


def _parse_kv_pairs(items: list[str]) -> dict[str, str]:
    """Parse repeated ``key=value`` arguments.

    Args:
        items: CLI values.

    Returns:
        dict[str, str]: Parsed variables.

    Raises:
        SystemExit: If a value is malformed.

    Examples:
        >>> _parse_kv_pairs(['name=Will'])
        {'name': 'Will'}
    """
    variables: dict[str, str] = {}
    for item in items:
        key, _, value = item.partition("=")
        if not key:
            raise SystemExit("--var entries must look like key=value")
        variables[key] = value
    return variables


def _render_version_table(client: PromptClient) -> Table:
    """Build a Rich table for stored versions.

    Args:
        client: Prompt client.

    Returns:
        Table: Renderable version table.

    Raises:
        None.

    Examples:
        .. code-block:: python

            table = _render_version_table(client)
    """
    table = Table(title="Registered Prompt Versions", expand=True)
    table.add_column("Namespace", style="accent")
    table.add_column("Name", style="accent")
    table.add_column("Selector")
    table.add_column("Version ID")
    table.add_column("Kind")
    table.add_column("User Version")
    table.add_column("Updated")

    for row in client.list_versions():
        table.add_row(
            row.namespace,
            row.name,
            row.ref.full_name,
            row.version_id,
            row.spec.kind.value,
            row.spec.metadata.user_version or "—",
            row.created_at.isoformat(timespec="seconds") if row.created_at else "—",
        )
    return table


def _cmd_init(root: Path, force: bool, console: Console) -> int:
    """Initialize a local workspace.

    Args:
        root: Workspace root.
        force: Whether to overwrite generated files.
        console: Rich console.

    Returns:
        int: Process exit status.

    Raises:
        OSError: If writing files fails.

    Examples:
        .. code-block:: python

            _cmd_init(Path('.'), False, get_console())
    """
    prompts_dir = root / "prompts"
    build_dir = root / "build"
    docs_dir = root / "docs"
    env_file = root / ".env.example"
    prompt_file = prompts_dir / "support_assistant.yaml"

    root.mkdir(parents=True, exist_ok=True)
    prompts_dir.mkdir(parents=True, exist_ok=True)
    build_dir.mkdir(parents=True, exist_ok=True)
    docs_dir.mkdir(parents=True, exist_ok=True)

    if force or not prompt_file.exists():
        save_prompt_spec(_SAMPLE_SPEC, prompt_file)
    if force or not env_file.exists():
        env_file.write_text(
            "PROMPTDB_DATABASE_URL=sqlite:///./promptdb.sqlite3\n"
            "PROMPTDB_BLOB_ROOT=.blobs\n"
            "PROMPTDB_STORAGE_BACKEND=local\n",
            encoding="utf-8",
        )

    console.print(
        Panel.fit(
            "[success]Workspace ready[/success]\n"
            f"[muted]Root:[/muted] {root.resolve()}\n"
            f"[muted]Prompt spec:[/muted] {prompt_file}\n"
            f"[muted]Environment example:[/muted] {env_file}",
            title="promptdb init",
            border_style="green",
        )
    )
    console.print(
        Columns(
            [
                Panel.fit(
                    "promptdb register-file prompts/support_assistant.yaml"
                    " demo assistant --alias production",
                    title="Register",
                ),
                Panel.fit(
                    "promptdb render demo/assistant:production"
                    " --var question='Where is my refund?'",
                    title="Render",
                ),
            ]
        )
    )
    return 0


[docs] def main(argv: Sequence[str] | None = None) -> int: """Run the CLI. Args: argv: Optional explicit command-line arguments. Returns: int: Process exit status. Raises: SystemExit: For invalid command usage. Examples: .. code-block:: python exit_code = main(['list']) """ parser = _build_parser() args = parser.parse_args(list(argv) if argv is not None else None) console = get_console() if args.command == "init": return _cmd_init(Path(args.root), args.force, console) client = PromptClient.from_env() if args.command == "list": versions = client.list_versions() if not versions: console.print( Panel.fit( "[warning]No prompt versions registered yet.[/warning]", title="promptdb", ) ) return 0 console.print(_render_version_table(client)) return 0 if args.command == "register-file": kind = PromptKind(args.kind) if args.kind is not None else None version = client.register_file( path=args.path, namespace=args.namespace, name=args.name, kind=kind, alias=args.alias, created_by=args.created_by, message_role=MessageRole(args.message_role), user_version=args.user_version, ) console.print( Panel.fit( f"[success]Registered[/success] {version.ref.full_name}", title="promptdb", ) ) console.print(JSON.from_data(version.model_dump(mode="json"))) return 0 if args.command == "resolve": version = client.resolve(args.ref) console.print( Panel.fit( f"[success]Resolved[/success] {version.ref.full_name}", title="promptdb", ) ) console.print(JSON.from_data(version.model_dump(mode="json"))) return 0 if args.command == "render": variables = _parse_kv_pairs(args.var) result = client.render(args.ref, variables) console.print(Panel.fit(f"[success]Rendered[/success] {args.ref}", title="promptdb")) console.print(JSON.from_data(result.model_dump(mode="json"))) if result.text: console.print( Panel( Syntax(result.text, "markdown", line_numbers=False), title="Rendered text", ) ) return 0 if args.command == "export-file": version = client.resolve(args.ref) path = client.export_file(version.ref, args.path) console.print( Panel.fit( f"[success]Exported[/success] {version.ref.full_name} -> {path}", title="promptdb", ) ) content = path.read_text(encoding="utf-8") if path.suffix.lower() == ".json": try: console.print(JSON.from_data(json.loads(content))) except json.JSONDecodeError: console.print(Syntax(content, "json", line_numbers=False)) else: language = "yaml" if path.suffix.lower() in {".yaml", ".yml"} else "text" console.print(Syntax(content, language, line_numbers=False)) return 0 parser.error(f"Unknown command: {args.command}") return 2