"""Contract logic.""" import logging from datetime import date import httpx from abacus_common_logic.connectors.database import db from abacus_common_logic.constants.constants import DATE_FORMAT from marshmallow import ValidationError from owsresponse import response from sqlalchemy import bindparam, select from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import aliased, joinedload from abacus_contract import models from abacus_contract.connectors import ows_abacus_account from abacus_contract.connectors.kafka import emit_contract_event from abacus_contract.constants import constants, error from abacus_contract.logic import contract_exclusion, legacy_contract from abacus_contract.logic.account_contract import create_account_contract from abacus_contract.logic.contract_lifecycle import ( _create_contract_lifecycle, reactivate_contract_lifecycle, terminate_contract_lifecycle, ) from abacus_contract.logic.contract_lifecycle_schedule import ( _create_contract_lifecycle_schedules, _validate_request_payload, ) from abacus_contract.models.contract import Contract from abacus_contract.schemas.contract import ( ContractDetail, ContractDetailSchema, ContractSapFormattedSchema, ContractVatInfoSchema, ) from abacus_contract.utils.features import ( is_abacus_prevent_run_controller_update_enabled, is_abacus_primary_contract_enabled, is_single_supply_chain_company_codes_enabled, ) from abacus_contract.utils.format_error import validation_error from abacus_contract.utils.format_response import ( prepare_dataload_response, prepare_dataload_with_data_as_list_response, ) from core.config import Config from royalties import models as royalties_models from royalties.constants.constants import ( ACCOUNTING_RUN_STATUSES as STATUSES, ) contract_detail_schema = ContractDetailSchema() CONTRACTS_SNAPSHOT_HEADER = ( 'contract_id', 'account_id', 'term_start', 'term_end', 'payee_currency_code', ) logger = logging.getLogger(Config.LOGGER_NAME) def create_contract( contract_name: str, contract_type: str, contract_id: int | None = None, execution_date: date | None = None, reference_signing_entity_id: int | None = None, reference_sap_profit_center_id: int | None = None, term_start: date | None = None, # TODO: remove (deprecated) term_end: date | None = None, # TODO: remove (deprecated) oa_contract_id: int | None = None, account_id: int | None = None, contract_exclusions: dict | None = None, summary_note: str | None = None, general_note: str | None = None, is_excluded_from_accounting_run: bool | None = None, is_primary_contract: bool | None = None, ) -> response.Response: """Create a contract. Args: contract_name (str): name of the contract contract_type (str): type of contract (i.e. 'distribution') contract_id (int): ID of the contract; optional execution_date (date): date on which the contract was countersigned reference_signing_entity_id (int): id of the reference_signing_entity; optional reference_sap_profit_center_id (int): id of the reference_sap_profit_center; optional — see _resolve_reference_sap_profit_center_id for FF-gated defaulting term_start (date): date the contract starts; DEPRECATED term_end (date): date the contract ends; DEPRECATED oa_contract_id (int): ID of legacy contract in OA; optional (legacy only) account_id (int): ID of the account to which the new contract belongs; optional contract_exclusions (dict): countries/stores excluded from delivery; optional summary_note (str): summary notes for the contract general_note (str): general notes for the contract is_excluded_from_accounting_run (bool): is the contract excluded from the accounting run is_primary_contract (bool): is the contract primary Returns: a response.Response """ try: if contract_id and models.Contract.get_by_id(contract_id): raise ValidationError( error.ERROR_CONTRACT_ID_ALREADY_EXISTS.format(contract_id=contract_id) ) if contract_type == constants.CONTRACT_TYPES.LEGACY_DISTRIBUTION: raise ValidationError( error.ERROR_INVALID_CONTRACT_TYPE.format(contract_type=contract_type) ) try: if not _account_exists(account_id): raise ValidationError( error.ERROR_ACCOUNT_NOT_FOUND.format(account_id=account_id) ) except httpx.HTTPError as e: logger.error(f'Failed to get account {account_id}: {e}') return response.create_error_response('error', str(e), 500) is_primary_contract = _can_mark_contract_as_primary( account_id, contract_type, is_primary_contract ) reference_sap_profit_center_id = _resolve_reference_sap_profit_center_id( reference_signing_entity_id, reference_sap_profit_center_id ) new_contract = models.Contract.create( contract_id=contract_id, contract_name=contract_name, contract_type=contract_type, execution_date=execution_date, reference_signing_entity_id=reference_signing_entity_id, reference_sap_profit_center_id=reference_sap_profit_center_id, is_primary_contract=is_primary_contract, term_start=term_start, term_end=term_end, summary_note=summary_note, general_note=general_note, is_excluded_from_accounting_run=is_excluded_from_accounting_run, ) if oa_contract_id: legacy_contract.create_legacy_contract( new_contract.contract_id, oa_contract_id ) else: if not contract_exclusions: contract_exclusions = constants.DEFAULT_CONTRACT_EXCLUSIONS new_contract_exclusion = contract_exclusion.create_contract_exclusions( new_contract.contract_id, contract_exclusions ) if new_contract_exclusion.status != 201: return new_contract_exclusion if account_id: account_contract = create_account_contract( account_id=account_id, contract_id=new_contract.contract_id ) if account_contract.status != 201: message = account_contract.errors and account_contract.errors.get( 'message' ) if not message: message = error.ERROR_FAILURE_TO_CREATE.format('account_contract') raise ValidationError(message) emit_contract_event( new_contract.contract_id, constants.CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED, ) return response.Response( message=contract_detail_schema.dump(new_contract), status=201 ) except ValidationError as e: return validation_error(str(e)) except SQLAlchemyError as e: db.session.rollback() raise e def _account_exists(account_id: int | None) -> bool: """Check if account exists.""" return ( account_id is None or ows_abacus_account.get_account(account_id).status_code == 200 ) def _is_signing_entity_authorized_for_profit_center( reference_signing_entity_id: int, reference_sap_profit_center_id: int, ) -> bool: """Check (SE, PC) pair exists in signing_entity_sap_profit_center (not soft-deleted). The lookup acquires a shared lock on the junction row (``SELECT ... FOR SHARE``) so that a concurrent ``DELETE /signing-entity-profit-center/:id`` cannot soft-delete the mapping between this check and the caller's subsequent contract write. The exclusive UPDATE that the DELETE handler issues will wait until this transaction commits. """ return ( models.SigningEntitySapProfitCenter.query.filter_by( reference_signing_entity_id=reference_signing_entity_id, reference_sap_profit_center_id=reference_sap_profit_center_id, deleted_at=None, ) .with_for_update(read=True) .first() is not None ) def _resolve_pc_update(contract: models.Contract, params: dict) -> None: """Resolve the PC field in a PATCH ``params`` dict in-place. Behavior matrix (FF = SINGLE_SUPPLY_CHAIN_COMPANY_CODES, SE/PC = field present in params): FF | SE | PC | Behavior ---+----+----+---------------------------------------------------------- F | F | F | No-op F | F | T | No-op (strips PC; legacy mode ignores explicit PC) F | T | F | Set PC to the new SE's legacy reference_sap_profit_center_id F | T | T | Strip PC; set PC to the new SE's legacy reference_sap_profit_center_id T | F | F | No-op T | F | T | Validate (current SE, new PC) against the junction T | T | F | ValidationError: PC required when SE changes T | T | T | Validate (new SE, new PC) against the junction """ se_in_params = 'reference_signing_entity_id' in params pc_in_params = 'reference_sap_profit_center_id' in params ## FF OFF if not is_single_supply_chain_company_codes_enabled(): # Legacy mode: PC is derived from SE, never set independently. params.pop('reference_sap_profit_center_id', None) if se_in_params: new_se_id = params['reference_signing_entity_id'] new_se = models.ReferenceSigningEntity.query.get(new_se_id) if new_se is None: raise ValidationError( error.ERROR_REFERENCE_SIGNING_ENTITY_NOT_FOUND.format( reference_signing_entity_id=new_se_id, ) ) params['reference_sap_profit_center_id'] = ( new_se.reference_sap_profit_center_id ) return # FF ON if not pc_in_params: # PC must be explicit if SE is given if se_in_params: raise ValidationError(error.ERROR_REFERENCE_SAP_PROFIT_CENTER_REQUIRED) return # Both PC and SE given. Verify relationship new_se_id = params.get( 'reference_signing_entity_id', contract.reference_signing_entity_id ) new_pc_id = params['reference_sap_profit_center_id'] if not _is_signing_entity_authorized_for_profit_center(new_se_id, new_pc_id): raise ValidationError( error.ERROR_INVALID_SIGNING_ENTITY_PROFIT_CENTER_PAIR.format( reference_signing_entity_id=new_se_id, reference_sap_profit_center_id=new_pc_id, ) ) def _resolve_reference_sap_profit_center_id( reference_signing_entity_id: int | None, reference_sap_profit_center_id: int | None, ) -> int | None: """Resolve and validate the profit center for a new contract per SINGLE_SUPPLY_CHAIN_COMPANY_CODES. - PC omitted + FF ON: raise (explicit selection required). - PC omitted + FF OFF: default to the SE's legacy reference_sap_profit_center_id. - The resolved (SE, PC) pair must exist in signing_entity_sap_profit_center. """ if reference_sap_profit_center_id is None: if is_single_supply_chain_company_codes_enabled(): raise ValidationError(error.ERROR_REFERENCE_SAP_PROFIT_CENTER_REQUIRED) signing_entity = models.ReferenceSigningEntity.query.get( reference_signing_entity_id ) if signing_entity is None: return None reference_sap_profit_center_id = signing_entity.reference_sap_profit_center_id if not _is_signing_entity_authorized_for_profit_center( reference_signing_entity_id, reference_sap_profit_center_id ): raise ValidationError( error.ERROR_INVALID_SIGNING_ENTITY_PROFIT_CENTER_PAIR.format( reference_signing_entity_id=reference_signing_entity_id, reference_sap_profit_center_id=reference_sap_profit_center_id, ) ) return reference_sap_profit_center_id def _can_mark_contract_as_primary( account_id: int, contract_type: str, is_primary_contract: bool, contract_id: int | None = None, ) -> bool: """Check if the contract can be designated as primary. Args: account_id (int): id of the account to which the new contract belongs contract_type (str): type of contract (i.e. 'distribution') is_primary_contract (bool): is the contract primary contract_id (int): id of the contract, or None if creating a new contract Returns: a boolean value """ if not is_primary_contract: return False # TODO remove ABACUS_PRIMARY_CONTRACT feature flag condition and # related constant variable if is_abacus_primary_contract_enabled() is False: return False if contract_type != constants.CONTRACT_TYPES.DISTRIBUTION: raise ValidationError(error.ERROR_ONLY_DISTRIBUTION_PRIMARY_CONTRACT) contract = models.Contract.get_primary_contract_by_account_id_and_contract_type( account_id, contract_type ) if contract and contract.contract_id != contract_id: raise ValidationError( error.ERROR_PRIMARY_CONTRACT_ALREADY_EXISTS.format( account_id, contract_type ) ) return True def update_contract(contract: models.Contract, **params) -> response.Response: """Update contract's fields. contract(models.Contract): Contract model object params(dict): PUT request body - Optional fields: - contract_name: name of the contract - sap_created_at: date when contract data is sent to SAP - reference_signing_entity_id: id of the reference_signing_entity - term_start: term start date of the contract; DEPRECATED - term_end: term end date of the contract; DEPRECATED - summary_note: summary notes for the contract - general_note: general notes for the contract - is_excluded_from_accounting_run: whether a contract should be used during the accounting run calculation - is_paythrough_contract: whether a contract is a paythrough contract - execution_date: date on which the contract was countersigned - initial_start_date: date on which the contract was first activated - run_controller_id: id of the run controller associated with the contract - is_primary_contract (bool): is the contract primary Returns: updated contract data """ try: rc_id = params.pop('run_controller_id', None) siblings = [] if rc_id is not None: siblings = _update_run_controller_and_sibling_contracts(rc_id, contract) is_primary_contract = params.get('is_primary_contract') account_id = contract.account_contract.account_id contract_type = contract.contract_type contract_id = contract.contract_id if is_abacus_primary_contract_enabled(): is_primary_contract = _can_mark_contract_as_primary( account_id, contract_type, is_primary_contract, contract_id ) params.update({'is_primary_contract': is_primary_contract}) elif 'is_primary_contract' in params: del params['is_primary_contract'] _resolve_pc_update(contract, params) contract.update_attributes(**params) models.Contract.commit_changes() except ValidationError as e: return validation_error(str(e)) except SQLAlchemyError as e: db.session.rollback() raise e emit_contract_event( contract.contract_id, constants.CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) for sibling in siblings: emit_contract_event( sibling.contract_id, constants.CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) return response.Response(message=contract_detail_schema.dump(contract), status=200) def _update_run_controller_and_sibling_contracts( run_controller_id: int, contract: models.Contract, ) -> list[Contract]: """Update run controller for a contract and its sibling contracts.""" existing_run_controller_id = contract.run_controller_contract.run_controller_id _can_update_run_controller( contract.contract_type, existing_run_controller_id, run_controller_id ) siblings: list[models.Contract] = ( models.Contract.get_filtered_query( account_ids=[contract.account_id], contract_type=contract.contract_type, ) .filter(models.Contract.contract_id != contract.contract_id) .all() ) for c in [contract] + siblings: if c.run_controller_contract is not None: c.run_controller_contract.update_attributes( run_controller_id=run_controller_id ) else: c.run_controller_contract = royalties_models.RunControllerContract( run_controller_id=run_controller_id ) return siblings def _can_update_run_controller( contract_type: str, existing_run_controller_id: int, new_run_controller_id: int ): """Check whether the run controller can be updated for the contract. Args: contract_type (str): type of contract. existing_run_controller_id (int): id of the run controller currently associated with the contract new_run_controller_id (int): id of the run controller to be assigned to the contract """ if new_run_controller_id == existing_run_controller_id: return existing_run_controller = royalties_models.RunController.get_by_id_or_error( existing_run_controller_id ) new_run_controller = royalties_models.RunController.get_by_id_or_error( new_run_controller_id ) new_run_controller_name = new_run_controller.run_controller_name existing_run_controller_name = existing_run_controller.run_controller_name if contract_type != new_run_controller.contract_type: raise ValidationError( error.ERROR_CONTRACT_AND_RUN_CONTROLLER_TYPE_MISMATCH.format( contract_type=contract_type, ) ) if is_abacus_prevent_run_controller_update_enabled() is False: return statuses = [ STATUSES.WAITING_TO_RUN, STATUSES.RUNNING, STATUSES.ERROR, STATUSES.COMPLETE, STATUSES.COMMITTING, STATUSES.COMMITTED, ] current_statement_period = ( royalties_models.StatementPeriod.get_current_statement_period() ) if current_statement_period is None: return open_accounting_period = royalties_models.AccountingPeriod.get_current_period( current_statement_period.statement_period_id, contract_type ) if open_accounting_period is None: return from_accounting_runs = ( royalties_models.AccountingRun.get_by_accounting_period_and_run_controller( open_accounting_period.accounting_period_id, existing_run_controller_id, ) ) is_from_accounting_run_in_process = any( [ accounting_run.run_status in statuses for accounting_run in from_accounting_runs ] ) to_accounting_runs = ( royalties_models.AccountingRun.get_by_accounting_period_and_run_controller( open_accounting_period.accounting_period_id, new_run_controller_id ) ) is_to_accounting_run_in_process = any( [accounting_run.run_status in statuses for accounting_run in to_accounting_runs] ) if is_from_accounting_run_in_process or is_to_accounting_run_in_process: accounting_run = ( existing_run_controller_name if is_from_accounting_run_in_process else new_run_controller_name ) raise ValidationError( error.ERROR_CAN_NOT_UPDATE_RUN_CONTROLLER.format(accounting_run) ) def get_contracts_by_ids(contract_ids: list[int]) -> list[ContractDetail]: """Get contracts by their identifiers.""" contracts = models.Contract.get_by_ids(contract_ids) return ContractDetailSchema(many=True).dump(contracts) def format_contracts_for_dataloader( contract_ids: list[int], contracts: list[ContractDetail] ) -> response.Response: """Get contracts by their identifiers with dataload format.""" result = prepare_dataload_response(contract_ids, contracts, 'contract_id') return response.Response(message=result, status=200) def get_contracts_by_account(account_ids: list, dataload: bool = False): """Get contracts associated to the specified accounts.""" contracts = models.Contract.get_by_accounts(account_ids) contracts_items = ContractDetailSchema(many=True).dump(contracts) if dataload: message = prepare_dataload_with_data_as_list_response( account_ids, contracts_items, 'account_id', ) else: message = {'items': contracts_items, 'total_count': len(contracts)} return response.Response(message=message, status=200) def get_contracts_by_oa_contract_ids(oa_contract_ids): """Get contracts related to specified oa_contract_ids.""" contracts = models.Contract.get_by_legacy_contract_ids(oa_contract_ids) message = ContractDetailSchema(many=True).dump(contracts) return response.Response(message=message, status=200) def get_vat_info_by_contract_ids(contract_ids): """GET contracts vat info by list of contract ids.""" vat_info = models.Contract.get_contract_vat_info_by_contract_ids(contract_ids) message = ContractVatInfoSchema(many=True).dump(vat_info) return response.Response(message=message, status=200) def contract_export(contract_ids=None): """Get contract terms as csv data.""" yield '\t'.join(CONTRACTS_SNAPSHOT_HEADER) + '\n' for item in models.Contract.stream_all(contract_ids=contract_ids): yield build_contract_row(item) def build_contract_row(item): """Format a contract row.""" return ( '\t'.join( [ str(item.contract_id), str(item.account_id), item.term_start.strftime(DATE_FORMAT) if item.term_start else '', item.term_end.strftime(DATE_FORMAT) if item.term_end else '', item.currency_code, ] ) + '\n' ) def _create_contract(post_request_payload: dict, initial_start_date: str) -> object: """Create contract, account_contract and legacy_contract. Args: post_request_payload (dict): POST request payload having below fields - contract_id (int): id of the contract - contract_name (str): name of the contract - contract_type (str): type of the contract either distribution or neighbouring_rights - execution_date (date): date on which the contract was countersigned - reference_signing_entity_id (int): id of the reference_signing_entity - oa_contract_id (int): id of the legacy contract in OA; optional (legacy only) - account_id (int): id of the account to which the new contract belongs - summary_note (str): summary notes for the contract - general_note (str): general notes for the contract - contract_exclusions (dict): countries/stores excluded from delivery - is_primary_contract (bool): is the contract primary initial_start_date (str): start date of the contract Returns: a contract record """ reference_signing_entity_id = post_request_payload.get( 'reference_signing_entity_id' ) reference_sap_profit_center_id = post_request_payload.get( 'reference_sap_profit_center_id' ) account_id = post_request_payload.get('account_id') contract_id = post_request_payload.get('contract_id', None) contract_name = post_request_payload.get('contract_name') contract_type = post_request_payload.get('contract_type') execution_date = post_request_payload.get('execution_date', None) summary_note = post_request_payload.get('summary_note', None) general_note = post_request_payload.get('general_note', None) oa_contract_id = post_request_payload.get('oa_contract_id', None) contract_exclusions = post_request_payload.get('contract_exclusions', None) is_primary_contract = post_request_payload.get('is_primary_contract', False) if contract_id and models.Contract.get_by_id(contract_id): raise ValidationError( error.ERROR_CONTRACT_ID_ALREADY_EXISTS.format(contract_id=contract_id) ) is_primary_contract = _can_mark_contract_as_primary( account_id, contract_type, is_primary_contract ) reference_sap_profit_center_id = _resolve_reference_sap_profit_center_id( reference_signing_entity_id, reference_sap_profit_center_id ) new_contract = models.Contract.build( contract_id=contract_id, contract_name=contract_name, contract_type=contract_type, execution_date=execution_date, reference_signing_entity_id=reference_signing_entity_id, reference_sap_profit_center_id=reference_sap_profit_center_id, summary_note=summary_note, general_note=general_note, initial_start_date=initial_start_date, is_primary_contract=is_primary_contract, ) db.session.flush() contract_id = new_contract.contract_id models.AccountContract.build(account_id=account_id, contract_id=contract_id) db.session.flush() if oa_contract_id: models.LegacyContract.build( contract_id=contract_id, oa_contract_id=oa_contract_id ) else: if not contract_exclusions: contract_exclusions = constants.DEFAULT_CONTRACT_EXCLUSIONS models.ContractExclusion.build( contract_id=contract_id, exclusions=contract_exclusions ) return new_contract def create_contract_with_lifecycle_and_schedules( contract: dict, contract_lifecycle: dict, contract_lifecycle_schedules: list ) -> response.Response: """Create a contract with contract lifecycle and schedules. Args: contract (dict): contract payload contract_lifecycle (dict): contract_lifecycle payload contract_lifecycle_schedules (list): contract_lifecycle_schedule payload Returns: a contract with contract_lifecycle and contract_lifecycle_schedules """ lifecycle_term_start = contract_lifecycle.get('lifecycle_term_start') try: contract_type = contract.get('contract_type') if contract_type == constants.CONTRACT_TYPES.LEGACY_DISTRIBUTION: raise ValidationError( error.ERROR_INVALID_CONTRACT_TYPE.format(contract_type=contract_type) ) new_contract = _create_contract(contract, lifecycle_term_start) contract_id = new_contract.contract_id contract_type = new_contract.contract_type _validate_request_payload( contract_id, contract_type, [], contract_lifecycle_schedules ) new_contract_lifecycle_schedules = _create_contract_lifecycle_schedules( contract_id, contract_lifecycle_schedules ) db.session.flush() _create_contract_lifecycle( contract_id, lifecycle_term_start, new_contract_lifecycle_schedules[0] ) db.session.flush() except ValidationError as e: db.session.rollback() return validation_error(str(e)) except SQLAlchemyError as e: db.session.rollback() raise e except Exception as e: db.session.rollback() raise e db.session.commit() emit_contract_event( new_contract.contract_id, constants.CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) return response.Response( message=ContractDetailSchema().dump(new_contract), status=201 ) def terminate_contract( contract_id: int, termination_effective: date, termination_notice_received: date ): """Terminate a contract. Args: contract_id (int): ID of the contract termination_effective (date): date of termination termination_notice_received (date): date of notice received Returns: a contract record """ contract = models.Contract.get_by_id_or_error(contract_id) try: terminate_contract_lifecycle( contract_id, termination_effective, termination_notice_received, ) except ValidationError as e: return validation_error(str(e)) except SQLAlchemyError as e: db.session.rollback() raise e return response.Response(message=contract_detail_schema.dump(contract), status=200) def reactivate_contract(contract_id: int) -> response.Response: """Reactivate a contract by id. Args: contract_id (int): id of the contract Returns: a contract record """ try: contract = models.Contract.get_by_id_or_error(contract_id) reactivate_contract_lifecycle(contract_id) except ValidationError as e: return validation_error(str(e)) return response.Response(message=contract_detail_schema.dump(contract), status=200) def get_account_id_map_by_contract_ids(contract_ids: list[int]) -> dict[int, int]: """Map each contract id to its account id, for the batch dataloader authz step. Assumes one account_contract per contract (the relationship is 1:1); if a contract ever mapped to more than one, the last row would win. """ if not contract_ids: return {} AC = aliased(models.AccountContract, name='ac') query = ( select(AC.contract_id, AC.account_id) .select_from(AC) .where(AC.contract_id.in_(bindparam('ids', expanding=True))) ) rows = db.session.execute(query, {'ids': list(contract_ids)}).all() return {contract_id: account_id for contract_id, account_id in rows} def get_account_ids_by_contract_ids(contract_ids: list[int]) -> list[int]: """Get account ids for the given contract ids, in request order. Derived from get_account_id_map_by_contract_ids so the account-lookup query lives in one place; ids that resolve to no account are dropped. Args: contract_ids (list[int]): list of contract ids Returns: list[int]: account ids for the resolvable contracts, in request order """ account_by_id = get_account_id_map_by_contract_ids(contract_ids) return [ account_by_id[contract_id] for contract_id in contract_ids if account_by_id.get(contract_id) is not None ] def get_account_id_by_contract_id(contract_id: int) -> int | None: """Get account id by contract id. Args: contract_id (int): id of the contract Returns: int: id of the account None: if no contract is found """ contract = ( db.session.query(models.Contract) .options(joinedload(models.Contract.account_contract)) .filter(models.Contract.contract_id == contract_id) .one_or_none() ) if contract and contract.account_contract: return contract.account_contract.account_id return None def sap_details(contract_id) -> response.Response: """Return formatted contract data for SAP using schema.""" sap_details = models.Contract.get_sap_profit_center_by_contract_id( contract_id=contract_id ) # Serialize with schema message = ContractSapFormattedSchema().dump(sap_details) return response.Response(message=message, status=200) def can_contract_be_deleted(contract_id: int) -> response.Response: """Check if a contract can be deleted.""" return response.Response( message={'can_be_deleted': models.Contract.can_be_deleted(contract_id)}, status=200, ) def get_can_be_deleted_records_by_contract_ids(authorized_contract_ids: list) -> list: """Flat {contract_id, can_be_deleted} records for the authorized contract ids.""" if not authorized_contract_ids: return [] can_delete = models.Contract.can_be_deleted_by_ids(authorized_contract_ids) return [ {'contract_id': cid, 'can_be_deleted': value} for cid, value in can_delete.items() ] def delete_contract(contract_id: int) -> response.Response: """Delete a contract.""" can_be_deleted = models.Contract.can_be_deleted(contract_id) if not can_be_deleted: raise ValidationError( error.ERROR_CONTRACT_CANNOT_BE_DELETED.format(contract_id=contract_id) ) models.Contract.delete(contract_id) return response.Response(message={'deleted': True}, status=200)