""" Celery tasks for bulk operations ================================ """ from collections import defaultdict from itertools import groupby from operator import itemgetter from application import celery from botocore.exceptions import ClientError from celery import chord from celery.exceptions import MaxRetriesExceededError from flask_celeryext import RequestContextTask from oto import response from masters_registry.connectors import sentry from masters_registry.constant import bulk_tasks_const from masters_registry.constant import db_const from masters_registry.constant import error from masters_registry.constant import field_const from masters_registry.constant import opcode_const from masters_registry.features import OwsFeaturesError from masters_registry.logic import bulk_tasks as bulk_tasks_logic from masters_registry.logic import dms_carveout from masters_registry.logic import locks from masters_registry.logic import masters_registry as mrl from masters_registry.logic import ownership as ownership_logic from masters_registry.models import bulk_tasks from masters_registry.models import ownership from masters_registry.models import ows_conflict_manager from masters_registry.models import yt_ownership from masters_registry.utils import retry from masters_registry.utils import RetryCountExceededError def generate_reports(task_id, report_types, correlation_id=None): """Generates CSV reports and uploads them to S3 Args: task_id (int): taks id report_types (list): list of report types correlation_id (str): correlation ID for logs """ try: for report_type in report_types: bulk_tasks_logic.generate_report( task_id, correlation_id, import_report_type=report_type, ) except (ClientError, OwsFeaturesError): if sentry.sentry_client: sentry.sentry_client.captureException() @celery.task(name='masters_registry.tasks.bulk.test_celery_task') def test_celery_task(correlation_id): """Test task for test purposes Args: correlation_id (str): unique ID Returns: str: "{correlation_id}: test passed." """ result = '{0}: test passed.'.format(correlation_id) return result def update_active_for_deleted_upc(ownership_info, tuid, correlation_id, user): """Function for updating ownership info in active table Args: ownership_info (dict): active table record or empty dict tuid (int): track unique identifier correlation_id (str): correlation id for logging user (str): Orchard user id Returns: dict: containing active table record, removed territories list and failed flag """ result = { field_const.OWNERSHIP_INFO: ownership_info, field_const.REMOVED_TERRITORIES: [], field_const.RESOLVED_CONFLICT: False, field_const.WARNING: False, } if bool(ownership_info): isrc = ownership_info[field_const.ISRC] territories = ownership_info[field_const.TERRITORIES] to_remove = [] for territory, territory_info in territories.items(): tuids = ownership_logic._get_tuids_from_territory(territory_info) if tuid in tuids: to_remove.append(territory) if to_remove: result[field_const.REMOVED_TERRITORIES] = sorted(to_remove) remove_result = ownership_logic.remove_ownership( isrc, tuid, to_remove, territories, correlation_id, user, daemon=True, source=field_const.BULK_UPDATE_UPC) was_conflict_resolved = bool( remove_result[field_const.RESOLVED_CONFLICT]) result[field_const.RESOLVED_CONFLICT] = was_conflict_resolved result[field_const.WARNING] = was_conflict_resolved return result def update_yt_for_deleted_upc(update_audit_result, isrc, correlation_id): """Function for updating youtube ownership Args: update_audit_result (dict): containing active table record, removed territories list and failed flag isrc (str): international standard recording code correlation_id (str): correlation id for logging Returns: dict: result of import """ ownership_info = update_audit_result[field_const.OWNERSHIP_INFO] removed_territories = update_audit_result[field_const.REMOVED_TERRITORIES] error_message = '' success = True if removed_territories: claimed = set(ownership_info[field_const.TERRITORIES].keys()) updated_claimed_territories = claimed.difference( set(removed_territories)) if len(updated_claimed_territories) > 0: store_carveout = dms_carveout.ownership_check_dms_carveout( ownership_info, updated_claimed_territories, correlation_id) if not store_carveout: success = False error_message = error.FAILED_TO_GET_DMS_CARVEOUTS elif store_carveout.message: yt_ownership.send_message( isrc, store_carveout.message, correlation_id) else: success = False error_message = error.UPC_YOUTUBE_CARVED_OUT_MESSAGE else: yt_ownership.send_message( isrc, updated_claimed_territories, correlation_id) return success, error_message @celery.task( name='masters_registry.tasks.bulk.bulk_import_upcs', base=RequestContextTask) def bulk_import_upcs(upcs, task_id, correlation_id, user): """Task for bulk import ownership data into the registry for all given UPCs Args: upcs (list(str)): list of UPCs to import task_id (int): task id correlation_id (str): correlation id for logging user (str): Orchard user id Returns: AsyncTaskResult: Celery task result """ upcs_isrcs_data = ownership.get_isrcs(upcs).message upcs_isrcs = {} deleted_upcs_isrcs = {} for upc, info in upcs_isrcs_data.items(): if info[0][field_const.DELETIONS] == db_const.DELETION_YES: deleted_upcs_isrcs[upc] = info else: upcs_isrcs[upc] = info upc_import_tasks = [] for upc, isrc_info in deleted_upcs_isrcs.items(): for info in isrc_info: isrc, tuid, vendor_id = itemgetter( field_const.ISRC, field_const.TUID, field_const.VENDOR_ID)(info) upc_import_tasks.append( update_isrc_for_deleted_upc.s( upc, isrc, tuid, vendor_id, user, correlation_id) ) for upc, isrc_info in upcs_isrcs.items(): carve_in_territories = mrl._get_carved_in_territories( upc, correlation_id) for isrc in isrc_info: upc_import_tasks.append( update_isrc_for_upc.s( upc, isrc, carve_in_territories, user, correlation_id) ) save_results_task = save_bulk_import_report.s( task_id, len(upc_import_tasks), user, correlation_id) bulk_import_result = chord(upc_import_tasks, save_results_task)() return bulk_import_result @celery.task( bind=True, max_retries=3, name='masters_registry.tasks.bulk.update_isrc_for_upc', base=RequestContextTask) def update_isrc_for_upc( self, upc, isrc_info, carve_in_territories, user, correlation_id): """Task for updating single ISRC Args upc (str): UPC isrc_info (dict): track information carve_in_territories: user (str): Orchard user id correlation_id (str): correlation id for logging Returns: dict: result of ISRC update """ isrc = isrc_info.get(field_const.ISRC) tuid = isrc_info.get(field_const.TUID) vendor_id = isrc_info.get(field_const.VENDOR_ID) failed_response = { field_const.UPC: upc, field_const.ISRC: isrc, field_const.TUID: tuid, field_const.VENDOR_ID: vendor_id, field_const.SUCCESS: False, field_const.ERROR_MESSAGE: error.TERRITORIES_CARVE_IN_NOT_FOUND } if not carve_in_territories: return failed_response if not isrc: failed_response[field_const.ERROR_MESSAGE] = error.NO_ISRC_ERROR return failed_response try: territories = carve_in_territories.message isrc_result = mrl._process_isrc_info( isrc_info, territories, correlation_id, user) isrc_result[field_const.UPC] = upc isrc_result[field_const.TUID] = tuid isrc_result[field_const.VENDOR_ID] = vendor_id return isrc_result except (ClientError, OwsFeaturesError): if sentry.sentry_client: sentry.sentry_client.captureException() try: self.retry() except MaxRetriesExceededError: failed_response[field_const.ERROR_MESSAGE] = \ error.ISRC_IMPORT_ERROR return failed_response except KeyError as e: if sentry.sentry_client: sentry.sentry_client.captureException() failed_response[field_const.ERROR_MESSAGE] = \ str(e) return failed_response @celery.task( bind=True, max_retries=3, name='masters_registry.tasks.bulk.update_isrc_for_deleted_upc', base=RequestContextTask) def update_isrc_for_deleted_upc( self, upc, isrc, tuid, vendor_id, user, correlation_id): """Task for updating single ISRC for deleted UPC Args: upc (str): UPC isrc (str): international standard recording code tuid (int): track unique identifier vendor_id (int): label id user (str): Orchard user id correlation_id (str): correlation id for logging Returns: dict: result of ISRC update """ result = { field_const.ISRC: isrc, field_const.UPC: upc, field_const.TUID: tuid, field_const.VENDOR_ID: vendor_id, } try: ownership_info = ownership.get_ownership(isrc).message update_result = update_active_for_deleted_upc( ownership_info, tuid, correlation_id, user) success, error_message = update_yt_for_deleted_upc( update_result, isrc, correlation_id) result[field_const.SUCCESS] = success result[field_const.ERROR_MESSAGE] = error_message update_result.pop(field_const.OWNERSHIP_INFO) result.update(update_result) except (ClientError, OwsFeaturesError): if sentry.sentry_client: sentry.sentry_client.captureException() try: self.retry() except MaxRetriesExceededError: result[field_const.SUCCESS] = False result[field_const.ERROR_MESSAGE] = error.DELETED_ISRC_UPDATE_ERROR return result @celery.task( bind=True, name='masters_registry.tasks.bulk.save_bulk_import_report', base=RequestContextTask, max_retries=3, soft_time_limit=bulk_tasks_const.CELERYD_TASK_TIME_LIMIT) def save_bulk_import_report( self, import_results, task_id, tasks_count, user, correlation_id): """Generate and save bulk UPC import report Args: import_results (list): import reports for each ISRC task_id (int): task id tasks_count (int): number of tasks user (str): Orchard user id correlation_id (str): correlation ID for logs """ import_results.sort(key=itemgetter(field_const.UPC)) import_upc_report = [] for upc, upc_results in groupby( import_results, key=itemgetter(field_const.UPC)): upc_results = list(upc_results) success_count = len( [isrc for isrc in upc_results if isrc[field_const.SUCCESS]]) warning_count = len( [isrc for isrc in upc_results if ( isrc.get(field_const.WARNING) and not isrc.get(field_const.RESOLVED_CONFLICT))]) error_count = len(upc_results) - success_count import_upc_report.append({ field_const.UPC: upc, field_const.ERROR_COUNT: error_count, field_const.SUCCESS_COUNT: success_count, field_const.WARNING_COUNT: warning_count, field_const.STATUS_REPORT: upc_results }) status = _make_upc_report_status(import_upc_report) r = bulk_tasks.update_task(task_id, status, import_upc_report) if not r and r.errors[field_const.CODE] != error.INVALID_BULK_STATUS: try: self.retry() except MaxRetriesExceededError: if sentry.sentry_client: sentry.sentry_client.captureMessage( 'Failed to save BULK task report TaskID: {0},' .format(task_id), extra={'report': import_upc_report} ) generate_reports( task_id, [field_const.ISRC, field_const.UPC], correlation_id) def _make_upc_report_status(upc_statuses): """Determine what status should be written to UPC report. Args: upc_statuses (list(dict)): results of celery tasks Returns: str: UPC import status """ status = bulk_tasks_const.PARTIAL_FAIL_STATUS all_success = all( upc[field_const.ERROR_COUNT] == 0 for upc in upc_statuses) if all_success: status = bulk_tasks_const.DONE_STATUS all_failed = all( upc[field_const.SUCCESS_COUNT] == 0 for upc in upc_statuses) if all_failed: status = bulk_tasks_const.FAILED_STATUS any_warning = any( upc[field_const.WARNING_COUNT] != 0 for upc in upc_statuses) if any_warning: status = bulk_tasks_const.WARNING_STATUS return status @celery.task( bind=True, name='masters_registry.tasks.bulk.save_bulk_lock_report', max_retries=3, soft_time_limit=bulk_tasks_const.CELERYD_TASK_TIME_LIMIT, base=RequestContextTask) def save_bulk_lock_report( self, lock_isrc_results, task_id, territories, reason, lock_tasks_amount): """Generate and save bulk lock report Args: lock_isrc_results (list(tuple)): list of results from update_active_record task. Each tuple contains ISRC and oto.response with details task_id (int): task id territories (list(str)): list of ISO-3166-1 territories reason (str): reason for locking lock_tasks_amount (int): number of lock_isrc tasks """ lock_result = make_lock_result_internal_conflict( lock_isrc_results, territories, reason) if not lock_result['failed_isrcs']: status = bulk_tasks_const.DONE_STATUS elif not lock_result['successful_isrcs']: status = bulk_tasks_const.FAILED_STATUS else: status = bulk_tasks_const.PARTIAL_FAIL_STATUS r = bulk_tasks.update_task(task_id, status, lock_result) if not r and r.errors[field_const.CODE] != error.INVALID_BULK_STATUS: try: self.retry() except MaxRetriesExceededError: if sentry.sentry_client: sentry.sentry_client.captureMessage( 'Failed to save BULK task report TaskID: {0},' .format(task_id), extra={'report': lock_result} ) generate_reports(task_id, [field_const.ISRC]) def make_lock_result_internal_conflict(lock_isrc_results, territories, reason): """Make data for report to save in TaskStatus. This version accounts for internal conflict case. Args: lock_isrc_results (list(tuple(str, oto.repsonse)): list of results from update_active_record task. territories (list(str)): list of ISO-3166-1 territories reason (str): reason for locking """ lock_result = { 'failed_isrcs': False, 'successful_isrcs': False, 'isrcs': defaultdict(dict), 'reason': reason, } for isrc, task_result in lock_isrc_results: if task_result: lock_result['isrcs'][isrc]['territories'] = territories[:] lock_result['isrcs'][isrc]['failed_territories'] = [] lock_result['successful_isrcs'] = True continue lock_result['isrcs'][isrc] = task_result.errors['message'] lock_result['failed_isrcs'] = True if task_result.errors['message']['territories']: lock_result['successful_isrcs'] = True lock_result['isrcs'] = dict(lock_result['isrcs']) return lock_result @celery.task( bind=True, max_retries=3, name='masters_registry.tasks.bulk.unlock_isrc', base=RequestContextTask) def unlock_isrc(self, isrc, territories, correlation_id, user): """Task for performing unlock for single ISRC Args: isrc (str): international standard recording code territories (list): list of territories to unlock correlation_id (str): correlation id for logging user (str): The user id for audit table logging Returns: str, bool: ISRCS, if unlock was successful """ try: locks.unlock_isrc(isrc, territories, correlation_id, user) return isrc, True except (ClientError, OwsFeaturesError): if sentry.sentry_client: sentry.sentry_client.captureException() try: self.retry() except MaxRetriesExceededError: return isrc, False @celery.task( bind=True, name='masters_registry.tasks.bulk.save_bulk_unlock_report', base=RequestContextTask, max_retries=3, soft_time_limit=bulk_tasks_const.CELERYD_TASK_TIME_LIMIT) def update_bulk_unlock_status( self, unlock_isrc_results, task_id, unlock_tasks_amount): """Updates bulk unlock action status Args: unlock_isrc_results (list): task_id (int): task id unlock_tasks_amount (int): number of unlock_isrc tasks """ successful = [isrc for isrc, result in unlock_isrc_results if result] failed = [isrc for isrc, result in unlock_isrc_results if not result] if not failed: status = bulk_tasks_const.DONE_STATUS elif not successful: status = bulk_tasks_const.FAILED_STATUS else: status = bulk_tasks_const.PARTIAL_FAIL_STATUS r = bulk_tasks.update_task(task_id, status, None) if not r and r.errors[field_const.CODE] != error.INVALID_BULK_STATUS: try: self.retry() except MaxRetriesExceededError: if sentry.sentry_client: sentry.sentry_client.captureMessage( 'Failed to save BULK task report TaskID: {0},' .format(task_id) ) @celery.task( name='masters_registry.tasks.bulk.bulk_lock_isrcs', base=RequestContextTask) def bulk_lock_isrcs( lock_reason, isrcs, territories_to_lock, correlation_id, user, task_id): """Task for lock a list of ISRC/territories Args: lock_reason (str): reason for locking isrcs (list(str)): list of international standard recording codes territories_to_lock (list(str)): list of ISO-3166-1 correlation_id (str): The correlation id for logging the microservice instance user (str): The user id for audit table logging task_id (int): bulk lock report id """ active_isrcs = ownership.get_existing_isrcs_in_active_table(isrcs) to_remove, to_unlock, to_lock = locks._check_territories( active_isrcs, territories_to_lock, lock_reason) active_records = locks._lock_create_active_table_records( to_remove, to_unlock, to_lock, lock_reason) active_isrcs = { isrc_data[field_const.ISRC]: isrc_data for isrc_data in active_isrcs } lock_tasks = [] for record in active_records: isrc = record[field_const.ISRC] isrc_record = active_isrcs[isrc] isrc_task = lock_isrc.s( isrc, isrc_record, record, to_remove[isrc], to_unlock[isrc], to_lock[isrc], correlation_id, user, territories_to_lock) lock_tasks.append(isrc_task) if not lock_tasks: result = [(isrc, True) for isrc in isrcs] save_bulk_lock_report.delay( result, task_id, territories_to_lock, lock_reason, len(lock_tasks)) else: chord( lock_tasks, save_bulk_lock_report.s( task_id, territories_to_lock, lock_reason, len(lock_tasks)) ).delay() @celery.task( bind=True, max_retries=3, name='masters_registry.tasks.bulk.lock_isrc', base=RequestContextTask) def lock_isrc( self, isrc, isrc_record, active_record, to_remove, to_unlock, to_lock, correlation_id, user, lock_territories=None): """Task for performing lock for single ISRC Args: isrc (str): international standard recording code isrc_record (dict): ISRC active table record active_record (dict): records for updating active table to_remove (list): list of territories to remove to_unlock (list): list of territories to unlock to_lock (list): list of territories to lock correlation_id (str): correlation id for logging user (str): The user id for audit table logging lock_territories (list(str)): Territories that were received by handler Returns: str, bool: ISRCS, if lock was successful """ try: result = locks.lock_isrc( isrc, isrc_record, active_record, to_remove, to_unlock, to_lock, correlation_id, user, lock_territories) return isrc, result except (ClientError, OwsFeaturesError): if sentry.sentry_client: sentry.sentry_client.captureException() try: self.retry() except MaxRetriesExceededError: error_response = response.create_error_response( error.LOCK_ISRC_ERROR_CODE, error.LOCK_ISRC_ERROR_MESSAGE) return isrc, error_response @celery.task( bind=True, max_retries=3, name='masters_registry.tasks.bulk.save_resolve_conflict_result', base=RequestContextTask) def save_resolve_conflict_result( self, territories, successful, failed, task_id, account_id): """Updates bulk resolve conflicts action status Args: territories (list): 2-letter territory codes successful (list): list of ISRC that were updated successfully failed (list): list of ISRC that weren't updated successfully task_id (int): task id account_id (int): Vendor/Subaccount id """ resolve_result = { field_const.TERRITORIES: territories, field_const.SUCCESSFUL_ISRCS: successful, field_const.FAILED_ISRCS: failed, field_const.ACCOUNT_ID: account_id } if not resolve_result[field_const.FAILED_ISRCS].keys(): status = bulk_tasks_const.DONE_STATUS elif not resolve_result[field_const.SUCCESSFUL_ISRCS].keys(): status = bulk_tasks_const.FAILED_STATUS else: status = bulk_tasks_const.PARTIAL_FAIL_STATUS update_response = bulk_tasks.update_task(task_id, status, resolve_result) if (not update_response and update_response.errors[field_const.CODE] != error.INVALID_BULK_STATUS): try: self.retry() except MaxRetriesExceededError: if sentry.sentry_client: sentry.sentry_client.captureMessage( 'Failed to save BULK task report TaskID: {0},' .format(task_id), extra={'report': resolve_result} ) generate_reports(task_id, [field_const.ISRC]) @retry(error_condition=( lambda err: isinstance(err, ClientError) or isinstance(err, OwsFeaturesError))) def resolve_conflict(isrc, tuids, territories, correlation_id, user): """For given ISRC releases ownership of multiple territories on given TUIDs Args: isrc (str): international standard recording code territories (set): 2-letter territory codes that should be released for given TUIDs user (str): Orchard user id correlation_id (str): The correlation id for logging """ results = {} try: for tuid in tuids: ownership_info = ownership.get_ownership(isrc).message existing_territories = ownership_info[field_const.TERRITORIES] territories_in_conflict = _get_territories_in_conflict( tuid, existing_territories) territories_to_remove = territories.intersection( territories_in_conflict) if territories_to_remove: removed_result = ownership_logic.remove_ownership( isrc, tuid, list(territories_to_remove), existing_territories, correlation_id, user, source=field_const.BULK_RESOLVE_CONFLICTS, daemon=True) updated_claimed_territories = ( removed_result[field_const.UPDATED_CLAIMED_TERRITORIES]) dms_carveout.send_message_to_yt_ownership( isrc, tuid, updated_claimed_territories, correlation_id) results[tuid] = territories_to_remove except (ClientError, OwsFeaturesError): if sentry.sentry_client: sentry.sentry_client.captureException() raise # trigger retry return results def _get_territories_in_conflict(tuid, ownership_territories): """Returns territories that are in conflict with given tuid. Args: tuid (str): track unique identifier ownership_territories (dict): Value of `territories` field of ISRC item Returns: set: 2-letter territory codes """ territories = set() for territory, territory_data in ownership_territories.items(): tuids = ownership_logic._get_tuids_from_territory(territory_data) if len(tuids) > 1 and tuid in tuids: territories.add(territory) return territories @celery.task( name='masters_registry.tasks.bulk.bulk_resolve_internal_conflicts', base=RequestContextTask) def bulk_resolve_conflicts( isrc_tuid_map, isrc_conflicts_map, orchard_user_id, task_id, correlation_id, account_id): """Task for removing ownership of multiple ISRCs on given territories. For account to own territory in ISRC item in DynamoDB means that account has tuid in given territory in ISRC item. Territories in ISRC can be associated with tuids that belong to different accounts. Args: isrc_tuid_map (dict(list)): Map between ISRC(key in DynamoDB) and tuid that belongs to given account. isrc_conflicts_map (dict): {ISRC: territories_with_internal_conflict} that should be released for each of given ISRCs. orchard_user_id (str): Orchard user id task_id (int): ID of created task correlation_id (str): The correlation id for logging account_id (int): Vendor/Subaccount id """ failed = {} successful = {} for isrc, tuids in isrc_tuid_map.items(): isrc_territories = set(isrc_conflicts_map[isrc]) try: result = resolve_conflict( isrc, tuids, isrc_territories, correlation_id, orchard_user_id) successful[isrc] = result except RetryCountExceededError: failed[isrc] = tuids all_conflict_territories = set() for isrc, conflict_territories in isrc_conflicts_map.items(): all_conflict_territories.update(conflict_territories) save_resolve_conflict_result.delay( all_conflict_territories, successful, failed, task_id, account_id) @celery.task( name='masters_registry.tasks.bulk.bulk_remove_territories', base=RequestContextTask) def bulk_remove_territories( items, account_type, account_id, correlation_id, task_id, orchard_user_id): """Celery task for bulk removing a list of territories. This task is used to remove territories for conflicts that were released in the Conflict Manager. After removing the territories from the registry an API call to ows-conflict-manager is made to update the conflict status in Snowflake. Args: items (list): list of conflict items Example: { 'isrc': 'US1234', 'tuid': 123, 'territories': ['US', 'CA'], 'conflict_date': '2017-03-21', 'conflicting_owner': 'BMG' } account_type (str): User account type (vendor or subaccount) account_id (str): User account id correlation_id (str): Correlation id for logs task_id (int): Bulk status task_id orchard_user_id (str): Orchard user_id for checking feature flags """ processing_result = [] for item in items: isrc = item['isrc'] tuid = item['tuid'] territories = item['territories'] user_id = '{}:{}'.format(account_type, account_id) response = mrl.remove_ownership( isrc, tuid, territories, correlation_id, user_id, field_const.AUTO_RELEASED_VIA_CONFLICT_MANAGER, opcode=opcode_const.AUTO_REMOVE) result = { 'isrc': isrc, 'tuid': tuid, 'territories': territories, 'conflict_date': item['conflict_date'], 'conflicting_owner': item['conflicting_owner'], 'success': response.status == 200, 'failed_reason': str(response.errors) if not response else '' } processing_result.append(result) try: _update_conflict_statuses( processing_result, account_type, account_id, correlation_id, orchard_user_id) except RetryCountExceededError: sentry.sentry_client.captureMessage( 'Update conflict status retry count exceeded.', extra={ 'result': processing_result, 'account_type': account_type, 'account_id': account_id, 'correlation_id': correlation_id } ) save_bulk_remove_territories_report.delay(processing_result, task_id) @celery.task( bind=True, retries=3, name='masters_registry.tasks.bulk.save_bulk_remove_territories_report', base=RequestContextTask) def save_bulk_remove_territories_report(self, result, task_id): remove_result = { field_const.ISRCS: result } if all([not item['success'] for item in result]): status = bulk_tasks_const.FAILED_STATUS elif any([not item['success'] for item in result]): status = bulk_tasks_const.PARTIAL_FAIL_STATUS else: status = bulk_tasks_const.DONE_STATUS update_response = bulk_tasks.update_task(task_id, status, remove_result) if (not update_response and update_response.errors[field_const.CODE] != error.INVALID_BULK_STATUS): try: self.retry() except MaxRetriesExceededError: if sentry.sentry_client: sentry.sentry_client.captureMessage( 'Failed to save BULK task report' ' TaskID: {0},'.format(task_id), extra={'report': remove_result} ) class UpdateConflictStatusError(Exception): """Error while updating conflict status via ows-conflict-manager.""" @retry(error_condition=lambda e: True) def _update_conflict_statuses( processed_items, account_type, account_id, correlation_id, orchard_user_id): """Update conflict statuses via ows-conflict-manager. Args: processed_items (list): list of conflict data removed from the registry account_type (str): User account type (vendor or subaccount) account_id (str): User account id correlation_id (str): Correlation id for logs orchard_user_id (str): Orchard user_id for checking feature flags Returns: response.Response: result of the operation. """ grouped_conflict_ids = [] for item in processed_items: if not item['success']: continue grouped_id = '{tuid}|{conflict_date}|{owner}|{action}'.format( tuid=item['tuid'], conflict_date=item['conflict_date'], owner=item['conflicting_owner'], action=field_const.RELEASE_ACTION) grouped_conflict_ids.append(grouped_id) response = ows_conflict_manager.bulk_update_conflict_status( grouped_conflicts_ids=grouped_conflict_ids, status=field_const.AUTO_RELEASED_VIA_CONFLICT_MANAGER, note='', account_type=account_type, account_id=account_id, orchard_user_id=orchard_user_id, correlation_id=correlation_id) if not response: # try to recover from server error if response.status == 500: if sentry.sentry_client: sentry.sentry_client.captureMessage( 'Failed to update conflict status.' ' Retrying.... Error: {}'.format( response.errors), extra=grouped_conflict_ids ) raise UpdateConflictStatusError(str(response.errors)) else: if sentry.sentry_client: sentry.sentry_client.captureMessage( 'bulk_update_conflict_status failed' 'Error: {}'.format(response.errors), extra=grouped_conflict_ids ) return response