INNER CODE UNIT · Python
exact_fields
yaojingang/GEOFlow · deployment/generate-upgrade-plan.py:23
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")
sources = plan["allowed_sources"]
if type(sources) is not list or any(type(source) is not int or source < 1 or source > 2**63 - 1 for source in sources) or len(sources) != len(set(sources)):
raise ValueError("Upgrade source sequences must be unique positive application integers")
compatible = plan["compatibility"]
exact_fields(compatible, {"schema", "queue", "cache", "storage"}, "Compatibility")
if any(type(value) is not bool for value in compatible.values()):
raise ValueError("Compatibility fields must be booleans")
if type(plan["migrations"]) is not list or type(plan["steps"]) is not list:
raise ValueError("Migrations and steps must be arrays")