"""Logic for Ledger Adjustment With Details Applied.""" import logging from typing import Dict, List from abacus_common_logic.connectors.database import db from owsresponse import response from werkzeug.exceptions import abort from ledger.constants.error import ( ERROR_APPLY_SOFT_DELETED_WORKSHEET_ADJUSTMENT, ERROR_INVALID_BULK_DATA, ERROR_UNKNOWN_CURRENCY_FOR_FIELD, ) from ledger.models.ledger_adjustment_applied import LedgerAdjustmentApplied from ledger.models.ledger_adjustment_detail_applied import ( LedgerAdjustmentDetailApplied, ) from ledger.utils.currency import currency_exists # The DB guard trigger raises SIGNAL SQLSTATE '45000' (MySQL errno 1644) when a # ledger_*_applied insert references a soft-deleted worksheet_adjustment (the # apply-after-delete race). It is the only user-defined SIGNAL on these tables # today; if others are added, also match on message/SQLSTATE. TRIGGER_SOFT_DELETED_SIGNAL_ERRNO = 1644 def _is_soft_deleted_worksheet_adjustment_error(error) -> bool: """Return True if error is the DB guard rejecting a soft-deleted apply.""" orig = getattr(error, 'orig', None) return bool( orig is not None and getattr(orig, 'args', None) and orig.args[0] == TRIGGER_SOFT_DELETED_SIGNAL_ERRNO ) def bulk_create(request: List[Dict]) -> response.Response: """Bulk create logic. The expected request format example: .. code-block:: python [ { 'abacus_event_id': 1, 'account_id': 1, 'contract_id': 1, 'statement_period_id': 1, 'ledger_adjustment_id': 1, 'worksheet_adjustment_id': 1, 'adjustment_currency_code': 'USD', 'adjustment_amount': Decimal('3000.00'), 'adjustment_payee_currency_code': 'GBP', 'adjustment_amount_payee_currency': Decimal('3026.55'), 'apply_to_flowthrough_payment': False, 'details': [ { 'ledger_adjustment_detail_id': 1, 'worksheet_adjustment_detail_id': 1, 'adjustment_currency_code': 'USD', 'adjustment_amount': Decimal('1500.00'), 'adjustment_payee_currency_code': 'GBP', 'adjustment_amount_payee_currency': Decimal('1395.20') }, { 'ledger_adjustment_detail_id': 2, 'worksheet_adjustment_detail_id': 2, 'adjustment_currency_code': 'GBP', 'adjustment_amount': Decimal('1500.00'), 'adjustment_payee_currency_code': 'USD', 'adjustment_amount_payee_currency': Decimal('1631.35') } ] }, ] The `details` field is optional and would not be processed if the feature `ledger_adjustments_applied_details` is not enabled. """ new_ledger_adjustment_applied = list() currency_fields = ['adjustment_currency_code', 'adjustment_payee_currency_code'] # we already have an active transaction from session init # so we either commit or rollback it in case of any issues try: for record in request: validate_record(record, currency_fields) new_item = LedgerAdjustmentApplied.build( **{k: v for k, v in record.items() if k != 'details'} ) if record.get('details'): new_item.details = [ LedgerAdjustmentDetailApplied.build(**detail) for detail in record['details'] ] new_ledger_adjustment_applied.append(new_item) db.session.commit() except Exception as e: db.session.rollback() if _is_soft_deleted_worksheet_adjustment_error(e): logging.warning(f'bulk_create rejected by DB guard: {e}') abort(status=409, description=ERROR_APPLY_SOFT_DELETED_WORKSHEET_ADJUSTMENT) abort(status=400, description=ERROR_INVALID_BULK_DATA.format(error=e)) return response.Response( message={'message': 'OK', 'created': len(new_ledger_adjustment_applied)}, status=201, ) def validate_record(record: Dict, currency_fields: List): """Validate record.""" for field_name in currency_fields: if not currency_exists(record.get(field_name)): abort( status=400, description=ERROR_UNKNOWN_CURRENCY_FOR_FIELD.format( code=record.get(field_name), field=field_name ), ) if record.get('details'): for detail_record in record.get('details', []): _validate_detail(detail_record, currency_fields) def _validate_detail(detail: Dict, currency_fields: List): """Validate details field items for records.""" for field_name in currency_fields: if not currency_exists(detail.get(field_name)): abort( status=400, description=ERROR_UNKNOWN_CURRENCY_FOR_FIELD.format( code=detail.get(field_name), field=field_name ), ) def bulk_create_with_deduplication(request: List[Dict]) -> response.Response: """Bulk create logic with deduplication to handle retries.""" new_ledger_adjustment_applied: List[LedgerAdjustmentApplied] = list() currency_fields = ['adjustment_currency_code', 'adjustment_payee_currency_code'] worksheet_adjustment_ids: List[int] = [] for record in request: wa_id = record.get('worksheet_adjustment_id') if wa_id is not None: worksheet_adjustment_ids.append(wa_id) worksheet_adjustment_detail_ids: List[int] = [] for record in request: for detail in record.get('details', []): wad_id = detail.get('worksheet_adjustment_detail_id') if wad_id is not None: worksheet_adjustment_detail_ids.append(wad_id) # Get applied ledger adjustments by their worksheet_adjustment_ids applied_adjustments = _get_applied_ledger_adjustments_by_worksheet_adjustment_ids( worksheet_adjustment_ids=worksheet_adjustment_ids ) applied_detail_ids = _get_applied_worksheet_adjustment_detail_ids( worksheet_adjustment_detail_ids=worksheet_adjustment_detail_ids ) # Create a map for quick lookup applied_adjustment_map = { adj.worksheet_adjustment_id: adj for adj in applied_adjustments if adj.worksheet_adjustment_id is not None } # Get sets of applied IDs for quick lookup applied_adjustment_ids = set(applied_adjustment_map.keys()) applied_detail_ids = _get_applied_worksheet_adjustment_detail_ids( worksheet_adjustment_detail_ids=worksheet_adjustment_detail_ids ) # Filter out adjustments that have already been applied filtered_request = [] for record in request: if record.get('details'): filtered_details = [ detail for detail in record['details'] if detail.get('worksheet_adjustment_detail_id') not in applied_detail_ids ] if filtered_details: record['details'] = filtered_details else: record.pop('details', None) has_details = record.get('details') is not None is_applied = record.get('worksheet_adjustment_id') in applied_adjustment_ids if not is_applied or has_details: filtered_request.append(record) # If nothing to process, return early if not filtered_request: return response.Response( message={'message': 'OK', 'created': 0, 'filtered': len(request)}, status=201, ) # we already have an active transaction from session init # so we either commit or rollback it in case of any issues try: for record in filtered_request: new_item = applied_adjustment_map.get(record.get('worksheet_adjustment_id')) if not new_item: validate_record(record, currency_fields) new_item = LedgerAdjustmentApplied.build( **{k: v for k, v in record.items() if k != 'details'} ) if record.get('details'): new_item.details.extend( [ LedgerAdjustmentDetailApplied.build(**detail) for detail in record['details'] ] ) new_ledger_adjustment_applied.append(new_item) db.session.commit() except Exception as e: db.session.rollback() if _is_soft_deleted_worksheet_adjustment_error(e): logging.warning(f'bulk_create_with_deduplication rejected by DB guard: {e}') abort(status=409, description=ERROR_APPLY_SOFT_DELETED_WORKSHEET_ADJUSTMENT) logging.error(f'bulk_create_with_deduplication error: {e}') abort(status=400, description=ERROR_INVALID_BULK_DATA.format(error=e)) return response.Response( message={ 'message': 'OK', 'created': len(new_ledger_adjustment_applied), 'filtered': len(request) - len(filtered_request), }, status=201, ) def _get_applied_ledger_adjustments_by_worksheet_adjustment_ids( worksheet_adjustment_ids: List[int], ) -> List[LedgerAdjustmentApplied]: """Get LedgerAdjustmentApplied records for worksheet_adjustment_ids that have already been applied. Args: worksheet_adjustment_ids (List[int]): List of worksheet_adjustment_ids to check for prior application. Returns: List[LedgerAdjustmentApplied]: LedgerAdjustmentApplied records that match the input IDs. """ if not worksheet_adjustment_ids: return [] try: valid_ids = [ adjustment_id for adjustment_id in worksheet_adjustment_ids if adjustment_id is not None ] if not valid_ids: return [] query = db.session.query(LedgerAdjustmentApplied).filter( LedgerAdjustmentApplied.worksheet_adjustment_id.in_(valid_ids) ) applied_records = query.all() return applied_records except Exception as e: logging.error(f'Error fetching applied worksheet_adjustment_ids: {e}') return [] def _get_applied_worksheet_adjustment_detail_ids( worksheet_adjustment_detail_ids: List[int], ) -> set: """Get worksheet_adjustment_detail_ids that have already been applied. Args: worksheet_adjustment_detail_ids (List[int]): List of detail IDs to check for prior application. Returns: set: Subset of input IDs that have already been applied. """ if not worksheet_adjustment_detail_ids: return set() try: # Filter out None values before querying valid_ids = [ detail_id for detail_id in worksheet_adjustment_detail_ids if detail_id is not None ] if not valid_ids: return set() query = ( db.session.query( LedgerAdjustmentDetailApplied.worksheet_adjustment_detail_id ) .filter( LedgerAdjustmentDetailApplied.worksheet_adjustment_detail_id.in_( valid_ids ) ) .distinct() ) applied_records = query.all() return set(record[0] for record in applied_records if record[0] is not None) except Exception as e: logging.error(f'Error fetching applied worksheet_adjustment_detail_ids: {e}') return set() def _get_applied_worksheet_adjustment_detail_ids( worksheet_adjustment_detail_ids: List[int], ) -> set: """Get worksheet_adjustment_detail_ids that have already been applied. Queries ledger_adjustment_detail_applied table to identify which of the provided worksheet_adjustment_detail_ids have already been processed. Args: worksheet_adjustment_detail_ids (List[int]): List of detail IDs to check for prior application. Returns: set: Subset of input IDs that have already been applied. """ if not worksheet_adjustment_detail_ids: return set() try: # Filter out None values before querying valid_ids = [ detail_id for detail_id in worksheet_adjustment_detail_ids if detail_id is not None ] if not valid_ids: return set() query = ( db.session.query( LedgerAdjustmentDetailApplied.worksheet_adjustment_detail_id ) .filter( LedgerAdjustmentDetailApplied.worksheet_adjustment_detail_id.in_( valid_ids ) ) .distinct() ) applied_records = query.all() return set(record[0] for record in applied_records if record[0] is not None) except Exception as e: logging.error(f'Error fetching applied worksheet_adjustment_detail_ids: {e}') return set()