"""Utility functions common to both PROPER feeds. Common methods to be used for both Proper feeds. """ import functools import json from owsrequest import request import smart_open from feed_sender.flows.proper_changed_releases import config from feed_sender.flows.proper_new_releases_tracks.conf import settings from feed_sender.util import correlation_id from feed_sender.util import job_status from feed_sender.util import mysql from feed_sender.util.aws import s3 from feed_sender.util.aws import sqs def get_proper_queue_name(db_connection, use_changed_configs=False): """Get queue name for Proper jobs. Args: db_connection (Connection): PyMySQL connection object. use_changed_configs (bool): Determines whether to use the values from changed releases configs Returns: str: Name of SQS queue containing Proper jobs. """ store_id = settings.STORE_ID queue_name = settings.VECTOR_JOB_QUEUE_NAME encoding_priority = 2 if use_changed_configs: store_id = config.STORE_ID queue_name = config.VECTOR_JOB_QUEUE_NAME encoding_priority = 1 sql = """ SELECT priority FROM direct_delivery.customer_master_master cmm WHERE cmm.customer_master_master_id = {store_id} LIMIT 1 """.format(store_id=store_id) rows = mysql.execute_query(db_connection, sql) # query is successful, note: empty values may be returned return queue_name.format( e_priority=str(encoding_priority), d_priority=str(rows[0].get('priority'))) def get_products_from_queue(db_connection, is_meta_update=False): """Get filtered Proper products. Args: db_connection (Connection): PyMySQL connection object. is_meta_update (bool): Flag for whether to get meta update message. Returns: dict: Release ids of the filtered releases. """ job_ids = [] release_ids = [] messages = sqs.get_messages( get_proper_queue_name(db_connection, is_meta_update)) if not messages: return None for message in messages: msg_body = json.loads(message.body) if msg_body['meta_update'] != is_meta_update: # The flow type, is_meta_update, must match job meta_update flag # If the flow type is meta update (changed releases), the job must # also be meta_update == True # Otherwise the job meta_update flag must be False. # # When the 2 don't match, skip over the message. continue message.delete() current_job = (msg_body['encoding_queue_detail_id'], msg_body['release_id']) if current_job in job_ids: # skip duplicate messages continue if current_job[1] in release_ids: for item in job_ids: # search for jobs with matching release IDs and omit & cancel # the one with a lower encoding queue detail id if item[1] != current_job[1]: continue if item[0] < current_job[0]: job_status.update( { 'error_log': 'Duplicate job found', 'eqd_id': item[0], 'status': 'system_cancelled' } ) job_ids.remove(item) job_ids.append(current_job) elif current_job[0] < item[0]: job_status.update( { 'error_log': 'Duplicate job found', 'eqd_id': current_job[0], 'status': 'system_cancelled' } ) else: release_ids.append(current_job[1]) job_ids.append(current_job) # query is successful, note: empty values may be returned return {'release_ids': release_ids, 'job_ids': job_ids} def save_release_ids_to_s3(s3path, release_ids_file, release_ids): """Persist list of Proper release ids and its last updated date in S3. Args: s3path (str): S3 bucket for file defined in settings.py. release_ids_file (str): Where release ids are saved. release_ids (list): list of release ID's """ full_path = '{path}/{file}'.format( path=s3path, file=release_ids_file) with smart_open.smart_open(full_path, 'wb') as fout: for release_id in release_ids: if release_id: fout.write( '{id}\n'.format(id=str(release_id))) else: fout.write(' ') def save_job_ids_to_s3(s3path, job_ids_file, job_ids): """Persist list of Proper job ids and its last updated date in S3. Args: s3path (str): S3 bucket for file defined in settings.py. job_ids_file (str): Where job ids are saved. job_ids (list): list of job ID's """ full_path = '{path}/{file}'.format( path=s3path, file=job_ids_file) with smart_open.smart_open(full_path, 'wb') as fout: for job_tuple in job_ids: fout.write( '{j_id},{r_id}\n'.format( j_id=str(job_tuple[0]), r_id=str(job_tuple[1]))) else: fout.write(' ') def get_ids_from_s3(s3path, job_ids_filename): """Read job IDs file and return file content. Args: s3path (str): S3 bucket for file defined in settings.py. job_ids_filename (str): Filename for job_ids file. Returns: dict: Job IDs and release IDs read from the file. """ full_path = '{path}/{file}'.format(path=s3path, file=job_ids_filename) data_from_s3 = s3.read_s3_object(full_path, 'utf-8').rstrip() parsed_ids = {'release_ids': [], 'job_ids': []} lines = data_from_s3.splitlines() for line in lines: job = line.split(',') parsed_ids['release_ids'].append(int(job[1])) parsed_ids['job_ids'].append(int(job[0])) return parsed_ids def get_ows_pricing(product_id): """Return response from ows-pricing microservice. ARGS: product_id (int): Releases.release_id from art_relations db Returns: price (str): Wholesale/dealer price from ows-pricing. """ get_request = functools.partial( request.process, settings.FEED_NAME, settings.ENV ) # Call microservice response = get_request( 'GET', settings.PRICING_SERVICE, settings.PRICING_SERVICE_URI.format(product_id=product_id), correlation_id.get_correlation_id()) response_content = json.loads(response.content.decode('utf-8')) # Raise exception if microservice cannot be reached if response.status_code != 200: raise Exception( 'Request to ows-pricing microservice failed: {message}'.format( message=response_content.get('message'))) items = response_content.get('items') for item in items: # WW - WorldWide pricing is not set or used if 'WW' in item['territories']: continue return item['price_code'] raise Exception( 'Pricing not found for product ID: {product_id}'.format( product_id=product_id))