74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""AIOS V2R1 Tenant Cell lifecycle CLI primitives."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from .tenant_context import ValidationError, require_ident
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CellRecord:
|
|
tenant_id: str
|
|
cell_id: str
|
|
version: str
|
|
state: str
|
|
config_digest: str
|
|
created_at: str
|
|
|
|
|
|
class CellManager:
|
|
def __init__(self, registry: str | Path) -> None:
|
|
self.registry = Path(registry)
|
|
self.registry.parent.mkdir(parents=True, exist_ok=True)
|
|
if not self.registry.exists():
|
|
self.registry.write_text("{\"cells\":{}}\n", encoding="utf-8")
|
|
|
|
def _load(self) -> dict:
|
|
return json.loads(self.registry.read_text(encoding="utf-8"))
|
|
|
|
def _save(self, value: dict) -> None:
|
|
self.registry.write_text(
|
|
json.dumps(value, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def create(self, tenant_id: str, cell_id: str, version: str, digest: str) -> CellRecord:
|
|
tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
|
|
cell_id = require_ident("cell_id", cell_id)
|
|
data = self._load()
|
|
if tenant_id in data["cells"]:
|
|
raise ValidationError("tenant already has a cell")
|
|
record = CellRecord(
|
|
tenant_id,
|
|
cell_id,
|
|
version,
|
|
"created",
|
|
digest,
|
|
datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
data["cells"][tenant_id] = record.__dict__
|
|
self._save(data)
|
|
return record
|
|
|
|
def status(self, tenant_id: str) -> CellRecord:
|
|
data = self._load()
|
|
item = data["cells"].get(tenant_id)
|
|
if not item:
|
|
raise ValidationError("unknown tenant")
|
|
return CellRecord(**item)
|
|
|
|
def transition(self, tenant_id: str, state: str) -> CellRecord:
|
|
allowed = {"created", "starting", "healthy", "upgrading", "rollback", "failed"}
|
|
if state not in allowed:
|
|
raise ValidationError("invalid cell state")
|
|
data = self._load()
|
|
item = data["cells"].get(tenant_id)
|
|
if not item:
|
|
raise ValidationError("unknown tenant")
|
|
item["state"] = state
|
|
self._save(data)
|
|
return CellRecord(**item)
|