80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
"""HTTP handler and display data for architecture owner gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from .architecture_owner_gate import ArchitectureOwnerGate, ArchitectureOwnerApprovalRequest
|
|
|
|
|
|
gate = ArchitectureOwnerGate()
|
|
|
|
# In-memory store for pending approval requests
|
|
_pending_requests: dict[str, dict[str, Any]] = {}
|
|
_decisions: dict[str, dict[str, Any]] = {}
|
|
|
|
|
|
def create_owner_approval_request(change: dict[str, Any], architect_decision: dict[str, Any]) -> dict[str, Any]:
|
|
req = gate.create_request(change, architect_decision)
|
|
d = asdict(req)
|
|
_pending_requests[req.change_id] = d
|
|
return d
|
|
|
|
|
|
def approve_request(change_id: str, decided_by: str = "owner", reason: str = "") -> dict[str, Any]:
|
|
req = _pending_requests.get(change_id)
|
|
if not req:
|
|
return {"error": f"no pending request for change_id={change_id}"}
|
|
decision = asdict(
|
|
ArchitectureOwnerApprovalRequest.__bases__[1].__new__(type("_ArchitectureOwnerDecision", (), {})) # noqa
|
|
)
|
|
try:
|
|
from .architecture_owner_gate import ArchitectureOwnerDecision
|
|
dec = ArchitectureOwnerDecision(
|
|
request_id=req["request_id"],
|
|
change_id=change_id,
|
|
decision="approved",
|
|
decided_by=decided_by,
|
|
reason=reason,
|
|
)
|
|
d = asdict(dec)
|
|
_decisions[change_id] = d
|
|
# remove from pending
|
|
_pending_requests.pop(change_id, None)
|
|
return {"allowed": True, "decision": d}
|
|
except Exception as exc:
|
|
return {"error": str(exc)}
|
|
|
|
|
|
def reject_request(change_id: str, decided_by: str = "owner", reason: str = "") -> dict[str, Any]:
|
|
from .architecture_owner_gate import ArchitectureOwnerDecision
|
|
req = _pending_requests.get(change_id)
|
|
if not req:
|
|
return {"error": f"no pending request for change_id={change_id}"}
|
|
try:
|
|
dec = ArchitectureOwnerDecision(
|
|
request_id=req["request_id"],
|
|
change_id=change_id,
|
|
decision="rejected",
|
|
decided_by=decided_by,
|
|
reason=reason,
|
|
)
|
|
d = asdict(dec)
|
|
_decisions[change_id] = d
|
|
_pending_requests.pop(change_id, None)
|
|
return {"allowed": False, "decision": d}
|
|
except Exception as exc:
|
|
return {"error": str(exc)}
|
|
|
|
|
|
def get_owner_gate_overview() -> dict[str, Any]:
|
|
return {
|
|
"pending_requests": list(_pending_requests.values()),
|
|
"decisions": list(_decisions.values()),
|
|
"pending_count": len(_pending_requests),
|
|
"approved_count": sum(1 for d in _decisions.values() if d.get("decision") == "approved"),
|
|
"rejected_count": sum(1 for d in _decisions.values() if d.get("decision") == "rejected"),
|
|
}
|