"""Logic functions for payment holds.""" from ows_accounting import response from ows_accounting.constants import error from ows_accounting.constants import payment_holds as constants from ows_accounting.models import ows_account from ows_accounting.models import vendor_payment_hold def _format_list_response(holds, total_records=None, limit=None, offset=None): """Format list of holds to have pagination with items of data. Args: holds (list): list of hold objects. total_records (int): total records without limit. limit (int): limit number of result records. offset (int): offset start of result records. Returns: response: Formatted list of hold objects. """ if not limit and not offset: return response.Response(message={ 'pagination': { 'type': 'none', 'total_records': len(holds) }, 'items': holds }) if not total_records: total_records = len(holds) if not offset: offset = 0 if not limit: limit = total_records offset = 0 return response.Response(message={ 'pagination': { 'type': 'standard', 'offset': offset, 'limit': limit, 'total_records': total_records }, 'items': holds }) def get_all_active_holds(limit=None, offset=None, vendor_ids=None): """Get all active holds for all vendors. Args: limit (int): limit number of result records. Optional param. offset (int): offset start of result records. Optional param. vendor_ids (set): vendor identifier list. Optional param. Returns: response: List of hold objects. """ total_records = vendor_payment_hold.count_holds_by_status( status='active', vendor_ids=vendor_ids) if not total_records: return total_records if total_records.message == 0: return _format_list_response([], total_records.message, limit, offset) # get holds data only if we have holds. holds = vendor_payment_hold.get_all_holds_by_status( status='active', limit=limit, offset=offset, vendor_ids=vendor_ids) if not holds: return holds return _format_list_response( holds.message, total_records.message, limit, offset) def update_hold_by_id(hold_id, user_id, status, description=''): """Edit hold by id with changes in data. Args: hold_id (int): hold identifier. user_id (str): Oa user identifier, that is editing the hold. status (str): Hold status. description (str): description for new hold. (Optional) Returns: response: Updated hold object. """ if not user_id or not status: return response.create_error_response( error.ERROR_CODE_INVALID_REQUEST, constants.MISSING_REQUIRED_PARAM) return vendor_payment_hold.update_hold_by_id( hold_id=hold_id, editor_id=user_id, status=status, description=description) def get_holds_by_id(hold_id): """Get a hold by id. Args: hold_id (int): hold identifier. Returns: response: hold object with logs. """ return vendor_payment_hold.get_hold_by_id(hold_id) def get_holds_by_vendor_id(vendor_id): """Get all active & inactive holds for vendor_id. Args: vendor_id (int): vendor identifier. Returns: response: List of hold objects. """ validate_vendor_id = ows_account.check_vendor_exists(vendor_id) if not validate_vendor_id: return validate_vendor_id holds = vendor_payment_hold.get_all_holds_by_vendor(vendor_id) # when getting holds for a vendor its ok, to not have holds. if holds.status == 404: return _format_list_response([]) if not holds: return holds return _format_list_response(holds.message) def create_hold_for_vendor(vendor_id, user_id, status, description): """Create a new hold for a vendor. Args: vendor_id (int): vendor identifier. user_id (str): Oa user identifier, that is editing the hold. status (str): Hold status. description (str): description for new hold. (Optional) Returns: flask.response: New holds object. """ validate_vendor_id = ows_account.check_vendor_exists(vendor_id) if not validate_vendor_id: return validate_vendor_id existing_holds = vendor_payment_hold.get_active_holds_for_vendor(vendor_id) if existing_holds: return response.create_error_response( error.ERROR_CODE_INVALID_REQUEST, constants.EXISTING_ACTIVE_HOLD) return vendor_payment_hold.create_hold_for_vendor( vendor_id=vendor_id, creator_id=user_id, status=status, description=description)