"""Process lambda instantiation events.""" from gql.transport.exceptions import TransportServerError from gql.transport.exceptions import TransportQueryError import httpx import copy import json from .common.connectors import s3_sound_recordings from .common.exceptions import exceptions from .common import logger from .utils import graphql from .utils import track_utils from .utils import asset_utils from .connectors import ows_sound_recordings from secrets_manager.lambda_ext import LambdaSecretsManager from soundrecording_utils.metadata.types import OrchardSoundRecording from owsclient import M2MTokenManager from owsclient import OwsClient from src import kafka_producer import config # Configure singletons outside of `handler(...)` # This allows us to re-use the same singletons on different executions # on the same started lambda. secrets_manager = LambdaSecretsManager( environment=config.ENVIRONMENT, service_name=config.APPLICATION_NAME, ) m2m_token_manager = M2MTokenManager( secrets_manager=secrets_manager, environment=config.ENVIRONMENT, service_name=config.APPLICATION_NAME, ) ows_client = OwsClient( environment=config.ENVIRONMENT, service_name=config.APPLICATION_NAME, m2m_token_manager=m2m_token_manager, timeout=httpx.Timeout(20.0), ) def handle_event(event): """ Add SR version to S3 if needed. Args: event: the event that triggered the lambda Returns: string: summary of work object """ ( sound_recording_ids, upcs, tuids, asset_ids, project_ids ) = _process_input(event) sound_recordings = ows_sound_recordings.get_sound_recordings( ows_client=ows_client, sound_recording_ids=sound_recording_ids, tuids=tuids, upcs=upcs, asset_ids=asset_ids, project_ids=project_ids ) # fetch all products to prevent duplication all_upcs = set() for sound_recording in sound_recordings: track_upcs = set([ track['upc'] for asset in sound_recording['assets'] for track in asset['tracks'] ]) all_upcs.update(track_upcs) products_by_id = {} try: products = graphql.get_products(list(all_upcs)) products_by_id = {p['id']: p for p in products.get('products', [])} except TransportServerError as e: if e.code in (504, 503, 502): message_error = f'Unexpected response code from GraphQL Gateway HTTP:{e.code}' # noqa:E501 raise exceptions.RetryableException(message_error) else: raise e results = [] for sound_recording in sound_recordings: # fetch details for connected tracks sound_recording_id = sound_recording['id'] tracks = [] if sound_recording: track_ids = set([ track['tuid'] for asset in sound_recording['assets'] for track in asset['tracks'] ]) if track_ids: try: result = graphql.get_tracks(list(track_ids)) except TransportServerError as e: if e.code in (504, 503, 502): message_error = f'Unexpected response code from GraphQL Gateway HTTP:{e.code}' # noqa:E501 raise exceptions.RetryableException(message_error) else: raise e except TransportQueryError as e: raise exceptions.RetryableException(str(e)) tracks = list(result['tracks']) if None in tracks: logger.warning( f'Skipping sound_recording_id={sound_recording_id}: ' 'GraphQL returned null track data' ) continue for track in tracks: # match products to tracks product_id = track['product']['id'] matching_product = products_by_id.get(product_id) if matching_product: track['product'] = copy.deepcopy(matching_product) else: track['product'] = None tracks = track_utils.clean_tracks_with_null_product_release_dates(tracks) # format explicit into valid enum value track_utils.clean_explicit_field(tracks) # format label data track_utils.format_label_id_field(tracks) # extract ISRCs from tracks tracks_isrcs = track_utils.extract_tracks_isrcs(tracks) # set primary boolean on single track track_utils.select_primary_track(tracks, sound_recording['primary_track_id']) # format into OrchardSoundRecording model in common types raw_sr_version = { 'track_connection': { 'tracks': tracks }, 'assets': asset_utils.select_primary_assets(sound_recording['assets']), 'isrc': sound_recording['isrc'] } # load into class, verifying structure OrchardSoundRecording(**raw_sr_version) # convert to JSON to be saved sr_version_json = json.dumps(raw_sr_version) # save in s3, returning verison id and last (prev) modified date # Note: new_version_id may be None if no changes were made # but in that case the step function will terminate anyway (previous_version_id, new_version_id, updated, last_modified) = s3_sound_recordings.write_sound_recording_version( sound_recording_id, sr_version_json) if (new_version_id is not None): data = { 'sound_recording_id': sound_recording_id, 'previous_version_id': previous_version_id, 'version_id': new_version_id, 'updated': updated, 'last_modified': last_modified.isoformat() if last_modified is not None else None, # noqa:E501 'tracks_isrcs': tracks_isrcs } results.append(data) kafka_producer.produce_message( json.dumps(data), config.KAFKA_TOPIC, sound_recording_id) return { 'results': results, 'updates': len([x for x in results if x.get('updated')]) > 0 } def _process_input(event): upcs = None asset_ids = None tuids = None project_ids = None # If sound_recordings_ids in event: Previous step sr-create sound_recording_ids = event.get('sound_recording_ids') # If label in event: Previous step InputChoice data_type = event.get('label') if not sound_recording_ids and data_type: data_id = event.get('id') if data_type == 'Track': tuids = [data_id] elif data_type == 'OrchardAsset': asset_ids = [data_id] elif data_type == 'OrchardSoundRecording': sound_recording_ids = [data_id] elif data_type == 'Product': upcs = [data_id] elif data_type == 'Project': project_ids = [data_id] # unable to find IDs to process if not sound_recording_ids and not upcs and not tuids and not asset_ids and not project_ids: raise Exception('Unexpected parameters') return ( sound_recording_ids, upcs, tuids, asset_ids, project_ids )