""" Pydantic schemas for the Subscriptions domain. """ from __future__ import annotations from datetime import datetime from enum import Enum from typing import List, Literal, Optional from fastapi import Query from pydantic import BaseModel, ConfigDict, Field # ─── Status text mapping ───────────────────────────────────────────────────── class SubscriptionStatusText(str, Enum): INITIALIZING = "Initializing" ACTIVE = "Active" PAST_DUE = "Past Due" PAUSED = "Paused" CANCELLED = "Cancelled" COMPLETED = "Completed" DUNNING_EXHAUSTED = "Dunning Exhausted" _STATUS_INT_TO_TEXT = { 100: SubscriptionStatusText.INITIALIZING, 200: SubscriptionStatusText.ACTIVE, 201: SubscriptionStatusText.PAST_DUE, 202: SubscriptionStatusText.PAUSED, 300: SubscriptionStatusText.CANCELLED, 301: SubscriptionStatusText.COMPLETED, 400: SubscriptionStatusText.DUNNING_EXHAUSTED, } def status_to_text(status: int) -> str: return _STATUS_INT_TO_TEXT.get(status, SubscriptionStatusText.INITIALIZING).value # ─── Activity ───────────────────────────────────────────────────────────────── class SubscriptionActivityResponse(BaseModel): model_config = ConfigDict(from_attributes=True, populate_by_name=True) id: int activity_type: str description: Optional[str] = None actor_type: Optional[str] = None actor_id: Optional[int] = None created_at: datetime metadata: Optional[dict] = Field(None, alias="metadata_") # ─── List item ─────────────────────────────────────────────────────────────── class SubscriptionListItemResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: int subscription_id: str subscription_literal: str name: Optional[str] = None status: int status_text: str customer_id: Optional[int] = None customer_name: Optional[str] = None amount: float interval_label: str next_billing_date: Optional[datetime] = None total_paid: float invoices_paid: int invoices_generated: int created_at: datetime # ─── Detail ─────────────────────────────────────────────────────────────────── class SubscriptionDetailResponse(SubscriptionListItemResponse): total_billed: float invoices_overdue: int = 0 invoices_pending: int = 0 invoices_cancelled: int = 0 past_due_since: Optional[datetime] = None authorization_type: str = "" payment_method_summary: Optional[str] = None interval: str = "" interval_value: int = 1 end_type: str = "" end_date: Optional[datetime] = None pay_until_count: Optional[int] = None start_date: Optional[datetime] = None dunning_retry_count: int = 0 dunning_next_retry_at: Optional[datetime] = None dunning_exhausted_at: Optional[datetime] = None recent_activities: List[SubscriptionActivityResponse] = [] # ─── Invoice timeline ───────────────────────────────────────────────────────── class SubscriptionInvoiceItem(BaseModel): sequence: int amount: float due_date: Optional[datetime] = None payment_date: Optional[datetime] = None status: str # PAID | PENDING | OVERDUE | CANCELLED | FUTURE | PROCESSING | FAILED is_generated: bool invoice_id: Optional[str] = None invoice_literal: Optional[str] = None can_edit: bool can_charge_now: bool can_retry: bool class SubscriptionInvoiceListResponse(BaseModel): items: List[SubscriptionInvoiceItem] total_generated: int total_virtual: int has_more: bool # ─── Edit invoice request ───────────────────────────────────────────────────── class EditSubscriptionLineItem(BaseModel): """A single line item to set on a subscription invoice.""" title: Optional[str] = None description: Optional[str] = None unit_price: float = 0.0 quantity: int = 1 tax: Optional[float] = None cost: Optional[float] = None upcharge: Optional[float] = None discount: Optional[float] = None discount_type: Optional[str] = None product_id: Optional[int] = None class EditSubscriptionInvoiceRequest(BaseModel): scope: Literal["single", "all_remaining"] amount: Optional[float] = Field(None, gt=0) due_date: Optional[datetime] = None line_items: Optional[List[EditSubscriptionLineItem]] = None # payment_method_id removed (VULN-005): field was never written to DB and # created a misleading API surface suggesting payment method override. # auth_type removed (VULN-012): PRD forbids merchant changing auth_type. class EditSubscriptionInvoiceResponse(BaseModel): updated_invoices: int virtual_cycles_affected: int hpp_reauth_links: Optional[List[str]] = None # ─── Summary ────────────────────────────────────────────────────────────────── class SubscriptionSummaryResponse(BaseModel): active_count: int initializing_count: int past_due_count: int failed_count: int cancelled_count: int completed_count: int dunning_exhausted_count: int paused_count: int total_billed: float total_collected: float avg_subscription_value: float mrr: float # ─── Action requests ────────────────────────────────────────────────────────── class AdminSubscriptionUpdateRequest(BaseModel): action: Literal["cancel"] reason: str = Field(..., min_length=5) class ChargeNowResponse(BaseModel): status: Literal["charge_submitted"] invoice_id: str invoice_literal: str class RetryResponse(BaseModel): auth_type: Literal["pre_auth", "request_auth"] status: Literal["retry_submitted", "hpp_link_generated"] hpp_link: Optional[str] = None expires_at: Optional[datetime] = None # ─── List filter ───────────────────────────────────────────────────────────── class SubscriptionListFilterSchema(BaseModel): # VULN-011: max_length=200 prevents excessively large LIKE queries search: Optional[str] = Query(default=None, max_length=200, description="Search by literal or customer name") status: Optional[str] = Query(default=None, description="Comma-separated status ints, e.g. 200,201") customer_id: Optional[str] = Query(default=None) customer_literal: Optional[str] = Query(default=None, description="Filter by account_literal or customer_id string") payer_literal: Optional[str] = Query(default=None, description="Filter by contact_id opaque string") invoice_literal: Optional[str] = Query(default=None, description="Filter by invoice literal (e.g., INV000001)") created_after: Optional[datetime] = Query(default=None) created_before: Optional[datetime] = Query(default=None) sort_by: Optional[str] = Query(default=None, description="created_at | next_billing_date | total_paid") auth_type: Optional[str] = Query(default=None, description="Filter by authorization type: pre_auth | request_auth") # ─── Schedule ───────────────────────────────────────────────────────────────── class ScheduleItem(BaseModel): sequence: int sequence_label: str = "" transaction_id: Optional[str] = None schedule_date: Optional[datetime] = None auth_amount: float paid_date: Optional[datetime] = None paid_amount: float = 0.0 status: str # "paid" | "scheduled" | "cancelled" | "overdue" can_edit: bool can_cancel: bool can_pay_now: bool invoice_id: Optional[str] = None invoice_literal: Optional[str] = None class SubscriptionScheduleResponse(BaseModel): items: List[ScheduleItem] total: int # ─── Associations ───────────────────────────────────────────────────────────── class AssociationTransaction(BaseModel): transaction_id: str txn_literal: Optional[str] = None customer_name: Optional[str] = None tender_summary: Optional[str] = None txn_type: Optional[str] = None txn_date: Optional[datetime] = None txn_time: Optional[str] = None amount: float status: str class AssociationInvoice(BaseModel): invoice_id: str invoice_literal: Optional[str] = None customer_name: Optional[str] = None reference: Optional[str] = None issue_date: Optional[datetime] = None due_date: Optional[datetime] = None amount_due: float amount_paid: float status: str class SubscriptionAssociationsResponse(BaseModel): transactions: List[AssociationTransaction] invoices: List[AssociationInvoice]