72 lines
2 KiB
Python
72 lines
2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
import shutil
|
|
import sys
|
|
|
|
from infra_controller.config import load_config
|
|
from infra_controller.controller import InfraController
|
|
from infra_controller.discovery import InfraMetadata
|
|
|
|
|
|
def register() -> None:
|
|
parser = argparse.ArgumentParser(prog="infra-register")
|
|
parser.add_argument("metadata_file", help="Path to .infra.toml/.infra.yml (or any metadata file)")
|
|
parser.add_argument(
|
|
"--name",
|
|
help="Override project name (otherwise read from metadata_file)",
|
|
)
|
|
args = parser.parse_args(sys.argv[1:])
|
|
|
|
cfg = load_config()
|
|
src = Path(args.metadata_file)
|
|
if not src.exists():
|
|
print(f"metadata file not found: {src}", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
|
|
md = InfraMetadata.from_file(src)
|
|
name = args.name or md.project
|
|
|
|
dst_dir = cfg.discovery.file_based_path
|
|
dst_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
ext = src.suffix if src.suffix in {".toml", ".yml", ".yaml"} else ".toml"
|
|
dst = dst_dir / f"{name}{ext}"
|
|
tmp = dst.with_suffix(dst.suffix + ".tmp")
|
|
shutil.copyfile(src, tmp)
|
|
tmp.replace(dst)
|
|
|
|
|
|
def deregister() -> None:
|
|
parser = argparse.ArgumentParser(prog="infra-deregister")
|
|
parser.add_argument("name", help="Project name to deregister")
|
|
args = parser.parse_args(sys.argv[1:])
|
|
|
|
cfg = load_config()
|
|
dst_dir = cfg.discovery.file_based_path
|
|
|
|
removed = False
|
|
for ext in (".toml", ".yml", ".yaml"):
|
|
p = dst_dir / f"{args.name}{ext}"
|
|
if p.exists():
|
|
p.unlink()
|
|
removed = True
|
|
|
|
if not removed:
|
|
print(f"no registration found for: {args.name}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def status() -> None:
|
|
cfg = load_config()
|
|
print(f"config={cfg}")
|
|
|
|
|
|
def ensure_service_cli(argv: list[str] | None = None) -> None:
|
|
parser = argparse.ArgumentParser(prog="infra-ensure")
|
|
parser.add_argument("service")
|
|
args = parser.parse_args(argv)
|
|
|
|
cfg = load_config()
|
|
InfraController(cfg).ensure_service(args.service)
|