"""Pydantic schemas for procurement-domain API responses + requests. Conventions mirror the rest of the codebase: * Read schemas (``*Read``) expose every field the UI may need, with derived display fields (e.g. ``supplier_name`` next to ``supplier_id``) so the FE doesn't need to denormalise on every render. * Create / update schemas accept the minimum fields the UI must provide; the service layer fills in derived values (e.g. ``supplier_code`` auto-generation, ``contract_number`` sequence). * Multi-currency fields carry ``currency`` + ``amount`` together and expose ``amount_cny`` (frozen at submit time via fx_rate service) as a separate field for reporting. """ from __future__ import annotations import uuid from collections.abc import Mapping from datetime import date, datetime from decimal import Decimal from typing import Any from pydantic import BaseModel, ConfigDict, Field from app.domains.procurement.models import ( CONTRACT_STATUSES, CONTRACT_TYPES, PAYMENT_STATUSES, PO_ORDER_TYPES, PO_STATUSES, PROCUREMENT_CATEGORY_CODES, RECEIPT_SEMANTICS, RECEIPT_STATUSES, SUPPLIER_STATUSES, ) _OrmCfg = ConfigDict(from_attributes=True) # ── Category ────────────────────────────────────────────────────────── class CategoryBase(BaseModel): code: str = Field(..., max_length=32) name_zh: str = Field(..., max_length=64) name_en: str = Field(..., max_length=64) parent_id: uuid.UUID | None = None description: str | None = Field(default=None, max_length=2000) default_owner_person_id: uuid.UUID | None = None default_cost_center: str | None = None default_legal_entity_id: uuid.UUID | None = None default_reminder_days: int | None = None receipt_semantics: str is_active: bool = True sort_order: int = 100 class CategoryCreate(CategoryBase): pass class CategoryUpdate(BaseModel): name_zh: str | None = None name_en: str | None = None parent_id: uuid.UUID | None = None description: str | None = Field(default=None, max_length=2000) default_owner_person_id: uuid.UUID | None = None default_cost_center: str | None = None default_legal_entity_id: uuid.UUID | None = None default_reminder_days: int | None = None receipt_semantics: str | None = None is_active: bool | None = None sort_order: int | None = None class CategoryRead(CategoryBase): model_config = _OrmCfg id: uuid.UUID created_at: datetime updated_at: datetime # Derived display field — service.py fills this from a person JOIN. default_owner_name: str | None = None # ── Supplier ────────────────────────────────────────────────────────── class SupplierContact(BaseModel): name: str = "" role: str | None = None email: str | None = None phone: str | None = None is_primary: bool = False class SupplierBase(BaseModel): # max_length on each string field matches its DB column so overlong # (paste-prone) input 422s at the schema layer instead of hitting a # MySQL/TiDB 1406 DataError -> 500 at insert time. name: str = Field(..., max_length=128) legal_name: str | None = Field(default=None, max_length=256) country: str | None = Field(default=None, max_length=64) tax_id: str | None = Field(default=None, max_length=64) registered_address: str | None = Field(default=None, max_length=512) website: str | None = Field(default=None, max_length=256) status: str = "active" default_currency: str | None = Field(default=None, max_length=8) payment_terms: str | None = Field(default=None, max_length=32) bank_name: str | None = Field(default=None, max_length=128) bank_account: str | None = Field(default=None, max_length=128) bank_account_name: str | None = Field(default=None, max_length=128) swift_code: str | None = Field(default=None, max_length=32) contacts: list[SupplierContact] = Field(default_factory=list) credit_rating: str | None = None last_evaluation_at: date | None = None supports_legal_entities: list[uuid.UUID] = Field(default_factory=list) notes: str | None = Field(default=None, max_length=2000) class SupplierCreate(SupplierBase): # Optional — auto-generated when omitted. supplier_code: str | None = Field(None, min_length=1, max_length=32) class SupplierUpdate(BaseModel): # max_length mirrors SupplierBase / the DB columns so an overlong # PATCH 422s at the schema layer instead of hitting a MySQL/TiDB # 1406 DataError -> 500 at update time. name: str | None = Field(default=None, max_length=128) legal_name: str | None = Field(default=None, max_length=256) country: str | None = Field(default=None, max_length=64) tax_id: str | None = Field(default=None, max_length=64) registered_address: str | None = Field(default=None, max_length=512) website: str | None = Field(default=None, max_length=256) status: str | None = None default_currency: str | None = Field(default=None, max_length=8) payment_terms: str | None = Field(default=None, max_length=32) bank_name: str | None = Field(default=None, max_length=128) bank_account: str | None = Field(default=None, max_length=128) bank_account_name: str | None = Field(default=None, max_length=128) swift_code: str | None = Field(default=None, max_length=32) contacts: list[SupplierContact] | None = None credit_rating: str | None = None last_evaluation_at: date | None = None supports_legal_entities: list[uuid.UUID] | None = None notes: str | None = Field(default=None, max_length=2000) class SupplierRead(BaseModel): """Read-only supplier view — fields with DB NULLs are normalised to empty containers via field_validators so the FE always sees consistent shapes.""" model_config = _OrmCfg id: uuid.UUID supplier_code: str name: str legal_name: str | None = None country: str | None = None tax_id: str | None = None registered_address: str | None = None website: str | None = None status: str = "active" default_currency: str | None = None payment_terms: str | None = None bank_name: str | None = None bank_account: str | None = None bank_account_name: str | None = None swift_code: str | None = None contacts: list[SupplierContact] = Field(default_factory=list) credit_rating: str | None = None last_evaluation_at: date | None = None supports_legal_entities: list[uuid.UUID] = Field(default_factory=list) notes: str | None = None created_at: datetime updated_at: datetime legacy_supplier_id: uuid.UUID | None = None # Derived counts so the supplier-list page can show "5 contracts / # 12 POs" inline without a second round-trip. contract_count: int = 0 active_contract_count: int = 0 po_count: int = 0 # True when the sensitive 银行/税号 fields above were masked for the # caller (non finance/admin). The FE renders a "reveal" affordance # that hits POST /suppliers/{id}/reveal (rate-limited + audited). sensitive_masked: bool = False @classmethod def model_validate(cls, obj, **kwargs): # type: ignore[override] # Normalise SQL NULL → empty containers BEFORE Pydantic # validates. ORM rows occasionally come back with # ``contacts=None`` / ``supports_legal_entities=None`` when the # column is set to NULL; the schema treats those as empty lists. # Never assign those defaults back to an ORM instance: the shared # request dependency commits successful GET transactions, so doing # so would turn response serialisation into an unaudited DB write. contacts = ( obj.get("contacts") if isinstance(obj, Mapping) else getattr(obj, "contacts", None) ) legal_entities = ( obj.get("supports_legal_entities") if isinstance(obj, Mapping) else getattr(obj, "supports_legal_entities", None) ) if contacts is not None and legal_entities is not None: return super().model_validate(obj, **kwargs) if isinstance(obj, Mapping): values = dict(obj) else: values = { field_name: getattr(obj, field_name) for field_name in cls.model_fields if hasattr(obj, field_name) } if contacts is None: values["contacts"] = [] if legal_entities is None: values["supports_legal_entities"] = [] return super().model_validate(values, **kwargs) # ── Contract ────────────────────────────────────────────────────────── class ContractAttachmentRead(BaseModel): model_config = _OrmCfg id: uuid.UUID contract_id: uuid.UUID file_name: str s3_key: str content_type: str | None size_bytes: int | None description: str | None uploaded_by_id: uuid.UUID | None uploaded_at: datetime download_url: str | None = None class ContractVersionRead(BaseModel): model_config = _OrmCfg id: uuid.UUID contract_id: uuid.UUID version_no: int changed_fields: dict[str, Any] change_summary: str | None changed_by_id: uuid.UUID | None changed_by_name: str | None = None changed_at: datetime class ContractBase(BaseModel): title: str = Field(..., max_length=256) supplier_id: uuid.UUID legal_entity_id: uuid.UUID | None = None category_id: uuid.UUID | None = None contract_type: str = "standalone" parent_contract_id: uuid.UUID | None = None total_amount: Decimal | None = None currency: str = "CNY" signed_date: date | None = None effective_date: date | None = None expiry_date: date | None = None auto_renew: bool = False renew_notice_days: int | None = None status: str = "draft" business_owner_id: uuid.UUID | None = None it_contact_id: uuid.UUID | None = None finance_contact_id: uuid.UUID | None = None procurement_owner_id: uuid.UUID | None = None description: str | None = Field(default=None, max_length=2000) notes: str | None = Field(default=None, max_length=4000) external_contract_no: str | None = Field(default=None, max_length=128) class ContractCreate(ContractBase): contract_number: str | None = Field(None, min_length=1, max_length=64) class ContractUpdate(BaseModel): title: str | None = None supplier_id: uuid.UUID | None = None legal_entity_id: uuid.UUID | None = None category_id: uuid.UUID | None = None contract_type: str | None = None parent_contract_id: uuid.UUID | None = None total_amount: Decimal | None = None currency: str | None = None signed_date: date | None = None effective_date: date | None = None expiry_date: date | None = None auto_renew: bool | None = None renew_notice_days: int | None = None status: str | None = None business_owner_id: uuid.UUID | None = None it_contact_id: uuid.UUID | None = None finance_contact_id: uuid.UUID | None = None procurement_owner_id: uuid.UUID | None = None description: str | None = Field(default=None, max_length=2000) notes: str | None = Field(default=None, max_length=4000) external_contract_no: str | None = Field(default=None, max_length=128) change_summary: str | None = None class ContractRead(ContractBase): model_config = _OrmCfg id: uuid.UUID contract_number: str created_at: datetime updated_at: datetime # Derived display fields. supplier_name: str | None = None legal_entity_name: str | None = None category_name: str | None = None business_owner_name: str | None = None it_contact_name: str | None = None finance_contact_name: str | None = None procurement_owner_name: str | None = None # Days until expiry (negative if already past). NULL if no expiry. days_until_expiry: int | None = None attachment_count: int = 0 po_count: int = 0 class ContractDetail(ContractRead): attachments: list[ContractAttachmentRead] = Field(default_factory=list) versions: list[ContractVersionRead] = Field(default_factory=list) # ── Purchase order + items ──────────────────────────────────────────── class OrderItemBase(BaseModel): # Auto-assigned by the service (next available line_no in order) # when the client omits it. line_no: int | None = Field(default=None, ge=1, le=2_147_483_647) description: str = Field(..., max_length=512) # Money / quantity fields are non-negative — a negative line_total # would flow into the PO total rollup and produce a negative or # understated total_amount, which lock_budget_for_order silently # skips (<= 0), so the PO looks committed while consuming no budget. quantity: Decimal = Field(Decimal(1), ge=0) # DB-column caps so an over-length value 422s at the schema instead of # raising MySQL/TiDB 1406 DataError → uncaught 500 (same hardening the # Supplier* schemas already have). unit: str | None = Field(default=None, max_length=32) unit_price: Decimal | None = Field(None, ge=0) line_total: Decimal | None = Field(None, ge=0) currency: str | None = Field(default=None, max_length=8) target_asset_id: uuid.UUID | None = None target_software_id: uuid.UUID | None = None target_license_plan_id: uuid.UUID | None = None target_descriptor: str | None = Field(default=None, max_length=256) notes: str | None = None class OrderItemCreate(OrderItemBase): pass class OrderItemUpdate(BaseModel): description: str | None = None quantity: Decimal | None = Field(None, ge=0) unit: str | None = None unit_price: Decimal | None = Field(None, ge=0) line_total: Decimal | None = Field(None, ge=0) currency: str | None = None target_asset_id: uuid.UUID | None = None target_software_id: uuid.UUID | None = None target_license_plan_id: uuid.UUID | None = None target_descriptor: str | None = None receipt_status: str | None = None received_at: datetime | None = None received_qty: Decimal | None = Field(None, ge=0) notes: str | None = None class OrderItemRead(OrderItemBase): model_config = _OrmCfg id: uuid.UUID order_id: uuid.UUID receipt_status: str received_at: datetime | None received_qty: Decimal | None created_at: datetime updated_at: datetime target_asset_tag: str | None = None target_software_name: str | None = None class OrderBase(BaseModel): order_type: str = "purchase" status: str = "draft" category_id: uuid.UUID | None = None supplier_id: uuid.UUID | None = None contract_id: uuid.UUID | None = None legal_entity_id: uuid.UUID | None = None software_catalog_id: uuid.UUID | None = None department: str | None = Field(default=None, max_length=128) cost_center: str | None = Field(default=None, max_length=64) currency: str = "CNY" total_amount: Decimal | None = Field(None, ge=0) business_owner_id: uuid.UUID | None = None it_contact_id: uuid.UUID | None = None finance_contact_id: uuid.UUID | None = None procurement_owner_id: uuid.UUID | None = None requested_at: datetime | None = None notes: str | None = None external_po_number: str | None = Field(default=None, max_length=128) external_link: str | None = Field(default=None, max_length=512) class OrderCreate(OrderBase): order_number: str | None = Field(default=None, min_length=1, max_length=64) applicant_id: uuid.UUID | None = None items: list[OrderItemCreate] = Field(default_factory=list) class OrderUpdate(BaseModel): order_type: str | None = None status: str | None = None category_id: uuid.UUID | None = None supplier_id: uuid.UUID | None = None contract_id: uuid.UUID | None = None legal_entity_id: uuid.UUID | None = None software_catalog_id: uuid.UUID | None = None department: str | None = None cost_center: str | None = None currency: str | None = None total_amount: Decimal | None = Field(None, ge=0) business_owner_id: uuid.UUID | None = None it_contact_id: uuid.UUID | None = None finance_contact_id: uuid.UUID | None = None procurement_owner_id: uuid.UUID | None = None receipt_status: str | None = None received_at: datetime | None = None receipt_evidence_url: str | None = None payment_status: str | None = None requested_at: datetime | None = None approved_at: datetime | None = None approved_by_id: uuid.UUID | None = None completed_at: datetime | None = None external_po_number: str | None = None external_link: str | None = None notes: str | None = None class OrderRead(OrderBase): model_config = _OrmCfg id: uuid.UUID order_number: str applicant_id: uuid.UUID receipt_status: str received_at: datetime | None receipt_evidence_url: str | None payment_status: str fx_rate_to_cny: Decimal | None total_cny: Decimal | None budget_id: uuid.UUID | None budget_locked_at: datetime | None budget_released_at: datetime | None approved_at: datetime | None approved_by_id: uuid.UUID | None completed_at: datetime | None legacy_order_id: uuid.UUID | None legacy_unresolved_fields: dict[str, Any] | None created_at: datetime updated_at: datetime # Derived display. applicant_name: str | None = None supplier_name: str | None = None contract_number: str | None = None category_name: str | None = None business_owner_name: str | None = None procurement_owner_name: str | None = None item_count: int = 0 class OrderDetail(OrderRead): items: list[OrderItemRead] = Field(default_factory=list) # ── Payments ────────────────────────────────────────────────────────── class PaymentBase(BaseModel): order_id: uuid.UUID | None = None contract_id: uuid.UUID | None = None supplier_id: uuid.UUID | None = None legal_entity_id: uuid.UUID | None = None software_catalog_id: uuid.UUID | None = None payment_frequency: str | None = None payment_method: str | None = None requester_id: uuid.UUID | None = None finance_contact_id: uuid.UUID | None = None amount: Decimal | None = Field(None, ge=0) currency: str = Field(default="CNY", max_length=32) status: str = "pending" invoice_number: str | None = Field(default=None, max_length=128) invoice_received_at: date | None = None paid_at: date | None = None external_voucher_no: str | None = Field(default=None, max_length=128) notes: str | None = None class PaymentCreate(PaymentBase): payment_number: str | None = Field(default=None, min_length=1, max_length=64) class PaymentUpdate(BaseModel): order_id: uuid.UUID | None = None contract_id: uuid.UUID | None = None supplier_id: uuid.UUID | None = None legal_entity_id: uuid.UUID | None = None software_catalog_id: uuid.UUID | None = None payment_frequency: str | None = None payment_method: str | None = None requester_id: uuid.UUID | None = None finance_contact_id: uuid.UUID | None = None amount: Decimal | None = Field(None, ge=0) currency: str | None = Field(default=None, max_length=32) status: str | None = None invoice_number: str | None = Field(default=None, max_length=128) invoice_received_at: date | None = None paid_at: date | None = None external_voucher_no: str | None = Field(default=None, max_length=128) notes: str | None = None class PaymentRead(PaymentBase): model_config = _OrmCfg id: uuid.UUID payment_number: str fx_rate_to_cny: Decimal | None amount_cny: Decimal | None legacy_payment_id: uuid.UUID | None legacy_unresolved_fields: dict[str, Any] | None created_at: datetime updated_at: datetime supplier_name: str | None = None order_number: str | None = None legal_entity_name: str | None = None software_name: str | None = None # ── Software procurement meta ───────────────────────────────────────── class SoftwareProcurementMetaBase(BaseModel): current_contract_id: uuid.UUID | None = None current_supplier_id: uuid.UUID | None = None expected_renewal_at: date | None = None current_period_amount: Decimal | None = None current_period_currency: str | None = None business_owner_id: uuid.UUID | None = None it_contact_id: uuid.UUID | None = None finance_contact_id: uuid.UUID | None = None procurement_owner_id: uuid.UUID | None = None notes: str | None = None class SoftwareProcurementMetaUpdate(SoftwareProcurementMetaBase): pass class SoftwareProcurementMetaRead(SoftwareProcurementMetaBase): model_config = _OrmCfg id: uuid.UUID software_id: uuid.UUID created_at: datetime updated_at: datetime current_contract_number: str | None = None current_contract_title: str | None = None current_supplier_name: str | None = None business_owner_name: str | None = None # ── Validators (sets exposed to FastAPI; used by service.py guards) ─── VALID_CATEGORY_CODES = PROCUREMENT_CATEGORY_CODES VALID_RECEIPT_SEMANTICS = RECEIPT_SEMANTICS VALID_SUPPLIER_STATUSES = SUPPLIER_STATUSES VALID_CONTRACT_STATUSES = CONTRACT_STATUSES VALID_CONTRACT_TYPES = CONTRACT_TYPES VALID_PO_STATUSES = PO_STATUSES VALID_PO_ORDER_TYPES = PO_ORDER_TYPES VALID_RECEIPT_STATUSES = RECEIPT_STATUSES VALID_PAYMENT_STATUSES = PAYMENT_STATUSES