"""WorksheetAdjustment model.""" import typing from decimal import Decimal from typing import NamedTuple from abacus_common_logic.connectors.database import db from abacus_common_logic.models.base import BaseModel from abacus_common_logic.utils.dates import current_timestamp from abacus_common_logic.utils.users import get_flask_user_id from sqlalchemy import and_, case, func, literal_column, or_, select, table from sqlalchemy.orm import column_property, lazyload from abacus_worksheet.constants.constants import DEFAULT_PAGE_LIMIT, DEFAULT_PAGE_OFFSET from abacus_worksheet.models.worksheet_adjustment_detail import ( WorksheetAdjustmentDetail, ) from abacus_worksheet.utils.deletion_status import predicates class AdjustmentQueryBuild(NamedTuple): """Return shape for _query_to_get_worksheet_adjustments_and_details.""" query: object flowthrough_expr: object class WorksheetAdjustment(BaseModel): """WorksheetAdjustment model.""" __tablename__ = 'worksheet_adjustment' worksheet_adjustment_id = db.Column(db.Integer, primary_key=True) statement_period_adjustment_file_id = db.Column(db.Integer, nullable=False) abacus_event_id = db.Column(db.Integer, nullable=False) account_id = db.Column(db.Integer, nullable=False) contract_id = db.Column(db.Integer, nullable=True) activity_statement_period_id = db.Column(db.Integer, nullable=False) apply_to_statement_period_id = db.Column(db.Integer, nullable=False) reference_adjustment_type_id = db.Column(db.Integer, nullable=False) apply_to_flowthrough_payment = db.Column(db.Boolean, nullable=True) adjustment_amount = db.Column(db.Numeric(20, 2), nullable=False) adjustment_currency_code = db.Column(db.String(3), nullable=False) note = db.Column(db.Text, nullable=True) internal_note = db.Column(db.Text, nullable=True) deleted_at = db.Column(db.DateTime, nullable=True) deleted_by = db.Column(db.String(180), nullable=True) details = db.relationship( 'WorksheetAdjustmentDetail', primaryjoin=( 'and_(' 'WorksheetAdjustment.worksheet_adjustment_id ' '== WorksheetAdjustmentDetail.worksheet_adjustment_id, ' 'WorksheetAdjustmentDetail.deleted_at.is_(None)' ')' ), lazy='joined', viewonly=True, ) account_currency_code = column_property( select([literal_column('account_payment_term.currency_code')]) .select_from(table('account_payment_term')) .where(literal_column('account_payment_term.account_id') == account_id) ) account_payment_entity_id = column_property( select([literal_column('account_payment_term.payment_entity_id')]) .select_from(table('account_payment_term')) .where(literal_column('account_payment_term.account_id') == account_id) ) @classmethod def get_by_statement_period_adjustment_file_id( cls, statement_period_adjustment_file_id: int, limit: int, offset: int ): """Get a list of worksheet adjustments by statement_period_adjustment_file_id. Args: statement_period_adjustment_file_id (int): id of the statement_period_adjustment_file limit (int): pagination limit offset (int): pagination offset Returns: A tuple of result items and total count. """ query = cls.query.filter( cls.statement_period_adjustment_file_id == statement_period_adjustment_file_id, cls.deleted_by.is_(None), cls.deleted_at.is_(None), ) items = ( query.order_by(cls.worksheet_adjustment_id.asc()) .limit(limit) .offset(offset) .all() ) total_count = query.count() return items, total_count @classmethod def soft_delete_worksheet_adjustments( cls, statement_period_adjustment_file_id: int ): """Soft delete worksheet adjustments by statement_period_adjustment_file_id. Args: statement_period_adjustment_file_id (int): id of the statement_period_adjustment_file """ db.session.query(WorksheetAdjustment).filter( WorksheetAdjustment.statement_period_adjustment_file_id == statement_period_adjustment_file_id, WorksheetAdjustment.deleted_at.is_(None), ).update( { WorksheetAdjustment.deleted_at: current_timestamp(), WorksheetAdjustment.deleted_by: get_flask_user_id(), } ) @classmethod def select_for_update_by_ids(cls, worksheet_adjustment_ids): """Lock the given worksheet_adjustment rows for the current transaction. The `details` relationship is loaded eagerly (lazy='joined') by default; here it is deferred with lazyload so the FOR UPDATE query does not also join and lock the detail rows. Locking details would invert lock order against the whole-file delete (which locks details then parents) and risk a deadlock. The callers only read scalar columns off these rows. """ return ( db.session.query(cls) .options(lazyload(cls.details)) .filter(cls.worksheet_adjustment_id.in_(worksheet_adjustment_ids)) .with_for_update() .all() ) @classmethod def get_applied_ids(cls, worksheet_adjustment_ids): """Return the subset of ids that have a ledger_adjustment_applied row.""" if not worksheet_adjustment_ids: return set() rows = ( db.session.query(literal_column('laa.worksheet_adjustment_id')) .select_from(table('ledger_adjustment_applied').alias('laa')) .filter( literal_column('laa.worksheet_adjustment_id').in_( worksheet_adjustment_ids ) ) .all() ) return {row[0] for row in rows} @classmethod def get_applied_ids_for_file(cls, statement_period_adjustment_file_id): """Return worksheet_adjustment_ids in a file that are already applied.""" rows = ( db.session.query(cls.worksheet_adjustment_id) .join( table('ledger_adjustment_applied').alias('laa'), cls.worksheet_adjustment_id == literal_column('laa.worksheet_adjustment_id'), ) .filter( cls.statement_period_adjustment_file_id == statement_period_adjustment_file_id ) .all() ) return {row[0] for row in rows} @classmethod def soft_delete_by_ids(cls, worksheet_adjustment_ids): """Soft-delete the live worksheet_adjustment rows in the id list.""" db.session.query(cls).filter( cls.worksheet_adjustment_id.in_(worksheet_adjustment_ids), cls.deleted_at.is_(None), ).update( { cls.deleted_at: current_timestamp(), cls.deleted_by: get_flask_user_id(), }, synchronize_session=False, ) @classmethod def restore_by_ids(cls, worksheet_adjustment_ids): """Restore the soft-deleted worksheet_adjustment rows in the id list.""" db.session.query(cls).filter( cls.worksheet_adjustment_id.in_(worksheet_adjustment_ids), cls.deleted_at.isnot(None), ).update( {cls.deleted_at: None, cls.deleted_by: None}, synchronize_session=False, ) @staticmethod def _query_to_get_worksheet_adjustments_and_details( is_deleted: bool | None = False, ): """Build a query to get worksheet adjustments and details. Args: is_deleted (bool | None): BooleanFilter.to_bool() value; False (default) active only, True deleted only, None all. Applied to both the parent and detail predicates via the shared helper. """ # This subquery returns expenses. subquery = WorksheetAdjustmentDetail._query_to_get_worksheet_adjustment_detail( is_deleted ).subquery() apply_to_flowthrough_payment_expr = case( [ ( subquery.c.worksheet_adjustment_detail_id.isnot(None), subquery.c.apply_to_flowthrough_payment, ) ], else_=WorksheetAdjustment.apply_to_flowthrough_payment, ) adjustment_amount_expr = case( [ ( subquery.c.worksheet_adjustment_detail_id.isnot(None), subquery.c.amount, ) ], else_=WorksheetAdjustment.adjustment_amount, ) query = ( db.session.query( subquery.c.worksheet_adjustment_detail_id, WorksheetAdjustment.worksheet_adjustment_id, WorksheetAdjustment.statement_period_adjustment_file_id, WorksheetAdjustment.account_id, WorksheetAdjustment.contract_id, subquery.c.upc, WorksheetAdjustment.adjustment_currency_code, WorksheetAdjustment.activity_statement_period_id, WorksheetAdjustment.apply_to_statement_period_id, apply_to_flowthrough_payment_expr.label('apply_to_flowthrough_payment'), adjustment_amount_expr.label('adjustment_amount'), case( [ ( subquery.c.worksheet_adjustment_detail_id.isnot(None), subquery.c.note, ) ], else_=WorksheetAdjustment.note, ).label('note'), case( [ ( subquery.c.worksheet_adjustment_detail_id.isnot(None), subquery.c.internal_note, ) ], else_=WorksheetAdjustment.internal_note, ).label('internal_note'), case( [ ( subquery.c.worksheet_adjustment_detail_id.isnot(None), subquery.c.reference_adjustment_type_id, ) ], else_=WorksheetAdjustment.reference_adjustment_type_id, ).label('reference_adjustment_type_id'), subquery.c.distribution_type, # Deletion is parent-authoritative (parent and details flip in # lockstep), so project the parent's markers directly. WorksheetAdjustment.deleted_at, WorksheetAdjustment.deleted_by, ) .outerjoin( WorksheetAdjustmentDetail, and_( WorksheetAdjustmentDetail.worksheet_adjustment_id == WorksheetAdjustment.worksheet_adjustment_id, WorksheetAdjustmentDetail.worksheet_adjustment_id.is_(None), ), ) .outerjoin( subquery, subquery.c.worksheet_adjustment_id == WorksheetAdjustment.worksheet_adjustment_id, ) .filter(*predicates(WorksheetAdjustment, is_deleted)) .order_by(WorksheetAdjustment.worksheet_adjustment_id.asc()) ) return AdjustmentQueryBuild(query, apply_to_flowthrough_payment_expr) @classmethod def get_worksheet_adjustments_and_details( cls, statement_period_adjustment_file_id: int, account_ids: str = None, contract_ids: str = None, apply_to_flowthrough_payment: str = None, limit: int = DEFAULT_PAGE_LIMIT, offset: int = DEFAULT_PAGE_OFFSET, is_deleted: bool | None = False, ) -> tuple: """Get a list of worksheet adjustments and details for a specified adjustment file id. Args: statement_period_adjustment_file_id (int): id of the statement_period_adjustment_file account_ids (str): comma separated list of account ids contract_ids (str): comma separated list of contract ids apply_to_flowthrough_payment (str): comma separate list of filter values i.e 0, 1 or null limit (int): pagination limit offset (int): pagination offset is_deleted (bool | None): False (default) active only, True deleted only, None all Returns: A tuple containing items and total count """ built = cls._query_to_get_worksheet_adjustments_and_details(is_deleted) query = built.query apply_to_flowthrough_payment_expr = built.flowthrough_expr query = query.filter( cls.statement_period_adjustment_file_id == statement_period_adjustment_file_id ) if contract_ids is not None: ids = [int(contract_id) for contract_id in contract_ids.split(',')] query = query.filter(WorksheetAdjustment.contract_id.in_(ids)) if account_ids is not None: ids = [int(account_id) for account_id in account_ids.split(',')] query = query.filter(WorksheetAdjustment.account_id.in_(ids)) if apply_to_flowthrough_payment is not None: flowthrough_payment_filters = list() filter_values = apply_to_flowthrough_payment.split(',') if 'null' in filter_values: flowthrough_payment_filters.append( apply_to_flowthrough_payment_expr.is_(None) ) flags = [int(flag) for flag in filter_values if flag != 'null'] flowthrough_payment_filters.append( apply_to_flowthrough_payment_expr.in_(flags) ) query = query.filter(or_(*flowthrough_payment_filters)) items = ( query.order_by(cls.worksheet_adjustment_id.asc()) .limit(limit) .offset(offset) .all() ) total_count = query.count() return items, total_count @classmethod def get_worksheet_adjustments_deleted_aggregate( cls, statement_period_adjustment_file_id: int ) -> dict: """Deleted-entry count and currency-agnostic amount at the grid's detail grain. Only deleted entries are aggregated; active totals are the grid reader's total_count for is_deleted=False. The amount sums across currencies deliberately (currency-agnostic), matching the legacy currency_agnostic_total_amount convention; it is not a real money total. This aggregate is whole-file: it does not honor the grid's account, contract, or flowthrough filters. """ built = cls._query_to_get_worksheet_adjustments_and_details(is_deleted=True) subquery = ( built.query.filter( cls.statement_period_adjustment_file_id == statement_period_adjustment_file_id ) .order_by(None) .subquery() ) deleted_count, deleted_amount = ( db.session.query(func.count(), func.sum(subquery.c.adjustment_amount)) .select_from(subquery) .one() ) return { 'deleted_count': deleted_count, 'currency_agnostic_deleted_amount': deleted_amount if deleted_amount is not None else Decimal('0.00'), } @classmethod def get_pending_worksheet_adjustments( cls, limit: int, offset: int, statement_period_id: typing.Optional[int] = None, reference_payment_entity_id: typing.Optional[int] = None, ): """Get a list of pending worksheet adjustments. Args: limit (int): pagination limit offset (int): pagination offset statement_period_id (int): id of the statement_period_id reference_payment_entity_id (int): id of the reference payment entity Returns: A tuple containing items and total count """ # Pinned to active (is_deleted=False): the apply pipeline must never # ingest soft-deleted rows. There is no DB-level guard for this; do not # parameterize. built = cls._query_to_get_worksheet_adjustments_and_details(is_deleted=False) query = built.query query = query.join( table('ledger_adjustment_applied').alias('laa'), cls.worksheet_adjustment_id == literal_column('laa.worksheet_adjustment_id'), isouter=True, ).filter(literal_column('laa.worksheet_adjustment_id').is_(None)) if statement_period_id is not None: query = query.filter( cls.apply_to_statement_period_id == statement_period_id ) if reference_payment_entity_id is not None: query = query.join( table('account_payment_term').alias('apt'), cls.account_id == literal_column('apt.account_id'), ).filter( literal_column('apt.payment_entity_id') == reference_payment_entity_id ) items = query.limit(limit).offset(offset).all() total_count = query.count() return items, total_count @classmethod def get_worksheet_adjustments_contracts_by_file_id( cls, statement_period_adjustment_file_id: int, contract_search_term: typing.Optional[str] = None, limit: int = DEFAULT_PAGE_LIMIT, offset: int = DEFAULT_PAGE_OFFSET, ): """Get a list of contracts for worksheet adjustments associated with a specified statement_period_adjustment_file_id. Args: statement_period_adjustment_file_id (int): id of the statement_period_adjustment_file contract_search_term(str): either contract_id or contract_name limit(int): pagination limit offset(int): pagination offset Returns: A tuple containing items and total count """ query = cls.query.join( table('contract').alias('c'), cls.contract_id == literal_column('c.contract_id'), ).filter( cls.statement_period_adjustment_file_id == statement_period_adjustment_file_id, cls.deleted_by.is_(None), cls.deleted_at.is_(None), ) if contract_search_term is not None: search_term_text = contract_search_term.replace('\\', '\\\\').replace( '%', '\\%' ) query = query.filter( or_( literal_column('c.contract_name').ilike(f'%{search_term_text}%'), literal_column('c.contract_id').ilike(f'%{search_term_text}%'), ) ) query = query.with_entities(cls.contract_id).distinct() items = ( query.order_by(func.lower(literal_column('c.contract_name'))) .offset(offset) .limit(limit) .all() ) total_count = query.count() return items, total_count @classmethod def get_worksheet_adjustments_accounts_by_file_id( cls, statement_period_adjustment_file_id: int, account_search_term: typing.Optional[str] = None, limit: int = DEFAULT_PAGE_LIMIT, offset: int = DEFAULT_PAGE_OFFSET, ): """Get a list of accounts for worksheet adjustments associated with a specified statement_period_adjustment_file_id. Args: statement_period_adjustment_file_id (int): id of the statement_period_adjustment_file account_search_term(str): either account_id or account_name limit(int): pagination limit offset(int): pagination offset Returns: A tuple containing items and total count """ query = cls.query.join( table('account').alias('a'), cls.account_id == literal_column('a.account_id'), ).filter( cls.statement_period_adjustment_file_id == statement_period_adjustment_file_id, cls.deleted_by.is_(None), cls.deleted_at.is_(None), ) if account_search_term is not None: search_term_text = account_search_term.replace('\\', '\\\\').replace( '%', '\\%' ) query = query.filter( or_( literal_column('a.account_name').ilike(f'%{search_term_text}%'), literal_column('a.account_id').ilike(f'%{search_term_text}%'), ) ) query = query.with_entities(cls.account_id).distinct() items = ( query.order_by(func.lower(literal_column('a.account_name'))) .offset(offset) .limit(limit) .all() ) total_count = query.count() return items, total_count @classmethod def get_adjustments_by_period_and_type_id(cls, period_id: int, type_id: int): """Test method to verify the worksheet adjustments model is working. Args: period_id (int): period id type_id (int): type id Returns: A tuple containing items, total count and total amount """ query = cls.query.join( table('ledger_adjustment_applied').alias('laa'), literal_column('laa.worksheet_adjustment_id') == cls.worksheet_adjustment_id, ).filter( cls.activity_statement_period_id == period_id, cls.reference_adjustment_type_id == type_id, cls.deleted_by.is_(None), cls.deleted_at.is_(None), ) items = query.all() total_count = query.count() adjustment_amount_sum = query.with_entities( func.sum(cls.adjustment_amount) ).scalar() total_adjustment_amount = float( adjustment_amount_sum if adjustment_amount_sum is not None else 0.0 ) return items, total_count, total_adjustment_amount