"""Logic related to LedgerAccountContract.""" from marshmallow import ValidationError from owsresponse import response from ledger.models.ledger_account_contract import LedgerAccountContract from ledger.schemas.ledger_account_contract import ( LedgerAccountContractFilterSchema, LedgerAccountContractListSchema, LedgerAccountContractPayableBalanceSchema, ) from ledger.utils.format_error import validation_error def get_payable_balance_by_account(account_id: int) -> response.Response: """Get an account's 'payable_balance' (the sum of positive contract balances). Args: account_id (int): ID of the account Returns: a response """ payable_balance = LedgerAccountContract.get_payable_balance_by_account(account_id) message = LedgerAccountContractPayableBalanceSchema().dump(payable_balance) return response.Response(message=message, status=200) def get_ledger_list_by_account_id( account_id: int, request_params: dict ) -> response.Response: """GET a list of ledgers for a specified account. Args: account_id (int): ID of the account request_params (dict)(Optional): query string parameters - limit (int): pagination limit; defaults to 100 - offset (int): pagination offset; defaults to 0 - event_names (str): a string of abacus events, separated by commas - statement_period_ids (str): a string of statement period IDs, separated by commas - contract_ids (str): a string of contract IDs, separated by commas Returns: a list of ledgers. """ try: params = LedgerAccountContractFilterSchema().load(request_params) items, total_count = LedgerAccountContract.get_ledger_list_by_account_id( account_id, **params ) message = dict( items=LedgerAccountContractListSchema().dump(items, many=True), total_count=total_count, ) except ValidationError as exc: return validation_error(str(exc)) except Exception as e: raise e return response.Response(message=message, status=200)