"""Rich-powered CLI for :mod:`promptdb`.
Purpose:
Provide local workflows for initializing a workspace, registering prompts,
listing versions, resolving selectors, rendering prompts, and exporting
bundles.
Design:
The CLI uses the same local client and service layers as the API while
presenting results with Rich tables, panels, and syntax-highlighted JSON.
Attributes:
main: CLI entry point.
Examples:
.. code-block:: bash
promptdb init
promptdb list
promptdb resolve support/triage:production
promptdb render support/triage:production --var question="Where is my refund?"
"""
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