"""Pydantic schemas for the provisioning domain.""" from __future__ import annotations from datetime import datetime from typing import Any from uuid import UUID from pydantic import BaseModel, Field # ─── Connector schemas ─── class SoftwareConnectorCreate(BaseModel): """Create a software → integration binding. Credentials live on the ``integration_id`` row (managed under /admin/integrations). ``binding_config`` holds only this software's per-binding params (group_email, operation, ...) as defined by the integration driver's ``binding_schema``. Offboarding behaviour is configured on the software itself, not here. """ software_id: UUID integration_id: UUID binding_config: dict[str, Any] = Field(default_factory=dict) is_active: bool = True class SoftwareConnectorUpdate(BaseModel): integration_id: UUID | None = None binding_config: dict[str, Any] | None = None is_active: bool | None = None class SoftwareConnectorResponse(BaseModel): id: UUID software_id: UUID software_name: str | None = None integration_id: UUID integration_name: str | None = None integration_provider: str | None = None binding_config: dict[str, Any] is_active: bool last_success_at: datetime | None last_error_at: datetime | None last_error_message: str | None created_at: datetime updated_at: datetime # ─── Task schemas ─── class ProvisioningTaskResponse(BaseModel): id: UUID user_id: UUID user_name: str | None = None user_email: str | None = None user_employee_no: str | None = None user_employment_status: str | None = None software_id: UUID | None software_name: str | None = None # Display label for the thing being granted / revoked. Mirrors # ``software_name`` for seat / permission tasks (where the catalog # row IS the target). For ``target_type='account'`` tasks — # full-access revokes whose ``target_id`` is an integration id # (Jira / Atlassian "remove from all groups", created by # offboarding ``_apply_account_policy``) — ``software_id`` is # legitimately NULL and this falls back to the integration name so # the Software Cancellations admin overview doesn't render an # opaque "—" for every Atlassian leaver row. Added 2026-06-02 # after owner-reported "为什么软件那列是空的" on chujie / # zequan.yang / yudingyi offboarding revokes. target_label: str | None = None action: str target_type: str target_id: UUID | None source: str execution_mode: str status: str assignee_id: UUID | None assignee_name: str | None = None assignee_employee_no: str | None = None assignee_employment_status: str | None = None external_ref: str | None error_message: str | None approval_request_id: UUID | None context_data: dict[str, Any] | None attempt_count: int not_before: datetime | None = None started_at: datetime | None completed_at: datetime | None created_at: datetime updated_at: datetime class ProvisioningTaskCreate(BaseModel): """Used internally by orchestration code, not exposed as admin endpoint.""" user_id: UUID software_id: UUID | None = None action: str target_type: str target_id: UUID | None = None source: str execution_mode: str | None = None # auto-resolved if None assignee_id: UUID | None = None approval_request_id: UUID | None = None context_data: dict[str, Any] | None = None # Defer dispatcher execution until this UTC moment. NULL = run as # soon as the dispatcher picks it up. Offboarding sets this to the # end of the leaver's last working day; onboarding (deferred new- # hire provisioning) sets it to hire_date 00:00 local. See # migration 221 for the column rationale. not_before: datetime | None = None class TaskCompleteRequest(BaseModel): """Admin marks a manual task as completed.""" note: str | None = Field(None, max_length=2000) external_ref: str | None = Field(None, max_length=2000) class TaskFailRequest(BaseModel): """Admin marks a task as failed / unable to complete.""" reason: str = Field(min_length=1, max_length=2000) class TaskRetryRequest(BaseModel): """Admin retries a failed/cancelled task.""" reset_assignee: bool = False # ─── Policy schemas ─── class ProvisionPolicyCreate(BaseModel): # Match keys — any subset may be NULL to mean "any". All three NULL # matches every employee. department_name: str | None = None match_job_title: str | None = None match_location: str | None = None target_type: str # 'seat' | 'permission' target_id: UUID is_active: bool = True notes: str | None = Field(None, max_length=2000) class ProvisionPolicyUpdate(BaseModel): department_name: str | None = None match_job_title: str | None = None match_location: str | None = None is_active: bool | None = None notes: str | None = Field(None, max_length=2000) class ProvisionPolicyResponse(BaseModel): id: UUID department_name: str | None match_job_title: str | None = None match_location: str | None = None target_type: str target_id: UUID target_label: str | None = None # software name + plan name for display software_id: UUID | None = None software_name: str | None = None is_active: bool notes: str | None created_by: UUID | None created_by_name: str | None = None created_at: datetime updated_at: datetime # ─── Onboarding orchestration ─── class OnboardingPreviewItem(BaseModel): """One item that would be provisioned for a person.""" policy_id: UUID target_type: str target_id: UUID target_label: str software_id: UUID | None = None software_name: str | None = None already_provisioned: bool = False # skip reason if True class OnboardingPreview(BaseModel): person_id: UUID person_name: str | None person_email: str person_employee_no: str | None = None person_employment_status: str | None = None department: str | None items: list[OnboardingPreviewItem] # ─── Bulk task execution ─── class BulkExecuteRequest(BaseModel): """Execute a set of pending tasks, optionally narrowed by source.""" source: str | None = None # e.g. 'onboarding', 'offboarding' user_id: UUID | None = None task_ids: list[UUID] | None = None # explicit whitelist; takes priority over other filters class BulkExecuteResponse(BaseModel): executed: int completed: int failed: int notified: int # manual_notify ones that became notified skipped: int errored: int = 0 # tasks whose execute_task raised (plan deleted, deadlock, …) transient_retried: int = 0 # transient connector errors re-queued with backoff (still live) # ─── Role change preview ─── class ReconcileDiffItem(BaseModel): """One seat or permission that the policy says should be granted or revoked relative to the person's current holdings.""" action: str # 'grant' | 'revoke' target_type: str # 'seat' | 'permission' target_id: UUID target_label: str software_id: UUID | None = None software_name: str | None = None reason: str # 'policy_match' | 'policy_unmatch' class RoleChangePreview(BaseModel): person_id: UUID person_name: str | None person_email: str person_employee_no: str | None = None person_employment_status: str | None = None current_department: str | None policies_matched: int items_to_add: list[ReconcileDiffItem] items_to_remove: list[ReconcileDiffItem] class ReconcileApplyRequest(BaseModel): """Admin applies the diff (subset or all) as provisioning tasks.""" add_target_ids: list[UUID] = [] remove_target_ids: list[UUID] = [] # ─── Billing ─── class BillingLineItem(BaseModel): seat_id: UUID user_id: UUID user_name: str | None department: str | None software_id: UUID software_name: str plan_id: UUID plan_name: str price: float billing_cycle: str # ISO 4217 code copied from the plan. ``"CNY"`` is the historical # default for plans created before the column existed. currency: str = "CNY" monthly_cost: float # normalised to monthly for aggregation class BillingAggregate(BaseModel): # Flat total across all lines. Populated only when every line is in # the same currency; ``0.0`` together with ``currency_warning=True`` # signals a multi-currency tenant — callers must use ``by_currency``. total_monthly_cost: float # Per-currency monthly subtotal. Always present. Single-currency # tenants get a one-key map mirroring ``total_monthly_cost``. by_currency: dict[str, float] = {} # Convenience flag — ``len(by_currency) > 1``. Surfaced separately # so the UI doesn't have to care about dict cardinality. currency_warning: bool = False total_currency_count: int = 0 seat_count: int by_software: list[dict] # [{software_id, software_name, monthly_cost, seat_count, currency}] by_department: list[dict] # [{department, monthly_cost, seat_count}] lines: list[BillingLineItem]