AkurAI-EC2 / libexec/akurai-ec2-manifest.py

Latest change d8aa29121486f4b02d0016888d94fa66cac36ffe - Checkpoint WIP + rustfmt + AkurAI Build CI/CD pipeline by Ólafur Búi Ólafsson

#!/usr/bin/env python3
"""Flatten AkurAI EC2 TOML manifests into shell-friendly key/value records."""

from __future__ import annotations

import argparse
import datetime as dt
import json
import sys
import tomllib
from pathlib import Path
from typing import Any, Iterable


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Flatten a TOML manifest into key<TAB>value records."
    )
    parser.add_argument("manifest", help="Path to a TOML manifest file.")
    return parser.parse_args()


def encode_scalar(value: Any) -> str | None:
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, (dt.date, dt.datetime, dt.time)):
        return value.isoformat()
    if isinstance(value, str):
        if any(ch in value for ch in ("\t", "\r", "\n", "\0")):
            # Not representable in the flat TSV map (one record per line).
            # Multi-line values (e.g. hooks.pre_deploy/post_deploy) are parsed
            # directly from the TOML by dedicated readers instead — skip here.
            return None
        return value
    return None


def encode_value(value: Any) -> str | None:
    if isinstance(value, list):
        if any(isinstance(item, (dict, list)) for item in value):
            # Array-of-tables (e.g. [[extra_unit]]) — parsed directly from the
            # TOML by a dedicated reader instead of the flat map.
            return None
        parts = [encode_scalar(item) for item in value]
        if any(p is None for p in parts):
            return None
        return json.dumps(parts, separators=(",", ":"))
    return encode_scalar(value)


def flatten(prefix: str, value: Any) -> Iterable[tuple[str, str]]:
    if isinstance(value, dict):
        for child_key, child_value in value.items():
            dotted = f"{prefix}.{child_key}" if prefix else child_key
            yield from flatten(dotted, child_value)
        return
    if not prefix:
        raise ValueError("top-level scalar manifests are not supported")
    if isinstance(value, list) and any(isinstance(item, (dict, list)) for item in value):
        if prefix == "extra_unit":
            return
        raise ValueError(f"nested arrays or inline tables are not supported at {prefix}")
    encoded = encode_value(value)
    if encoded is None:
        return
    yield prefix, encoded


def main() -> int:
    args = parse_args()
    manifest_path = Path(args.manifest)
    with manifest_path.open("rb") as handle:
        data = tomllib.load(handle)
    for key, value in flatten("", data):
        sys.stdout.write(f"{key}\t{value}\n")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        raise SystemExit(1)