"""Pydantic schemas for the Leaver Blocked admin API.""" from __future__ import annotations from datetime import datetime from typing import Any from uuid import UUID from pydantic import BaseModel, Field, field_validator from app.domains.leaver_blocked.models import MIN_REASON_CHARS # ── Dashboard ─────────────────────────────────────────────────────── class LeaverBlockedSummaryOut(BaseModel): total_blocked: int sla_warn: int sla_urgent: int sla_breach: int pending_overrides: int approved_overrides_ready_to_execute: int today_completed: int class LeaverBlockedRowOut(BaseModel): process_id: UUID person_id: UUID person_name: str person_email: str department: str | None = None leaver_date: str | None = None blocked_for_hours: int | None = None sla_level: str pending_blocker_count: int pending_blocker_types: list[str] current_step: str | None = None has_active_override: bool = False override_status: str | None = None workflow_run_id: UUID | None = None last_step_at: str | None = None class LeaverBlockedListOut(BaseModel): items: list[LeaverBlockedRowOut] total: int # ── Detail ────────────────────────────────────────────────────────── class LeaverBlockedDetailOut(BaseModel): process: dict[str, Any] person: dict[str, Any] blockers: list[dict[str, Any]] leaver_block_audit: list[dict[str, Any]] overrides: list[dict[str, Any]] workflow_run: dict[str, Any] | None = None workflow_steps: list[dict[str, Any]] # ── Override ──────────────────────────────────────────────────────── class InitiateOverrideIn(BaseModel): person_id: UUID blocker_ids: list[UUID] primary_reason: str = Field(min_length=MIN_REASON_CHARS) secondary_signer_user_id: UUID @field_validator("blocker_ids") @classmethod def _non_empty(cls, v: list[UUID]) -> list[UUID]: if not v: raise ValueError("blocker_ids must contain at least one ID") return v class SecondaryDecisionIn(BaseModel): secondary_reason: str = Field(min_length=MIN_REASON_CHARS) class CancelOverrideIn(BaseModel): # Cancel reasons live in the same audit trail as initiate / approve / # reject reasons; enforce the same minimum length so cancellations # cannot be drive-by ("ok", "n/a") and lose attribution context. reason: str = Field(min_length=MIN_REASON_CHARS) class ForceBypassGraceIn(BaseModel): # The T+0 24h IT grace window is the platform's last manual-cleanup # buffer; skipping it requires a written reason ≥ MIN_REASON_CHARS so # forensics show *why* the bypass was warranted (HRIS-late-detection # case, IT has already drained the leaver's resources manually, etc.). # Field name kept parallel to ``CancelOverrideIn.reason`` for FE # consistency. reason: str = Field(min_length=MIN_REASON_CHARS) class OverrideOut(BaseModel): id: UUID process_id: UUID person_id: UUID # Human display names resolved server-side so the secondary-signer # review surface shows WHO is leaving / WHO signed instead of UUIDs. person_display: str | None = None status: str primary_signer_user_id: UUID primary_signer_display: str | None = None primary_signed_at: datetime primary_reason: str secondary_signer_user_id: UUID | None = None secondary_signer_display: str | None = None secondary_signed_at: datetime | None = None secondary_reason: str | None = None overridden_blocker_ids: list[str] overridden_blockers_snapshot: list[dict[str, Any]] executed_at: datetime | None = None rejected_at: datetime | None = None rejected_reason: str | None = None expires_at: datetime | None = None created_at: datetime class OverrideListOut(BaseModel): items: list[OverrideOut] total: int # ── Audit log query ───────────────────────────────────────────────── class LeaverBlockAuditOut(BaseModel): id: UUID person_id: UUID attempted_at: datetime attempted_by_user_id: UUID | None = None attempted_via: str target_status: str denied_reason: str pending_blocker_count: int pending_blockers: list[dict[str, Any]] detail: dict[str, Any] | None = None client_ip: str | None = None