"""Logic for account payee tax form info.""" import datetime from datetime import date from typing import Any, Dict, List, Optional from abacus_common_logic.models.base import db from flask import current_app, g from sqlalchemy import or_ from payee.connectors.secure_data.document import SecureDocument from payee.constants.constants import TAX_FORM_TYPES, TAX_FORM_TYPES_W8 from payee.logic.exceptions import LogicError from payee.logic.secure_document import ( create_secure_document_details, delete_secure_document, has_secure_document_details, save_secure_document_details_by_payee_id, save_secure_document_details_version, ) from payee.models.tax_form import TaxFormInfo, TaxFormInfoDetails from payee.utils.models import object_as_dict def get_records( account_payee_ids: List[int] = None, expiration_date_start: str = None, expiration_date_end: str = None, is_active: bool = None, limit: int = None, offset: int = None, ) -> Dict[str, Any]: """Get filtered records.""" query = TaxFormInfo.query.filter(TaxFormInfo.deleted_at.is_(None)) if account_payee_ids: query = query.filter(TaxFormInfo.account_payee_id.in_(account_payee_ids)) if expiration_date_start: query = query.filter(TaxFormInfo.expiration_date >= expiration_date_start) if expiration_date_end: query = query.filter(TaxFormInfo.expiration_date < expiration_date_end) if is_active is not None: if is_active: query = query.filter( or_( TaxFormInfo.expiration_date >= date.today(), TaxFormInfo.expiration_date.is_(None), ) ) else: query = query.filter(TaxFormInfo.expiration_date < date.today()) total_count = query.count() if limit: query = query.limit(limit) if offset: query = query.offset(offset) return { 'items': query.all(), 'total_count': total_count, } def get_records_details( account_payee_ids: List[int] = None, expiration_date_start: str = None, expiration_date_end: str = None, is_active: bool = None, limit: int = None, offset: int = None, obscure_pii: bool = True, ) -> Dict[str, Any]: """Get secure records details.""" records = get_records( account_payee_ids, expiration_date_start, expiration_date_end, is_active, limit, offset, ) items = [] for item in records['items']: document_class = item.tax_form_document_class # on heavy dataloader usage with multiple items it makes sense to # implement some new method with Dynamo BatchGetItem under the hood details = has_secure_document_details(item.account_payee_id, document_class) if details is None: # Generally this should not happen, because the corresponding secure # tax form part should exist for the tax form info instance # so let`s spam into logs on such cases of data inconsistency g.log.error( f'No secure tax details in Dynamo for account_payee_id {item.account_payee_id}' ) base_data = object_as_dict(item, ignore_fields=('revision_id',)) details_response = document_class.build_response( item.account_payee_id, details, obscure_pii=obscure_pii ) items.append(TaxFormInfoDetails(**base_data, details=details_response)) return {'items': items, 'total_count': records['total_count']} def create_or_update_record( account_payee_id: int, tax_form_type: TAX_FORM_TYPES, signed_date: Optional[datetime.date] = None, **details_payload, ) -> TaxFormInfoDetails: """Creates or updates tax form for payee.""" expiration_date = None if signed_date and tax_form_type in TAX_FORM_TYPES_W8: expiration_date = datetime.date(signed_date.year + 3, 12, 31) # creating/updating without persistence till the session committed tax_form_info = TaxFormInfo.get_by_account_payee_id(account_payee_id) if tax_form_info and tax_form_info.tax_form_type != tax_form_type: raise LogicError('Tax form type field is not editable.') previous_details_document = None if tax_form_info: previous_details_document = has_secure_document_details( account_payee_id, tax_form_info.tax_form_document_class ) tax_form_info = tax_form_info.update_attributes( tax_form_type=tax_form_type, signed_date=signed_date, expiration_date=expiration_date, ) else: tax_form_info = TaxFormInfo.build( account_payee_id=account_payee_id, tax_form_type=tax_form_type, signed_date=signed_date, expiration_date=expiration_date, ) document_class = tax_form_info.tax_form_document_class if not document_class: raise LogicError('Tax form type is not supported.') # the new version is being created every time under the hood # so we have both current(last) and versioned records in dynamo # after this call details = create_secure_document_details( current_app.config.get('SDM_CONFIG'), document_class, account_payee_id=account_payee_id, **details_payload, ) # commit/flush only if we reached here, # otherwise we should not save tax_form_info instance # and rollback secure details to previous state if failed to commit try: db.session.commit() except Exception as e: _rollback_to_document(account_payee_id, previous_details_document) db.session.rollback() raise e return TaxFormInfoDetails( **object_as_dict(tax_form_info, ignore_fields=('revision_id',)), details=details, ) def delete_record( tax_form_info_id: int, commit: bool = True ) -> tuple[str, SecureDocument]: """Soft delete tax form for payee and return revision and deleted doc.""" # get required tax form with checking that it exists and is active tax_form_info = TaxFormInfo.get_by_id_or_error(tax_form_info_id) # get current tax form from DynamoDB previous_details_document = has_secure_document_details( tax_form_info.account_payee_id, tax_form_info.tax_form_document_class ) if previous_details_document is None: raise LogicError('Document was not found') # create new DynamoDB record to get revision_id, # scanning all the records to find the last one could be costly revision_id = save_secure_document_details_version( tax_form_info.account_payee_id, previous_details_document.__class__, **previous_details_document.values, ) # delete current version (#0) in DynamoDB delete_secure_document( tax_form_info.account_payee_id, tax_form_info.tax_form_document_class ) try: # try to soft delete (set deleted_* fields and revision_id) in MySQL TaxFormInfo.delete_with_revision(tax_form_info, revision_id, commit) except Exception: # in case of errors create a current version (#0) of the document in DynamoDB (plus revision) _rollback_to_document(tax_form_info.account_payee_id, previous_details_document) db.session.rollback() raise return revision_id, previous_details_document def _rollback_to_document(account_payee_id: int, document: SecureDocument): """Rollback document to initial fields.""" if not document: return save_secure_document_details_by_payee_id( account_payee_id, document.__class__, **document.values, )