"""Pydantic request / response schemas for the Identity Sync admin API.""" from __future__ import annotations from datetime import datetime from typing import Annotated, Any from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, StringConstraints ConfirmationToken = Annotated[str, StringConstraints(min_length=16, max_length=128)] class TargetCreateRequest(BaseModel): name: str = Field(min_length=1, max_length=128) slug: str = Field(min_length=1, max_length=64, pattern=r"^[a-z0-9][a-z0-9\-_]{0,62}[a-z0-9]$") description: str | None = Field(default=None, max_length=2000) kind: str = Field(min_length=1, max_length=32) integration_id: UUID | None = None enabled: bool = False on_offboard: str = Field(default="deactivate", pattern=r"^(deactivate|delete|keep)$") write_mode: str = Field(default="create_update", pattern=r"^(create_update|create_only|update_only)$") mapping_config: dict[str, Any] filter_config: dict[str, Any] schedule_config: dict[str, Any] runner_options: dict[str, Any] | None = None # Master on/off switch for this target's failure alerting. Default # on; when off the notifier sends nothing even if recipients exist. alert_enabled: bool = True # List of ``identity_persons.id`` (UUID strings) paged on run # failure. ``None`` / ``[]`` ⇒ nobody is notified (no admin fallback). alert_recipient_ids: list[str] | None = None class TargetUpdateRequest(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=128) description: str | None = Field(default=None, max_length=2000) integration_id: UUID | None = None integration_id_set: bool = Field( default=False, description=( "True when the request is intentionally setting " "integration_id (including to None to detach). Distinguishes " "PATCH-omitted vs PATCH-explicit-null." ), ) enabled: bool | None = None on_offboard: str | None = Field(default=None, pattern=r"^(deactivate|delete|keep)$") write_mode: str | None = Field(default=None, pattern=r"^(create_update|create_only|update_only)$") mapping_config: dict[str, Any] | None = None filter_config: dict[str, Any] | None = None schedule_config: dict[str, Any] | None = None runner_options: dict[str, Any] | None = None # On/off switch; omitted ⇒ unchanged. alert_enabled: bool | None = None # Tri-state via ``model_fields_set`` (PATCH-omitted vs explicit): # omitted ⇒ unchanged; ``[]`` ⇒ clear (nobody notified); ``[ids]`` ⇒ # replace. alert_recipient_ids: list[str] | None = None class AlertRecipientBrief(BaseModel): """Resolved alert recipient for display in the target list / drawer.""" id: UUID name: str | None = None email: str | None = None class TargetResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: UUID name: str slug: str description: str | None kind: str integration_id: UUID | None enabled: bool on_offboard: str write_mode: str = "create_update" mapping_config: dict[str, Any] filter_config: dict[str, Any] schedule_config: dict[str, Any] runner_options: dict[str, Any] | None last_run_at: datetime | None last_run_status: str | None last_run_summary: dict[str, Any] | None next_run_at: datetime | None alert_enabled: bool = True alert_recipient_ids: list[str] | None = None # Resolved recipient briefs (id/name/email) for direct UI render — # populated by the router, not a column. Empty when no recipients # are configured (the notifier then notifies nobody — no fallback). alert_recipients: list[AlertRecipientBrief] = [] created_at: datetime updated_at: datetime class StateResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: UUID target_id: UUID person_id: UUID person_email: str | None = None person_name: str | None = None person_employee_no: str | None = None person_employment_status: str | None = None external_id: str | None last_payload_hash: str | None last_status: str last_synced_at: datetime | None last_verified_at: datetime | None = None last_error: str | None retry_count: int updated_at: datetime class RunResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: UUID target_id: UUID started_at: datetime finished_at: datetime | None duration_ms: int | None status: str trigger_source: str persons_in_scope: int | None created_count: int | None updated_count: int | None skipped_count: int | None deactivated_count: int | None failed_count: int | None blocked_count: int | None error_message: str | None result_summary: dict[str, Any] | None class PreviewRequest(BaseModel): """Render a payload for one person against draft mapping/filter config without actually pushing. Used by the admin UI's "preview" button to debug mapping rules before saving. When ``mapping_config`` is None the server uses the target's stored mapping_config — that's the read-only "Mapping Preview" drawer mode where the admin picks a person to see live render. """ person_id: UUID mapping_config: dict[str, Any] | None = None runner_options: dict[str, Any] | None = None class RuleTraceEntry(BaseModel): source_template: str target_field: str transform: list[str] = [] value_map_hit: bool | None = None default: str | None = None raw: str after_transforms: str after_value_map: str | None = None rendered: str | None = None dropped: bool = False used_default: bool = False class PreviewResponse(BaseModel): person_id: UUID person_email: str rendered_payload: dict[str, Any] rendered_payload_hash: str context: dict[str, str] rule_trace: list[RuleTraceEntry] = [] warnings: list[str] = [] class TriggerResponse(BaseModel): target_id: UUID run_id: UUID status: str summary: dict[str, Any] | None = None class TriggerRunRequest(BaseModel): full_reconcile: bool = False force_reverify: bool = False confirmation_tokens: list[ConfirmationToken] = Field(default_factory=list, max_length=8) class DestructivePlanConfirmRequest(BaseModel): fingerprint: str = Field(min_length=64, max_length=64, pattern=r"^[0-9a-f]{64}$") class DestructivePlanResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: UUID target_id: UUID scope: str action: str fingerprint: str item_count: int item_snapshot: list[dict[str, Any]] status: str plan_expires_at: datetime confirmed_at: datetime | None confirmation_expires_at: datetime | None created_at: datetime class DestructivePlanConfirmResponse(BaseModel): plan_id: UUID confirmation_token: str confirmation_expires_at: datetime class PresetResponse(BaseModel): key: str label: str description: str kind: str mapping_rules: list[dict[str, Any]] filter_config: dict[str, Any] schedule_config: dict[str, Any] runner_options: dict[str, Any] requires_integration_kind: str | None requires_integration_provider: str | None write_mode: str = "create_update"