INNER CODE UNIT · Python

unique_object

yaojingang/GEOFlow · deployment/generate-upgrade-plan.py:14

def unique_object(pairs: list[tuple]) -> dict:
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"Duplicate upgrade plan field: {key}")
        result[key] = value
    return result


def exact_fields(value, keys: set[str], label: str) -> None:
    if type(value) is not dict or set(value) != keys:
        raise ValueError(f"{label} must contain exactly its required fields")


def validate_plan(plan: dict) -> None:
    exact_fields(plan, {"schema_version", "strategy", "allowed_sources", "migrations", "compatibility", "steps"}, "Upgrade plan")
    if type(plan["schema_version"]) is not int or plan["schema_version"] != 1 or plan["strategy"] not in ("maintenance", "online"):
        raise ValueError("Upgrade plan schema or strategy is unsupported")

View source record →