"""Worksheet adjustment detail model.""" from sqlalchemy import Column from sqlalchemy import DateTime from sqlalchemy import desc from sqlalchemy import Enum from sqlalchemy import ForeignKey from sqlalchemy import func from sqlalchemy import Integer from sqlalchemy import literal_column from sqlalchemy import Numeric from sqlalchemy import String from sqlalchemy import table from sqlalchemy import Text from sqlalchemy.engine import Row from moneyhub.constants.constants import DistributionType from moneyhub.constants.constants import GroupBy from moneyhub.models.mysql_base import BaseModel from moneyhub.models.reference_adjustment_type import ReferenceAdjustmentType class WorksheetAdjustmentDetail(BaseModel): """Worksheet adjustment model.""" __tablename__ = 'worksheet_adjustment_detail' worksheet_adjustment_detail_id = Column(Integer, primary_key=True) statement_period_adjustment_file_id = Column(Integer, nullable=False) worksheet_adjustment_id = Column(Integer, nullable=False) reference_adjustment_type_id = Column( Integer, ForeignKey(ReferenceAdjustmentType.reference_adjustment_type_id)) internal_note = Column(Text, nullable=True) deleted_by = Column(String(180), nullable=True) deleted_at = Column(DateTime, nullable=True) account_id = Column(Integer, nullable=False) activity_statement_period_id = Column(Integer, nullable=False) amount = Column(Numeric(20, 2), nullable=False) apply_to_statement_period_id = Column(Integer, nullable=False) contract_id = Column(Integer, nullable=True) currency_code = Column(String(3), nullable=False) created_by = Column(String(180), nullable=False) created_at = Column(DateTime, nullable=True) distribution_type = Column( Enum( *DistributionType, name='distribution_type', create_type=False), nullable=False ) last_modified = Column(DateTime, nullable=False) last_modified_by = Column(String(180), nullable=False) note = Column(Text, nullable=True) upc = Column(String(20), nullable=False) @classmethod def get_by_account_id( cls, account_id: int, visible_periods: list, contract_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, upc: str | None, expense_type_id: int | None, distribution_type: str | None, group_by: GroupBy | None = None, ) -> list: """Get worksheet adjustments detail by an account_id. Args: account_id (int): The id of an account visible_periods (list): List of the visible periods for account contract_id (int): Optional id of the contract statement_period_id_start (int): Optional id of the statement period to range from statement_period_id_end (int): Optional id of the statement period to range to upc (str): Optional UPC to filter by expense_type_id (int): Optional expense type id to filter by distribution_type (int): Optional distributon type to filter by group_by (str): Optional dimension to group expenses by Returns: list: list of ledger adjustments """ filters = [ cls.account_id == account_id, cls.apply_to_statement_period_id.in_(visible_periods) ] grouping = [] selects = [ cls.worksheet_adjustment_detail_id, cls.account_id, cls.activity_statement_period_id, cls.apply_to_statement_period_id, cls.currency_code, cls.contract_id, cls.created_at, cls.created_by, cls.distribution_type, cls.last_modified, cls.last_modified_by, cls.note, cls.reference_adjustment_type_id, literal_column('rat.type_name').label('reference_adjustment_type_name'), cls.upc, literal_column('lada.adjustment_payee_currency_code').label('adjustment_payee_currency_code') # noqa: E501 ] if contract_id: filters.append(cls.contract_id == contract_id) if statement_period_id_start and statement_period_id_end: filters.append(cls.apply_to_statement_period_id.between( statement_period_id_start, statement_period_id_end)) if upc: filters.append(cls.upc == upc) if expense_type_id: filters.append(cls.reference_adjustment_type_id == expense_type_id) if distribution_type: filters.append(cls.distribution_type == distribution_type) if group_by: match group_by: case GroupBy.UPC: selects = [ cls.account_id, cls.upc, cls.distribution_type, literal_column('lada.adjustment_payee_currency_code').label('adjustment_payee_currency_code'), # noqa: E501 func.sum(cls.amount).label('amount'), func.sum(literal_column('lada.adjustment_amount_payee_currency')).label( 'adjustment_amount_payee_currency') ] grouping.append(cls.upc) grouping.append(cls.distribution_type) grouping.append(literal_column('lada.adjustment_payee_currency_code').label( 'adjustment_payee_currency_code')) case GroupBy.EXPENSE_TYPE_ID: selects = [ cls.reference_adjustment_type_id, cls.account_id, literal_column('rat.type_name').label('reference_adjustment_type_name'), cls.apply_to_statement_period_id, literal_column('lada.adjustment_payee_currency_code').label('adjustment_payee_currency_code'), # noqa: E501 func.sum(cls.amount).label('amount'), func.sum(literal_column('lada.adjustment_amount_payee_currency')).label( 'adjustment_amount_payee_currency') ] grouping.append(cls.reference_adjustment_type_id) grouping.append(literal_column('lada.adjustment_payee_currency_code').label( 'adjustment_payee_currency_code')) # noqa: E501 else: selects.append(cls.amount) selects.append(literal_column('lada.adjustment_amount_payee_currency').label( 'adjustment_amount_payee_currency')) return cls.query.with_entities(*selects) \ .join(table('ledger_adjustment_detail_applied').alias('lada'), cls.worksheet_adjustment_detail_id == literal_column('lada.worksheet_adjustment_detail_id')) \ .join(table('reference_adjustment_type').alias('rat'), literal_column('rat.reference_adjustment_type_id') == cls.reference_adjustment_type_id ) \ .distinct() \ .where(*filters) \ .group_by(*grouping) \ .order_by(desc(cls.worksheet_adjustment_detail_id)) \ .all() @classmethod def get_upcs_by_account(cls, account_id: int) -> list: """Get a list of available UPCs based on an account ID. Args: account_id (int): ID of the account. Returns: list: list of available UPCs and distribution types """ return cls.query \ .with_entities( cls.account_id, cls.upc, cls.distribution_type, ) \ .distinct() \ .filter(cls.account_id == account_id) \ .order_by(cls.upc.asc()) \ .all() @classmethod def get_account_adjustment_detail_activity(cls, account_id: int) -> Row: """Get adjustment account activity. Args: account_id (int): Account to get revenue for. Returns: Row: SqlAlchemy row containing adjustment detail information. """ return cls.query.filter(cls.account_id == account_id).first()