68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
"""Runtime deployment-state provider for governed health contracts."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
_OID = re.compile(r"^[0-9a-f]{40}$")
|
|
_SHA = re.compile(r"^[0-9a-f]{64}$")
|
|
_STATES = {"STAGED", "ACTIVE", "ROLLED_BACK"}
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeDeploymentState:
|
|
version: str
|
|
commit_oid: str
|
|
artifact_sha256: str
|
|
deployment_state: str
|
|
|
|
@property
|
|
def production_deployed(self) -> bool:
|
|
return self.deployment_state == "ACTIVE"
|
|
|
|
def health_payload(self) -> dict[str, Any]:
|
|
return {
|
|
"status": "ok",
|
|
"version": self.version,
|
|
"commit_oid": self.commit_oid,
|
|
"artifact_sha256": self.artifact_sha256,
|
|
"deployment_state": self.deployment_state,
|
|
"production_deployed": self.production_deployed,
|
|
"health_timestamp": dt.datetime.now(dt.timezone.utc).isoformat(),
|
|
}
|
|
|
|
@classmethod
|
|
def from_environment(cls, environ: Mapping[str, str] | None = None) -> "RuntimeDeploymentState":
|
|
env = dict(environ or os.environ)
|
|
values: dict[str, Any] = {}
|
|
state_file = env.get("DEPLOYMENT_STATE_FILE", "").strip()
|
|
if state_file:
|
|
path = Path(state_file)
|
|
if path.is_file() and not path.is_symlink():
|
|
loaded = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(loaded, dict):
|
|
raise ValueError("deployment_state_file_object_required")
|
|
values.update(loaded)
|
|
file_deployment_state = values.get("deployment_state")
|
|
values.update({k: v for k, v in {
|
|
"version": env.get("RELEASE_VERSION", values.get("version", "")),
|
|
"commit_oid": env.get("TARGET_OID", values.get("commit_oid", "")),
|
|
"artifact_sha256": env.get("ARTIFACT_SHA256", values.get("artifact_sha256", "")),
|
|
"deployment_state": env.get("DEPLOYMENT_STATE", values.get("deployment_state", "")),
|
|
}.items() if v != ""})
|
|
if file_deployment_state not in (None, ""):
|
|
values["deployment_state"] = file_deployment_state
|
|
version = str(values.get("version", ""))
|
|
commit_oid = str(values.get("commit_oid", ""))
|
|
artifact_sha256 = str(values.get("artifact_sha256", ""))
|
|
deployment_state = str(values.get("deployment_state", ""))
|
|
if not version or not _OID.fullmatch(commit_oid) or not _SHA.fullmatch(artifact_sha256):
|
|
raise ValueError("deployment_binding_invalid")
|
|
if deployment_state not in _STATES:
|
|
raise ValueError("deployment_state_invalid")
|
|
return cls(version, commit_oid, artifact_sha256, deployment_state)
|