Implement generic approval runtime

src/govoplan_approvals/backend/schemas.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from govoplan_core.core.approvals import (
+    ApprovalActorSelector,
+    ApprovalDecisionCommand,
+    ApprovalRequestCreateCommand,
+    ApprovalStepDefinition,
+    ApprovalTemplateCreateCommand,
+)
+
+
+class ApprovalSelectorInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    kind: Literal["account", "group", "role", "function_assignment", "any_account"]
+    value: str = Field(min_length=1, max_length=255)
+    label: str | None = Field(default=None, max_length=255)
+
+
+class ApprovalStepInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    key: str = Field(min_length=1, max_length=120)
+    label: str = Field(min_length=1, max_length=255)
+    selectors: list[ApprovalSelectorInput] = Field(min_length=1, max_length=500)
+    required_approvals: int = Field(default=1, ge=1, le=500)
+    rejection_policy: Literal["fail_fast", "collect"] = "fail_fast"
+    due_at: datetime | None = None
+    signature_required: bool = False
+    forbidden_evidence_roles: list[str] = Field(default_factory=list, max_length=100)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    def to_definition(self) -> ApprovalStepDefinition:
+        return ApprovalStepDefinition(
+            key=self.key,
+            label=self.label,
+            selectors=tuple(
+                ApprovalActorSelector(item.kind, item.value, item.label)
+                for item in self.selectors
+            ),
+            required_approvals=self.required_approvals,
+            rejection_policy=self.rejection_policy,
+            due_at=self.due_at,
+            signature_required=self.signature_required,
+            forbidden_evidence_roles=tuple(self.forbidden_evidence_roles),
+            metadata=self.metadata,
+        )
+
+
+class ApprovalRequestInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    title: str = Field(min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=10_000)
+    subject_module: str = Field(min_length=1, max_length=120)
+    subject_type: str = Field(min_length=1, max_length=120)
+    subject_id: str = Field(min_length=1, max_length=255)
+    subject_version: str | None = Field(default=None, max_length=120)
+    subject_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
+    steps: list[ApprovalStepInput] = Field(default_factory=list, max_length=100)
+    separation_of_duties: bool = True
+    unique_actors_across_steps: bool = False
+    expires_at: datetime | None = None
+    policy_refs: list[str] = Field(default_factory=list, max_length=500)
+    evidence_actors: dict[str, list[str]] = Field(default_factory=dict)
+    template_id: str | None = Field(default=None, max_length=36)
+    template_revision: int | None = Field(default=None, ge=1)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    def to_command(self) -> ApprovalRequestCreateCommand:
+        return ApprovalRequestCreateCommand(
+            title=self.title,
+            description=self.description,
+            subject_module=self.subject_module,
+            subject_type=self.subject_type,
+            subject_id=self.subject_id,
+            subject_version=self.subject_version,
+            subject_digest=self.subject_digest,
+            steps=tuple(step.to_definition() for step in self.steps),
+            separation_of_duties=self.separation_of_duties,
+            unique_actors_across_steps=self.unique_actors_across_steps,
+            expires_at=self.expires_at,
+            policy_refs=tuple(self.policy_refs),
+            evidence_actors={
+                key: tuple(values) for key, values in self.evidence_actors.items()
+            },
+            template_id=self.template_id,
+            template_revision=self.template_revision,
+            metadata=self.metadata,
+        )
+
+
+class ApprovalTemplateInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    key: str = Field(min_length=1, max_length=120)
+    title: str = Field(min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=10_000)
+    steps: list[ApprovalStepInput] = Field(min_length=1, max_length=100)
+    separation_of_duties: bool = True
+    unique_actors_across_steps: bool = False
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    def to_command(self) -> ApprovalTemplateCreateCommand:
+        return ApprovalTemplateCreateCommand(
+            key=self.key,
+            title=self.title,
+            description=self.description,
+            steps=tuple(step.to_definition() for step in self.steps),
+            separation_of_duties=self.separation_of_duties,
+            unique_actors_across_steps=self.unique_actors_across_steps,
+            metadata=self.metadata,
+        )
+
+
+class ApprovalTemplateCreateRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    template: ApprovalTemplateInput
+    idempotency_key: str = Field(min_length=1, max_length=160)
+
+
+class ApprovalTemplateReviseRequest(ApprovalTemplateCreateRequest):
+    expected_revision: int = Field(ge=1)
+
+
+class ApprovalCreateRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    request: ApprovalRequestInput
+    idempotency_key: str = Field(min_length=1, max_length=160)
+
+
+class ApprovalDecisionInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    outcome: Literal["approved", "rejected"]
+    reason: str = Field(min_length=1, max_length=4000)
+    expected_revision: int = Field(ge=1)
+    idempotency_key: str = Field(min_length=1, max_length=160)
+    delegated_for_account_id: str | None = Field(default=None, max_length=255)
+    signature_ref: dict[str, Any] | None = None
+
+    def to_command(self) -> ApprovalDecisionCommand:
+        return ApprovalDecisionCommand(
+            outcome=self.outcome,
+            reason=self.reason,
+            expected_revision=self.expected_revision,
+            idempotency_key=self.idempotency_key,
+            delegated_for_account_id=self.delegated_for_account_id,
+            signature_ref=self.signature_ref,
+        )
+
+
+class ApprovalTransitionInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    expected_revision: int = Field(ge=1)
+    idempotency_key: str = Field(min_length=1, max_length=160)
+
+
+class ApprovalListResponse(BaseModel):
+    requests: list[dict[str, Any]]
+
+
+__all__ = [
+    "ApprovalCreateRequest",
+    "ApprovalDecisionInput",
+    "ApprovalListResponse",
+    "ApprovalTransitionInput",
+    "ApprovalTemplateCreateRequest",
+    "ApprovalTemplateReviseRequest",
+]