90 lines
3.9 KiB
Python
90 lines
3.9 KiB
Python
"""Governed API facade for branch-bound repair leases.
|
|
|
|
The facade is transport-neutral so the permanent control plane can expose it
|
|
through its existing authenticated loopback/API seam without creating a second
|
|
Git writer or a global credential registry.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Mapping
|
|
|
|
from branch_repair_lease import LeaseError, RepairLease, RepairLeaseManager, sanitized_lease_view
|
|
|
|
|
|
class BranchRepairLeaseAPI:
|
|
"""Validate operation envelopes and return sanitized lease metadata."""
|
|
|
|
def __init__(self, manager: RepairLeaseManager) -> None:
|
|
self.manager = manager
|
|
|
|
@staticmethod
|
|
def _required(request: Mapping[str, Any], name: str) -> Any:
|
|
value = request.get(name)
|
|
if value is None or value == "":
|
|
raise LeaseError(f"missing required field: {name}")
|
|
return value
|
|
|
|
@staticmethod
|
|
def _view(lease: RepairLease) -> dict[str, Any]:
|
|
return {"status": "ok", "lease": sanitized_lease_view(lease)}
|
|
|
|
def handle(self, request: Mapping[str, Any]) -> dict[str, Any]:
|
|
try:
|
|
operation = self._required(request, "operation")
|
|
owner = str(self._required(request, "owner_identity"))
|
|
if operation == "issue":
|
|
lease = self.manager.issue(
|
|
task_id=str(self._required(request, "task_id")),
|
|
owner_identity=owner,
|
|
branch=str(self._required(request, "branch")),
|
|
description=str(self._required(request, "description")),
|
|
allowed_paths=tuple(self._required(request, "allowed_paths")),
|
|
ttl_seconds=int(request.get("ttl_seconds", 3600)),
|
|
)
|
|
elif operation == "attach_existing_branch":
|
|
lease = self.manager.attach_existing_branch(
|
|
str(self._required(request, "lease_id")),
|
|
owner_identity=owner,
|
|
expected_head_oid=str(self._required(request, "expected_head_oid")),
|
|
)
|
|
elif operation == "authorize_commit":
|
|
lease = self.manager.authorize_commit(
|
|
str(self._required(request, "lease_id")),
|
|
owner_identity=owner,
|
|
expected_head_oid=str(self._required(request, "expected_head_oid")),
|
|
paths=tuple(self._required(request, "paths")),
|
|
content=request.get("content"),
|
|
)
|
|
elif operation == "record_commit":
|
|
lease = self.manager.record_commit(
|
|
str(self._required(request, "lease_id")),
|
|
owner_identity=owner,
|
|
expected_head_oid=str(self._required(request, "expected_head_oid")),
|
|
new_head_oid=str(self._required(request, "new_head_oid")),
|
|
paths=tuple(self._required(request, "paths")),
|
|
)
|
|
elif operation == "recover":
|
|
lease = self.manager.recover(
|
|
str(self._required(request, "lease_id")), owner_identity=owner
|
|
)
|
|
elif operation == "renew":
|
|
lease = self.manager.renew(
|
|
str(self._required(request, "lease_id")),
|
|
owner_identity=owner,
|
|
ttl_seconds=int(self._required(request, "ttl_seconds")),
|
|
)
|
|
elif operation == "complete":
|
|
lease = self.manager.complete(
|
|
str(self._required(request, "lease_id")), owner_identity=owner
|
|
)
|
|
elif operation == "revoke":
|
|
lease = self.manager.revoke(
|
|
str(self._required(request, "lease_id")), owner_identity=owner
|
|
)
|
|
else:
|
|
raise LeaseError("unsupported lease operation")
|
|
return self._view(lease)
|
|
except (LeaseError, TypeError, ValueError):
|
|
return {"status": "blocked", "error": "lease_request_rejected"}
|