463 lines
14 KiB
Django/Jinja
463 lines
14 KiB
Django/Jinja
{% if database == 'postgres' -%}
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from rich.console import Console
|
|
from rich.json import JSON
|
|
from rich.panel import Panel
|
|
from rich.table import Table
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
|
|
|
from collections.abc import AsyncGenerator
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from dependencies.container import container
|
|
from utils.db.models.config import DynamicConfigBase
|
|
from utils.db.repositories import ConfigRepository
|
|
|
|
console = Console()
|
|
_MISSING = object()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def get_repo() -> AsyncGenerator[ConfigRepository]:
|
|
repo = await container.get(ConfigRepository)
|
|
try:
|
|
yield repo
|
|
finally:
|
|
await container.close()
|
|
|
|
|
|
def parse_value(raw: str) -> Any:
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
lowered = raw.lower()
|
|
if lowered in {"true", "false"}:
|
|
return lowered == "true"
|
|
if lowered in {"null", "none"}:
|
|
return None
|
|
try:
|
|
return int(raw)
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
return float(raw)
|
|
except ValueError:
|
|
return raw
|
|
|
|
|
|
def dig(data: dict[str, Any], path: str) -> Any:
|
|
current: Any = data
|
|
for key in path.split("."):
|
|
if not isinstance(current, dict) or key not in current:
|
|
return _MISSING
|
|
current = current[key]
|
|
return current
|
|
|
|
|
|
def set_path(data: dict[str, Any], path: str, value: Any) -> None:
|
|
keys = path.split(".")
|
|
current = data
|
|
for key in keys[:-1]:
|
|
nxt = current.get(key)
|
|
if not isinstance(nxt, dict):
|
|
nxt = {}
|
|
current[key] = nxt
|
|
current = nxt
|
|
current[keys[-1]] = value
|
|
|
|
|
|
def unset_path(data: dict[str, Any], path: str) -> bool:
|
|
keys = path.split(".")
|
|
current = data
|
|
for key in keys[:-1]:
|
|
nxt = current.get(key)
|
|
if not isinstance(nxt, dict):
|
|
return False
|
|
current = nxt
|
|
return current.pop(keys[-1], _MISSING) is not _MISSING
|
|
|
|
|
|
def validate(data: dict[str, Any]) -> DynamicConfigBase:
|
|
try:
|
|
return DynamicConfigBase.model_validate(data)
|
|
except ValidationError as exc:
|
|
console.print("[bold red]✗ Invalid config[/]")
|
|
for error in exc.errors():
|
|
loc = ".".join(str(part) for part in error["loc"])
|
|
console.print(f" [red]{loc or '<root>'}[/]: {error['msg']}")
|
|
raise SystemExit(1) from exc
|
|
|
|
|
|
def render(config: DynamicConfigBase) -> None:
|
|
console.print(
|
|
Panel(
|
|
JSON(config.model_dump_json(indent=2)),
|
|
title="[bold cyan]DynamicConfig[/]",
|
|
border_style="cyan",
|
|
expand=False,
|
|
)
|
|
)
|
|
|
|
|
|
async def cmd_show() -> None:
|
|
async with get_repo() as repo:
|
|
render(await repo.get())
|
|
|
|
|
|
async def cmd_get(path: str) -> None:
|
|
async with get_repo() as repo:
|
|
config = await repo.get()
|
|
value = dig(config.model_dump(), path)
|
|
if value is _MISSING:
|
|
console.print(f"[red]✗[/] no such field: [bold]{path}[/]")
|
|
raise SystemExit(1)
|
|
table = Table(show_header=False, box=None)
|
|
table.add_column(style="cyan")
|
|
table.add_column(style="white")
|
|
table.add_row(path, json.dumps(value, ensure_ascii=False))
|
|
console.print(table)
|
|
|
|
|
|
async def cmd_set(path: str, raw: str) -> None:
|
|
async with get_repo() as repo:
|
|
data = (await repo.get()).model_dump()
|
|
value = parse_value(raw)
|
|
set_path(data, path, value)
|
|
config = await repo.save(validate(data))
|
|
shown = json.dumps(value, ensure_ascii=False)
|
|
console.print(f"[green]✓[/] set [cyan]{path}[/] = {shown}")
|
|
render(config)
|
|
|
|
|
|
async def cmd_unset(path: str) -> None:
|
|
async with get_repo() as repo:
|
|
data = (await repo.get()).model_dump()
|
|
if not unset_path(data, path):
|
|
console.print(f"[red]✗[/] no such field: [bold]{path}[/]")
|
|
raise SystemExit(1)
|
|
config = await repo.save(validate(data))
|
|
console.print(f"[green]✓[/] unset [cyan]{path}[/] (reset to default)")
|
|
render(config)
|
|
|
|
|
|
async def _mutate_list(path: str, raw: str, *, add: bool) -> None:
|
|
async with get_repo() as repo:
|
|
data = (await repo.get()).model_dump()
|
|
current = dig(data, path)
|
|
items = list(current) if isinstance(current, list) else []
|
|
value = parse_value(raw)
|
|
if add:
|
|
if value not in items:
|
|
items.append(value)
|
|
else:
|
|
items = [item for item in items if item != value]
|
|
set_path(data, path, items)
|
|
config = await repo.save(validate(data))
|
|
verb = "added to" if add else "removed from"
|
|
shown = json.dumps(value, ensure_ascii=False)
|
|
console.print(f"[green]✓[/] {shown} {verb} [cyan]{path}[/]")
|
|
render(config)
|
|
|
|
|
|
async def cmd_reset() -> None:
|
|
async with get_repo() as repo:
|
|
config = await repo.reset()
|
|
console.print("[yellow]↺[/] config reset to defaults")
|
|
render(config)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Manage DynamicConfig in Postgres")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("show", help="Show the current config")
|
|
|
|
get_p = sub.add_parser("get", help="Show a single field")
|
|
get_p.add_argument("path", help="Dotted field path, e.g. bot.admins")
|
|
|
|
set_p = sub.add_parser("set", help="Set a field (value is smart-parsed)")
|
|
set_p.add_argument("path", help="Dotted field path, e.g. bot.admins")
|
|
set_p.add_argument("value", help="New value (JSON or scalar)")
|
|
|
|
unset_p = sub.add_parser("unset", help="Remove a field (falls back to default)")
|
|
unset_p.add_argument("path")
|
|
|
|
add_p = sub.add_parser("add", help="Append an item to a list field")
|
|
add_p.add_argument("path", help="Dotted path to a list, e.g. bot.admins")
|
|
add_p.add_argument("value")
|
|
|
|
rm_p = sub.add_parser("remove", help="Remove an item from a list field")
|
|
rm_p.add_argument("path", help="Dotted path to a list, e.g. bot.admins")
|
|
rm_p.add_argument("value")
|
|
|
|
sub.add_parser("reset", help="Reset the whole config to defaults")
|
|
return parser
|
|
|
|
|
|
async def run(args: argparse.Namespace) -> None:
|
|
match args.command:
|
|
case "show":
|
|
await cmd_show()
|
|
case "get":
|
|
await cmd_get(args.path)
|
|
case "set":
|
|
await cmd_set(args.path, args.value)
|
|
case "unset":
|
|
await cmd_unset(args.path)
|
|
case "add":
|
|
await _mutate_list(args.path, args.value, add=True)
|
|
case "remove":
|
|
await _mutate_list(args.path, args.value, add=False)
|
|
case "reset":
|
|
await cmd_reset()
|
|
|
|
|
|
def main() -> None:
|
|
args = build_parser().parse_args()
|
|
asyncio.run(run(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
{%- else -%}
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from rich.console import Console
|
|
from rich.json import JSON
|
|
from rich.panel import Panel
|
|
from rich.table import Table
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from utils.db import client, init_db
|
|
from utils.db.models.config import DynamicConfig, DynamicConfigBase
|
|
|
|
console = Console()
|
|
_MISSING = object()
|
|
|
|
|
|
async def read_config() -> DynamicConfigBase:
|
|
await init_db()
|
|
doc = await DynamicConfig.get_or_create()
|
|
return DynamicConfigBase.model_validate(doc, from_attributes=True)
|
|
|
|
|
|
async def write_config(config: DynamicConfigBase) -> DynamicConfigBase:
|
|
await init_db()
|
|
doc = await DynamicConfig.get_or_create()
|
|
for name in DynamicConfigBase.model_fields:
|
|
setattr(doc, name, getattr(config, name))
|
|
await doc.save()
|
|
return config
|
|
|
|
|
|
def parse_value(raw: str) -> Any:
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
lowered = raw.lower()
|
|
if lowered in {"true", "false"}:
|
|
return lowered == "true"
|
|
if lowered in {"null", "none"}:
|
|
return None
|
|
try:
|
|
return int(raw)
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
return float(raw)
|
|
except ValueError:
|
|
return raw
|
|
|
|
|
|
def dig(data: dict[str, Any], path: str) -> Any:
|
|
current: Any = data
|
|
for key in path.split("."):
|
|
if not isinstance(current, dict) or key not in current:
|
|
return _MISSING
|
|
current = current[key]
|
|
return current
|
|
|
|
|
|
def set_path(data: dict[str, Any], path: str, value: Any) -> None:
|
|
keys = path.split(".")
|
|
current = data
|
|
for key in keys[:-1]:
|
|
nxt = current.get(key)
|
|
if not isinstance(nxt, dict):
|
|
nxt = {}
|
|
current[key] = nxt
|
|
current = nxt
|
|
current[keys[-1]] = value
|
|
|
|
|
|
def unset_path(data: dict[str, Any], path: str) -> bool:
|
|
keys = path.split(".")
|
|
current = data
|
|
for key in keys[:-1]:
|
|
nxt = current.get(key)
|
|
if not isinstance(nxt, dict):
|
|
return False
|
|
current = nxt
|
|
return current.pop(keys[-1], _MISSING) is not _MISSING
|
|
|
|
|
|
def validate(data: dict[str, Any]) -> DynamicConfigBase:
|
|
try:
|
|
return DynamicConfigBase.model_validate(data)
|
|
except ValidationError as exc:
|
|
console.print("[bold red]✗ Invalid config[/]")
|
|
for error in exc.errors():
|
|
loc = ".".join(str(part) for part in error["loc"])
|
|
console.print(f" [red]{loc or '<root>'}[/]: {error['msg']}")
|
|
raise SystemExit(1) from exc
|
|
|
|
|
|
def render(config: DynamicConfigBase) -> None:
|
|
console.print(
|
|
Panel(
|
|
JSON(config.model_dump_json(indent=2)),
|
|
title="[bold cyan]DynamicConfig[/]",
|
|
border_style="cyan",
|
|
expand=False,
|
|
)
|
|
)
|
|
|
|
|
|
async def cmd_show() -> None:
|
|
render(await read_config())
|
|
|
|
|
|
async def cmd_get(path: str) -> None:
|
|
config = await read_config()
|
|
value = dig(config.model_dump(), path)
|
|
if value is _MISSING:
|
|
console.print(f"[red]✗[/] no such field: [bold]{path}[/]")
|
|
raise SystemExit(1)
|
|
table = Table(show_header=False, box=None)
|
|
table.add_column(style="cyan")
|
|
table.add_column(style="white")
|
|
table.add_row(path, json.dumps(value, ensure_ascii=False))
|
|
console.print(table)
|
|
|
|
|
|
async def cmd_set(path: str, raw: str) -> None:
|
|
data = (await read_config()).model_dump()
|
|
value = parse_value(raw)
|
|
set_path(data, path, value)
|
|
config = await write_config(validate(data))
|
|
shown = json.dumps(value, ensure_ascii=False)
|
|
console.print(f"[green]✓[/] set [cyan]{path}[/] = {shown}")
|
|
render(config)
|
|
|
|
|
|
async def cmd_unset(path: str) -> None:
|
|
data = (await read_config()).model_dump()
|
|
if not unset_path(data, path):
|
|
console.print(f"[red]✗[/] no such field: [bold]{path}[/]")
|
|
raise SystemExit(1)
|
|
config = await write_config(validate(data))
|
|
console.print(f"[green]✓[/] unset [cyan]{path}[/] (reset to default)")
|
|
render(config)
|
|
|
|
|
|
async def _mutate_list(path: str, raw: str, *, add: bool) -> None:
|
|
data = (await read_config()).model_dump()
|
|
current = dig(data, path)
|
|
items = list(current) if isinstance(current, list) else []
|
|
value = parse_value(raw)
|
|
if add:
|
|
if value not in items:
|
|
items.append(value)
|
|
else:
|
|
items = [item for item in items if item != value]
|
|
set_path(data, path, items)
|
|
config = await write_config(validate(data))
|
|
verb = "added to" if add else "removed from"
|
|
shown = json.dumps(value, ensure_ascii=False)
|
|
console.print(f"[green]✓[/] {shown} {verb} [cyan]{path}[/]")
|
|
render(config)
|
|
|
|
|
|
async def cmd_reset() -> None:
|
|
config = await write_config(DynamicConfigBase())
|
|
console.print("[yellow]↺[/] config reset to defaults")
|
|
render(config)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Manage DynamicConfig in MongoDB")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("show", help="Show the current config")
|
|
|
|
get_p = sub.add_parser("get", help="Show a single field")
|
|
get_p.add_argument("path", help="Dotted field path, e.g. bot.admins")
|
|
|
|
set_p = sub.add_parser("set", help="Set a field (value is smart-parsed)")
|
|
set_p.add_argument("path", help="Dotted field path, e.g. bot.admins")
|
|
set_p.add_argument("value", help="New value (JSON or scalar)")
|
|
|
|
unset_p = sub.add_parser("unset", help="Remove a field (falls back to default)")
|
|
unset_p.add_argument("path")
|
|
|
|
add_p = sub.add_parser("add", help="Append an item to a list field")
|
|
add_p.add_argument("path", help="Dotted path to a list, e.g. bot.admins")
|
|
add_p.add_argument("value")
|
|
|
|
rm_p = sub.add_parser("remove", help="Remove an item from a list field")
|
|
rm_p.add_argument("path", help="Dotted path to a list, e.g. bot.admins")
|
|
rm_p.add_argument("value")
|
|
|
|
sub.add_parser("reset", help="Reset the whole config to defaults")
|
|
return parser
|
|
|
|
|
|
async def run(args: argparse.Namespace) -> None:
|
|
try:
|
|
match args.command:
|
|
case "show":
|
|
await cmd_show()
|
|
case "get":
|
|
await cmd_get(args.path)
|
|
case "set":
|
|
await cmd_set(args.path, args.value)
|
|
case "unset":
|
|
await cmd_unset(args.path)
|
|
case "add":
|
|
await _mutate_list(args.path, args.value, add=True)
|
|
case "remove":
|
|
await _mutate_list(args.path, args.value, add=False)
|
|
case "reset":
|
|
await cmd_reset()
|
|
finally:
|
|
await client.close()
|
|
|
|
|
|
def main() -> None:
|
|
args = build_parser().parse_args()
|
|
asyncio.run(run(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
{%- endif %}
|