"""Payment allocation models""" from typing import Iterable, List, Optional, Self, Tuple from abacus_common_logic.models.base import BaseSoftDeleteModel, db from abacus_common_logic.utils.users import get_flask_user_id from sqlalchemy import Enum, func, select, update from sqlalchemy.orm import validates from payment.constants.constants import ( PAYEE_TYPES, PAYMENT_ALLOCATION_FLOWTHROUGH_STATUS_TRANSITIONS, PAYMENT_ALLOCATION_LEDGER_STATUSES, PAYMENT_ALLOCATION_STATUSES, PAYMENT_ALLOCATION_TYPES, ) class BasePaymentAllocation(BaseSoftDeleteModel): """Base Payment Allocation model for single-table inheritance.""" __tablename__ = 'payment_allocation' __mapper_args__ = { 'polymorphic_on': 'payment_allocation_type', } __allocation_type__ = None PAYMENT_STATUS_TRANSITIONS = {} def __init__(self, **kwargs): """Initialize payment allocation with validation. Validates that payment_allocation_type matches the subclass's __allocation_type__ if the subclass defines one. Raises: ValueError: If payment_allocation_type is provided and doesn't match the subclass's __allocation_type__. """ provided_type = kwargs.pop('payment_allocation_type', None) if self.__allocation_type__ is not None: if provided_type is not None and provided_type != self.__allocation_type__: raise ValueError( f'{self.__class__.__name__} requires payment_allocation_type to be ' f'{self.__allocation_type__}, got {provided_type}' ) kwargs['payment_allocation_type'] = self.__allocation_type__ elif provided_type is not None: kwargs['payment_allocation_type'] = provided_type super().__init__(**kwargs) payment_allocation_id = db.Column( db.Integer, primary_key=True, autoincrement=True, comment='Primary key.' ) contract_id = db.Column( db.Integer, nullable=False, comment='Foreign key to the contract we are adjusting the payment for.', ) payee_type = db.Column( Enum(*PAYEE_TYPES, name='payee_type'), nullable=False, comment='Payee type that this payment adjustment is for.', ) payee_id = db.Column( db.Integer, nullable=False, comment='ID of the payee. References either payee or account_payee table depending on payee_type.', ) statement_period_id = db.Column( db.Integer, nullable=False, comment='Foreign key to the statement period this is adjusting.', ) payment_allocation_type = db.Column( Enum(*PAYMENT_ALLOCATION_TYPES, name='payment_allocation_type'), nullable=False, comment='Type of payment allocation.', ) amount_to_payment = db.Column( db.Numeric(20, 2), nullable=False, comment='Amount of money to pay to payee.' ) payment_status = db.Column( Enum(*PAYMENT_ALLOCATION_STATUSES, name='payment_status'), nullable=False, comment='Current payment status.', ) payment_status_modified = db.Column( db.DateTime, nullable=True, comment='Timestamp when payment_status was last modified.', ) amount_to_ledger = db.Column( db.Numeric(20, 2), nullable=False, comment='Amount of money to add/deduct from contract ledger.', ) ledger_status = db.Column( Enum(*PAYMENT_ALLOCATION_LEDGER_STATUSES, name='ledger_status'), nullable=False, comment='Current ledger status.', ) ledger_status_modified = db.Column( db.DateTime, nullable=True, comment='Timestamp when ledger_status was last modified.', ) currency_code = db.Column(db.String(3), nullable=False, comment='Currency code.') description = db.Column( db.Text, nullable=True, comment='General description about purpose of this allocation.', ) @classmethod def filter_active(cls): """Return a select() statement pre-filtered for non-deleted records.""" return select(cls).where(cls.deleted_at.is_(None)) @validates('payment_status') def validate_payment_status(self, key, payment_status): """Validation of the payment status transition.""" if not self.PAYMENT_STATUS_TRANSITIONS: return payment_status if payment_status not in self.PAYMENT_STATUS_TRANSITIONS.get( self.payment_status, set() ): raise ValueError( f'Invalid payment status transition: {self.payment_status} >> {payment_status}' ) return payment_status @classmethod def soft_delete_by_ids(cls, payment_allocation_ids: Iterable[int]): """Delete items by payment_allocation_ids.""" db.session.execute( update(cls) .where( cls.deleted_at.is_(None), cls.payment_allocation_id.in_(payment_allocation_ids), cls.payment_status != PAYMENT_ALLOCATION_STATUSES.PAID, ) .values( deleted_at=cls.current_timestamp(), deleted_by=get_flask_user_id(), ) .execution_options(synchronize_session=False) ) db.session.commit() class PaymentAllocationFlowthrough(BasePaymentAllocation): """Payment Allocation Flowthrough model.""" __allocation_type__ = PAYMENT_ALLOCATION_TYPES.FLOWTHROUGH __mapper_args__ = { 'polymorphic_identity': __allocation_type__, } PAYMENT_STATUS_TRANSITIONS = PAYMENT_ALLOCATION_FLOWTHROUGH_STATUS_TRANSITIONS @classmethod def get_filtered_items( cls, limit: int, offset: int, payment_allocation_ids: Optional[List[int]] = None, contract_ids: Optional[List[int]] = None, payment_statuses: Optional[List[str]] = None, ledger_statuses: Optional[List[str]] = None, ) -> Tuple[List[Self], int]: """Get filtered flowthrough payment allocations. Args: limit: Maximum number of records to return offset: Pagination offset payment_allocation_ids: Optional list of payment allocation IDs to filter by contract_ids: Optional list of contract IDs to filter by payment_statuses: Optional list of payment statuses to filter by ledger_statuses: Optional list of ledger statuses to filter by Returns: Tuple of records list and total count """ stmt = cls.filter_active() if payment_allocation_ids: stmt = stmt.where(cls.payment_allocation_id.in_(payment_allocation_ids)) if contract_ids: stmt = stmt.where(cls.contract_id.in_(contract_ids)) if payment_statuses: stmt = stmt.where(cls.payment_status.in_(payment_statuses)) if ledger_statuses: stmt = stmt.where(cls.ledger_status.in_(ledger_statuses)) items = db.session.execute(stmt.offset(offset).limit(limit + 1)).scalars().all() has_more = len(items) > limit if has_more: items = items[:limit] if not has_more and (offset == 0 or items): # We're on the last page, so total = offset + remaining items. # Avoids an expensive COUNT query when we can deduce the total. # The `items` check excludes an offset past the end (empty last page # that isn't the first page), where arithmetic would be wrong. total_count = offset + len(items) else: total_count = db.session.execute( select(func.count()).select_from(stmt.subquery()) ).scalar_one() return items, total_count @classmethod def get_active_by_ids( cls, ids: Iterable[int] ) -> list['PaymentAllocationFlowthrough']: """Get active records by IDs.""" return ( db.session.execute( cls.filter_active().where(cls.payment_allocation_id.in_(list(ids))) ) .scalars() .all() )