"""Track Persister. Handles doing CRUD operations on the track and related tables. """ import time from collections import defaultdict from flask import g from oto import response from sentry_sdk import capture_exception from oto import status as response_code import sqlalchemy from backend.connectors import mysql from backend.connectors.sentry import send_to_sentry from backend.constants import error from backend.constants import track_field from backend.constants import track_role as track_role_const from backend.constants import validation from backend.constants.field import NAMES_FIELD_MODIFIER from backend.exceptions import IntegrityError from backend.models.sql import track as tracksql from backend.models.track import Track from backend.models.track import TrackArtist from backend.models.track import TrackWriter from backend.models.track_query import TrackQuery from backend.utils import api as api_utils from backend.utils.validation import is_list_unique_and_contiguous class TrackPersister(object): """Handles high level operations for tracks.""" SUCCESS = 'Success' @classmethod @mysql.db_session_wrap def get_by_tuid(cls, tuid, session): """Get track by its unique id. Args: tuid (int): Primary key of track Returns: response.Response: dictionary of track data """ track = TrackQuery.get_by_tuid(tuid, session) return cls._make_track_response(track) @classmethod @mysql.db_session_wrap def get_by_upc_and_isrc(cls, upc, isrc, session): """Get track by upc and isrc. Args: upc (int): UPC of the track's product isrc (str): ISRC of the track session (object): SQLAlchemy database session (injected) Returns: response.Response: full track dict if found, not found response otherwise """ track = TrackQuery.get_by_upc_and_isrc(upc, isrc, session) return cls._make_track_response(track) @classmethod @mysql.db_session_wrap def get_by_isrc_and_type(cls, isrc, track_type, session): """Get digital product isrcs by track_type.""" track = session.execute( tracksql.GET_DIGITAL_ISRC_BY_TRACK_TYPE, {'track_type': track_type, 'isrc': isrc}).first() if track: return track return {} @classmethod @mysql.db_session_wrap def get_all_by_isrc_and_type(cls, isrc, track_type, session): """Get all results by isrc and track_type.""" tracks = TrackQuery.get_all_by_isrc_and_type(isrc, track_type, session) if tracks: return [track.to_dict() for track in tracks] return [] @classmethod @mysql.db_session_wrap def get_all_for_account_by_isrc_and_type(cls, account_type, account_id, isrc, track_type, session): """Get all results for a specific account by isrc and track_type.""" if account_type == 'vendor': query = session.execute( tracksql.GET_DIGITAL_ISRC_BY_TRACK_TYPE_AND_VENDOR_ID, { 'track_type': track_type, 'isrc': isrc, 'vendor_id': account_id, }, ) elif account_type == 'subaccount': query = session.execute( tracksql.GET_DIGITAL_ISRC_BY_TRACK_TYPE_AND_SUBACCOUNT_ID, { 'track_type': track_type, 'isrc': isrc, 'subaccount_id': account_id, }, ) return cls.get_multiple_by_tuids( [track.tuid for track in query.fetchall()], session=session, ) @classmethod @mysql.db_session_wrap def get_multiple_by_tuids(cls, tuids, session): """Get tracks by list of tuids. Args: tuids (list): Primary key of track Returns: response.Response: dictionary of track data """ try: tracks = TrackQuery.get_by_tuids(tuids, session) except ValueError: session.rollback() return response.create_not_found_response() return api_utils.create_get_list_response( [track.to_dict() for track in tracks]) @classmethod @mysql.db_session_wrap def get_multiple_by_tuids_with_nones(cls, tuids, session): """Get tracks by list of tuids. Args: tuids (list): Primary key of track Returns: response.Response: list of dictionary of track data """ tracks = TrackQuery.get_by_tuids_with_nones(tuids, session) return api_utils.create_get_list_response( [track.to_dict() if track else None for track in tracks]) @classmethod @mysql.db_session_wrap def get_all_by_product_id(cls, product_id, session, is_overview=False): """Get list of all tracks by product_id. Args: product_id (int): id of product session (object): SQLAlchemy database session (optional) Returns: response.Response: List of track dictionaries """ tracks = TrackQuery.get_all_by_product_id( product_id, session, is_overview=is_overview) # Convert results to array of dicts if is_overview: items = [track.overview_to_dict() for track in tracks] else: items = [track.to_dict() for track in tracks] return api_utils.create_get_list_response(items) @classmethod @mysql.db_session_wrap def get_tuids_by_product_ids_with_order( cls, product_ids, order_by_fields, session, ): """Get list of all tracks by product_id. Args: product_ids (list): id of product order_by_fields (iterable): iterable of fields names for ordering session (object): SQLAlchemy database session (optional) Returns: Query Object """ selected_tracks = TrackQuery.get_tuids_by_product_ids_with_order( product_ids=product_ids, order_by_fields=order_by_fields, session=session, ) tracks = [ track.columns_values_to_dict( fields=[track_field.TUID, track_field.PRODUCT_ID], ) for track in selected_tracks ] return tracks @classmethod @mysql.db_session_wrap def get_all_focus_track_by_product_id(cls, product_id, session, is_overview=False): """Get list of all focus tracks by product_id. Args: product_id (int): id of product session (object): SQLAlchemy database session (optional) Returns: response.Response: List of track dictionaries """ tracks = TrackQuery.get_all_focus_track_by_product_id( product_id, session, is_overview=is_overview) # Convert results to array of dicts if is_overview: items = [track.overview_to_dict() for track in tracks] else: items = [track.to_dict() for track in tracks] return api_utils.create_get_list_response(items) @classmethod @mysql.db_session_wrap def get_all_by_product_id_light(cls, product_id, session): """Get list of all tracks by product_id. Args: product_id (int): id of product session (object): SQLAlchemy database session (optional) Returns: response.Response: List of track dictionaries """ tracks = TrackQuery.get_all_by_product_id_light(product_id, session) # Convert results to array of dicts items = [track.to_light_dict() for track in tracks] return api_utils.create_get_list_response(items) @classmethod @mysql.db_session_wrap def get_all_by_product_ids_medium(cls, product_ids, session): """Get list of all tracks by product_ids. Args: product_ids (list): ids of product session (object): SQLAlchemy database session (optional) Returns: response.Response: List of track dictionaries """ tracks = TrackQuery.get_all_by_product_ids_medium(product_ids, session) # Convert results to array of dicts items = [track.to_medium_dict() for track in tracks] return api_utils.create_get_list_response(items) @classmethod @mysql.db_session_wrap def get_all_track_isrc_by_product_id(cls, product_id, session): """Get list of all track ISRCs by product_id. Args: product_id (int): ID of product session (object): SQLAlchemy database session (optional) Returns: response.Response: List of track ISRCs """ tracks = TrackQuery.get_all_track_isrc_by_product_id(product_id, session) # Convert results to array of dicts items = [track.to_dict() for track in tracks] return api_utils.create_get_list_response(items) @classmethod @mysql.db_session_wrap def get_spatial_isrc_map_by_product_id(cls, product_id, session): """Get a mapping of track_id to spatial ISRC for all tracks in a product. Args: product_id (int): ID of product session (object): SQLAlchemy database session (optional) Returns: response.Response: dict of {track_id: isrc} """ return response.Response( message=TrackQuery.get_spatial_isrc_map_by_product_id(product_id, session)) @classmethod def create_track(cls, product_id, upc, track_data, for_volume=None): """Create new track. Creates just a single track. Piggy-backs on bulk_create_tracks by just sending a list of one track to create and changing the response message to just a track dictionary. Args: product_id (int): Foreign Key to Product upc (int): UPC of Product track_data_list: List of tracks to add for_volume (int): Volume Number to add tracks to """ result = cls.bulk_create_tracks( product_id, upc, [track_data], for_volume=for_volume) if result: track = result.message['items'][0] return response.Response( message=track, status=response_code.CREATED) return result @classmethod @mysql.db_session_retry_wrap def bulk_create_tracks( cls, product_id, upc, track_data_list, session, for_volume=None, commit=True, retry_count=1): """Bulk create new tracks. The volume number to add the track to can be specified, but must be for an already existing volume, or for a new volume that is after the last volume of the product. This function does row locking so race conditions doesn't result in invalid volume and track numbers, but a deadlock is possible if there are no tracks for the product and multiple requests to create a new track are made at the same time. The db_session_retry_wrap will retry to create the track again if this happens. Usually this will be successful since the new track is created by this point. Args: session (object): SQLAlchemy database session product_id (int): Foreign Key to Product upc (int): UPC of Product track_data_list: List of tracks to add for_volume (int): Volume Number to add tracks to commit (bool): If commit required Returns: response.Response: Contains status code and payload """ for track_data in track_data_list: for key in track_data.keys(): if key not in track_field.CREATE_MODEL_FIELDS: msg = error.VALIDATION_ERROR_SUPERFLUOUS_FIELD_MSG.format( key) return api_utils.create_validation_error_response(msg) tracks_query = TrackQuery.get_all_by_product_id( product_id, session, eager_loading=False).with_for_update() # We need to have the tracks be a collection. This will allow for the # lock to occur (will force query to execute) and for quickly finding # the next track vol and num to update without doing more database # queries, reducing the risk of a deadlock occuring. existing_tracks = [] for track in tracks_query: existing_tracks.append(track) try: # Find intial volume and track number to start with (volume_number, track_number) = cls._find_next_track_position( existing_tracks, for_volume=for_volume) # Add all the tracks, increasing the track_number for each track tracks = [] for track_data in track_data_list: create_data = track_data.copy() create_data.update({ track_field.UPC: upc, track_field.PRODUCT_ID: product_id, track_field.VOLUME_NUMBER: volume_number, track_field.TRACK_NUMBER: track_number }) track = Track(**create_data) session.add(track) tracks.append(track) track_number += 1 if commit: session.commit() else: session.flush() # Convert results to array of dicts items = [track.to_dict() for track in tracks] return api_utils.create_get_list_response( items, status=response_code.CREATED) except ValueError as e: session.rollback() return response.create_error_response( code=error.VALIDATION_ERROR_CODE, message=str(e)) except sqlalchemy.exc.IntegrityError as e: session.rollback() return api_utils.create_validation_error_response(str(e.orig)) except sqlalchemy.exc.OperationalError: if retry_count >= 3: raise # Retry on deadlock session.rollback() time.sleep(retry_count) return cls.bulk_create_tracks( product_id, upc, track_data_list, session=session, for_volume=for_volume, commit=commit, retry_count=retry_count+1) @classmethod @mysql.db_session_wrap def update_track( cls, tuid, data, session, generate_isrc=False, get_artist_info_ids_from_existing_rows=False): """Update track. Args: tuid (int): Primary key of track data (dict): Track data to update session (object): SQLAlchemy database session (optional) generate_isrc (bool): Generates ISRC field when True get_artist_info_ids_from_existing_rows (bool): Save artist_info_ids provided without track_artist_id Returns: response.Response: Contains status code and payload """ try: tracks = TrackQuery.get_by_tuids( [tuid], session, with_for_update=True) except ValueError: session.rollback() return response.create_not_found_response() track = tracks[0] update_data = data.copy() if generate_isrc: # generate isrc and push isrc into data isrc_result = TrackQuery.claim_new_isrcs(session) update_data['isrc'] = isrc_result[0] result = cls._update_track_helper( track, update_data, session, get_artist_info_ids_from_existing_rows=get_artist_info_ids_from_existing_rows, ) if result != cls.SUCCESS: return result return cls._make_track_response(track) @classmethod @mysql.db_session_wrap def update_multiple_tracks( cls, track_update_list, session, belongs_to_product_id=None, update_track_volume_numbers=False, get_artist_info_ids_from_existing_rows=False): """Update multiple tracks. Args: track_update_list (list): Dict of track updates session (object): SQLAlchemy database session (optional) belongs_to_product_id (int): Product Id track must belong to update_track_volume_numbers (bool): Allow track volume and number fields update. get_artist_info_ids_from_existing_rows (bool): Save artist_info_ids provided without track_artist_id Returns: response.Response: Contains status code and payload """ try: tuids = [item[track_field.TUID] for item in track_update_list] tracks = TrackQuery.get_by_tuids( tuids, session, belongs_to_product_id=belongs_to_product_id) except ValueError as e: session.rollback() return response.create_error_response( code=error.VALIDATION_ERROR_CODE, message=str(e)) track_update_lookup = { update[track_field.TUID]: update for update in track_update_list} for track in tracks: update_data = track_update_lookup[track.tuid] result = cls._update_track_helper( track, update_data, session, rollback=True, update_track_volume_numbers=update_track_volume_numbers, get_artist_info_ids_from_existing_rows=get_artist_info_ids_from_existing_rows, # noqa ) if result != cls.SUCCESS: return result return api_utils.create_get_list_response( [track.to_dict() for track in tracks]) @classmethod @mysql.db_session_wrap def update_track_duration( cls, tuid, data, session): """Update track duration fields. Args: tuid (int): Primary key of track data (dict): Track data to update session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ try: tracks = TrackQuery.get_by_tuids( [tuid], session, with_for_update=True) except ValueError: session.rollback() return response.create_not_found_response() track = tracks[0] track_update_data = {} for field in data: if field in (track_field.DURATION_MINUTES, track_field.DURATION_SECONDS): track_update_data[field] = data[field] try: track.update(**track_update_data) return response.Response( message=track_update_data, status=response_code.OK ) except (TypeError, ValueError) as e: session.rollback() return api_utils.create_validation_error_response(str(e)) @classmethod def _update_track_helper( cls, track, update_data, session, rollback=False, update_track_volume_numbers=False, get_artist_info_ids_from_existing_rows=False, retry_count=1): """Update a track with update data. Args: track (object): Track object update_data (dict): Data used to update the track session (object): SQLAlchemy database session (optional) rollback (bool): Rollback session on failure of format_artists_and_writers_response # noqa update_track_volume_numbers (bool): Allow track volume and number fields update. get_artist_info_ids_from_existing_rows (bool): Save artist_info_ids provided without track_artist_id Returns: object: Contains error or success string """ artist_info_ids_to_set = {} if get_artist_info_ids_from_existing_rows: artist_info_ids_to_set = ( cls._extract_track_artist_and_writer_ids(update_data).message) format_artists_and_writers_response = ( cls._format_artists_and_writers_for_update(update_data)) if not format_artists_and_writers_response: if rollback: session.rollback() return format_artists_and_writers_response result = cls._apply_track_update( track, update_data, update_track_volume_numbers=update_track_volume_numbers, artist_info_ids_to_set=artist_info_ids_to_set) if not result: session.rollback() return result # FIXME: It's unknown at this point if track must have ownership # rights in order to populate master rights fields. If ownership # rights ends up being a requirement, then session should be rolled # back at this point. Otherwise, remove this comment. try: session.merge(track) session.flush() except sqlalchemy.exc.IntegrityError as ex: # An integrity error can happen when there is still another update # request pending for the same track. session.rollback() capture_exception(ex) return response.create_error_response( code=error.TOO_MANY_WRITES_CODE, message=error.TOO_MANY_WRITES_MSG) except sqlalchemy.exc.OperationalError: if retry_count >= 3: raise # Retry on deadlock session.rollback() time.sleep(retry_count) return cls._update_track_helper( track, update_data, session, rollback=rollback, update_track_volume_numbers=update_track_volume_numbers, get_artist_info_ids_from_existing_rows=get_artist_info_ids_from_existing_rows, retry_count=retry_count+1) return cls.SUCCESS @classmethod @mysql.db_session_wrap def create_and_update_multiple_tracks( cls, product_id, upc, track_data_list, session): """Create multiple tracks and populate them with data. Args: product_id (int): Product id. upc (str): Product upc. track_data_list (list): List of tracks data. session (sqlalchemy.orm.session.Session): Database session. Returns: response.Response: Contains status code and payload """ create_track_data = [ {'track_name': item['track_name']} for item in track_data_list ] bulk_create_tracks_response = cls.bulk_create_tracks( product_id, upc, create_track_data, session=session, commit=False ) if not bulk_create_tracks_response: return bulk_create_tracks_response created_tracks = bulk_create_tracks_response.message['items'] for i in range(len(created_tracks)): tuid = created_tracks[i]['tuid'] track_data_list[i]['tuid'] = tuid return cls.update_multiple_tracks( track_data_list, session=session, update_track_volume_numbers=True, get_artist_info_ids_from_existing_rows=True) @classmethod @mysql.db_session_wrap def delete_by_tuid(cls, tuid, product_id, session): """Delete track. Args: tuid (int): Primary key of track product_id (int): Foreign Key to Product session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ return cls.delete_many_by_tuids([tuid], product_id, session=session) @classmethod @mysql.db_session_wrap def delete_many_by_tuids(cls, tuids, product_id, session): """Delete several tracks. All tracks must exist for this operation to succeed, otherwise session will rollback. Args: tuids (list): List of track primary keys product_id (int): Foreign Key to Product session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ delete_tuid_set = set(tuids) if len(tuids) != len(delete_tuid_set): return response.create_error_response( code=error.DUPLICATE_CODE, message='Invalid tuid list') original_tracks = TrackQuery.get_all_by_product_id( product_id, session, eager_loading=False).with_for_update().all() # A list of remaining tracks that retain the original sort order will # be needed later to fix any gaps in the track order. remaining_tracks = [] num_deleted_tracks = 0 for track in original_tracks: if track.tuid in delete_tuid_set: session.delete(track) num_deleted_tracks += 1 else: remaining_tracks.append(track) # Verify all tracks are deleted if num_deleted_tracks != len(delete_tuid_set): session.rollback() return response.create_not_found_response( error.ERROR_MESSAGE_TRACK_NOT_FOUND) arrangement = cls._fix_track_order(remaining_tracks) session.commit() return api_utils.create_get_list_response(arrangement) @classmethod def _fix_track_order(cls, tracks): """Fix track order so it's contiguous. Args: tracks: List of track models in ascending order by volume number and track number Return: (list): New arrangement of tracks """ arrangement = [] if not tracks: return arrangement volume_number = 1 track_number = 1 last_volume = tracks[0].volume_number for track in tracks: if track.volume_number != last_volume: # Start a new volume last_volume = track.volume_number volume_number += 1 track_number = 1 # Make track number contigious for volume track.volume_number = volume_number track.track_number = track_number arrangement.append({ track_field.TUID: track.tuid, track_field.VOLUME_NUMBER: volume_number, track_field.TRACK_NUMBER: track_number }) track_number += 1 return arrangement @classmethod @mysql.db_session_wrap def update_field_for_all_tracks_in_product( cls, product_id, field_name, field_value, session, response_include_fields=None): """Update all tracks for product for specified field. This can only handle fields with basic primitive types. This can also update fields in other tables with a one-to-one relationship. This function does row-locking to avoid race conditions. Args: product_id (int): Foreign Key to Product field_name (string): Name of field to update field_value (bool, int, string, None): Value to set field to session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ response_include_fields = response_include_fields or [] if field_name not in track_field.APPLY_TO_ALL_FIELDS \ or field_name not in track_field.BASIC_MODEL_FIELDS: return api_utils.create_validation_error_response( error.VALIDATION_ERROR_INVALID_FIELD_MSG.format(field_name)) eager_loading = field_name in track_field.RELATIONSHIP_FIELDS tracks = TrackQuery.get_all_by_product_id( product_id, session, eager_loading).with_for_update() update_deltas = [] for track in tracks: setattr(track, field_name, field_value) track_dict = track.to_dict( filter_by_fields=[field_name] + response_include_fields) update_deltas.append(track_dict) session.bulk_save_objects(tracks) return api_utils.create_get_list_response(update_deltas) @classmethod @mysql.db_session_wrap def sync_role_for_all_tracks_in_product( cls, product_id, role_type, names, session): """Sync all tracks for product to have listed role. Returns a list of deltas of track fields changed and includes the track's metalanguage code. Args: product_id (int): Foreign Key to Product role_type (str): Type of role to update names (list): List of names for new roles session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ try: role_track_field = cls._role_type_to_field(role_type) except ValueError as e: return response.create_error_response( code=error.INVALID_ROLE_ERROR_CODE, message=str(e)) tracks = TrackQuery.get_all_by_product_id( product_id, session).with_for_update() deltas = [] for track in tracks: track.sync_role(role_type, names) # Flush so ids are generated for any new roles session.flush() deltas.append( track.to_dict(filter_by_fields=[ track_field.META_LANGUAGE_CODE, role_track_field])) return api_utils.create_get_list_response(deltas) @classmethod @mysql.db_session_wrap def add_track_artist(cls, tuid, artist_type, artist_name, session): """Add an artist to a track. Args: tuid (int): Primary key of track artist_type (string): Artist type (performer, featured, etc...) artist_name (string): Artist Name session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ track = TrackQuery.get_by_tuid(tuid, session) if not track: return response.create_not_found_response() try: artist = track.add_artist(artist_type, artist_name) except IntegrityError as e: session.rollback() return response.create_error_response( code=error.DUPLICATE_CODE, message=str(e)) except ValueError as e: session.rollback() return response.create_error_response( code=error.INVALID_VALUE_ERROR_CODE, message=str(e)) # Save artist and generate primary key session.flush() return response.Response( message=artist.to_dict(), status=response_code.CREATED) @classmethod @mysql.db_session_wrap def update_track_artist( cls, tuid, artist_type, track_artist_id, data, session): """Update a track artist. Args: tuid (int): Primary key of track artist_type (string): Artist type (performer, featured, etc...) track_artist_id (int): Primary key of track artist data (dict): Dictionary of attributes to update. session (object): SQLAlchemy database session Returns: response.Response: Contains status code and payload """ track_artist = session\ .query(TrackArtist)\ .filter_by( artist_type=artist_type, track_artist_id=track_artist_id, tuid=tuid, ).first() if not track_artist: return response.create_not_found_response() for key, value in data.items(): setattr(track_artist, key, value) return response.Response(message=track_artist.to_dict()) @classmethod @mysql.db_session_wrap def remove_track_artist(cls, tuid, artist_type, track_artist_id, session): """Remove artist from track. Args: tuid (int): Primary key of track artist_type (string): Artist type (performer, featured, etc...) track_artist_id (int): Primary key of track_artist session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ track = TrackQuery.get_by_tuid(tuid, session) try: if track and track.remove_artist( track_artist_id, artist_type=artist_type): return api_utils.create_ok_response() return response.create_not_found_response() except ValueError as e: session.rollback() return response.create_error_response( code=error.INVALID_VALUE_ERROR_CODE, message=str(e)) @classmethod @mysql.db_session_wrap def add_track_publisher(cls, tuid, publisher_name, session): """Add a publisher to a track. Args: tuid (int): Primary key of track publisher_name (string): Publisher Name session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ track = TrackQuery.get_by_tuid(tuid, session) if not track: return response.create_not_found_response() try: publisher = track.add_publisher(publisher_name) except IntegrityError as e: session.rollback() return response.create_error_response( code=error.DUPLICATE_CODE, message=str(e)) # Save publisher and generate primary key session.flush() return response.Response( message=publisher.to_dict(), status=response_code.CREATED) @classmethod @mysql.db_session_wrap def remove_track_publisher(cls, tuid, track_publisher_id, session): """Remove publisher from track. Args: tuid (int): Primary key of track track_publisher_id (int): Primary key of track_publisher session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ track = TrackQuery.get_by_tuid(tuid, session) if track and track.remove_publisher(track_publisher_id): return api_utils.create_ok_response() return response.create_not_found_response() @classmethod @mysql.db_session_wrap def add_track_writer(cls, tuid, writer_name, session): """Add a writer to a track. Args: tuid (int): Primary key of track writer_name (string): Writer Name session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ track = TrackQuery.get_by_tuid(tuid, session) if not track: return response.create_not_found_response() try: writer = track.add_writer(writer_name) except IntegrityError as e: session.rollback() return response.create_error_response( code=error.DUPLICATE_CODE, message=str(e)) # Save writer and generate primary key session.flush() return response.Response( message=writer.to_dict(), status=response_code.CREATED) @classmethod @mysql.db_session_wrap def update_track_writer(cls, tuid, track_writer_id, data, session): """Update a track writer. Args: tuid (int): Primary key of track track_writer_id (int): Primary key of track writer data (dict): Dictionary of attributes to update. session (object): SQLAlchemy database session Returns: response.Response: Contains status code and payload """ track_writer = session.query(TrackWriter) \ .filter_by( track_writer_id=track_writer_id, tuid=tuid, ).first() if not track_writer: return response.create_not_found_response() for key, value in data.items(): setattr(track_writer, key, value) return response.Response(message=track_writer.to_dict()) @classmethod @mysql.db_session_wrap def remove_track_writer(cls, tuid, track_writer_id, session): """Remove writer from track. Args: tuid (int): Primary key of track track_writer_id (int): Primary key of track_writer session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ track = TrackQuery.get_by_tuid(tuid, session) if track and track.remove_writer(track_writer_id): return api_utils.create_ok_response() return response.create_not_found_response() @classmethod @mysql.db_session_wrap def reorder(cls, product_id, arrangement, session, tracks_query=None): """Reorder tracks to specified arrangement. Note that a race condition is possible if tracks are being created, deleted, or reordered while this operation is being completed. This could potentially result in invalid track volume and number data. Example arrangement list: [{'tuid': 5, 'volume_number': 1, 'track_number': 1}, {'tuid': 6, 'volume_number': 1, 'track_number': 2}, {'tuid': 7, 'volume_number': 2, 'track_number': 1}] Args: product_id (int): product_id of tracks to arrange. arrangement (list): List of track arrangements (example above) session (object): SQLAlchemy database session (optional) tracks_query (TrackQuery): tracks that need to be reordered Returns: response.Response: Contains status code and payload """ # Lookup of (volume, track #) by TUID. Example tuid_arrgt_lookup: # {5: (1, 1), 6: (1, 2), 7: (2, 1)} tuid_arrgt_lookup = {} for track in arrangement: tuid = track[track_field.TUID] if tuid in tuid_arrgt_lookup: # Same tuid is being updated return api_utils.create_validation_error_response( error.VALIDATION_ERROR_DUPLICATE_DATA_MSG) tuid_arrgt_lookup[tuid] = ( track[track_field.VOLUME_NUMBER], track[track_field.TRACK_NUMBER] ) # SQLAlchemy will throw a stale data exception if rows aren't # locked using 'with_for_update'. tracks = tracks_query or TrackQuery.get_all_by_product_id( product_id, session, eager_loading=False).with_for_update() # Go through each track and update, but do not commit. updated_tracks = [] for track in tracks: if track.tuid in tuid_arrgt_lookup: track_arrange = tuid_arrgt_lookup[track.tuid] track.volume_number, track.track_number = track_arrange updated_tracks.append(track) # Make sure all track arrangements were found if len(tuid_arrgt_lookup) != len(updated_tracks): session.rollback() return api_utils.create_validation_error_response( error.VALIDATION_ERROR_INVALID_TRACK_FOUND_MSG) # Verify track volume and numbers are sorted correctly is_valid, error_code = cls._validate_tracks_order(tracks) if not is_valid: session.rollback() msg = error.ERROR_TRACK_ORDER_CODE_TO_MSG[error_code] return response.create_error_response( code=error_code, message=msg) # Save updated tracks session.bulk_save_objects(updated_tracks) return response.Response(message='success') @classmethod @mysql.db_session_wrap def assign_track_isrcs_for_product_id(cls, product_id, session): """Assign ISRCs for all tracks in a product. Args: product_id (int): product_id of tracks to arrange. session (object): SQLAlchemy database session (optional) Returns: response.Response: Contains status code and payload """ tracks = TrackQuery.get_all_by_product_id( product_id, session, eager_loading=False).with_for_update() tracks_count = tracks.count() if not tracks_count: return response.create_not_found_response() isrcs = TrackQuery.claim_new_isrcs(session, tracks_count) isrcs = iter(isrcs) for track in tracks: # TODO: catch possible errors during ISRC generation track.isrc = next(isrcs) # Convert results to array of dicts items = [ { track_field.TUID: track.tuid, track_field.ISRC: track.isrc } for track in tracks ] return api_utils.create_get_list_response( items, status=response_code.OK) @classmethod @mysql.db_session_wrap def import_tracks(cls, src_tuids, dest_product_id, dest_upc, session): """Import existing tracks. Args: src_tuids (list): list of tuids of the source tracks. dest_product_id (int): product_id for the destination track. dest_upc (int): UPC for the destination tracks. session (object): SQLAlchemy database session (optional) Returns: response.Response: Response object with a list of tuples with source tuid and new track response. Response can be successfull or not successfull. Example list: [(123, Response), (234, Response)] """ results = [] tracks = TrackQuery.get_all_by_product_id( dest_product_id, session, with_for_update=True, eager_loading=False).all() volume_number, track_number = cls._find_next_track_position(tracks) try: src_tracks = TrackQuery.get_by_tuids(src_tuids, session) except ValueError as e: session.rollback() return response.create_not_found_response(message=str(e)) # Import tracks try: for src_track in src_tracks: results.append(cls._import_track( src_track, dest_product_id, dest_upc, volume_number, track_number, session)) track_number += 1 except Exception as ex: # Abort on failure (likely due to source track data issue) capture_exception(ex) session.rollback() return response.create_fatal_response('Failed to clone track') # Save before creating response session.commit() return api_utils.create_get_list_response([ {'source': source.to_dict(), 'destination': dest.to_dict()} for source, dest in results ]) @classmethod @mysql.db_session_wrap def copy_product_tracks( cls, src_product_id, dest_product_id, dest_upc, session, exclude_fields=None, track_list=None): """Import all tracks from src product to empty dest product. Args: src_product_id (int): product_id of the source product. dest_product_id (int): product_id of the destination product. dest_upc (int): UPC of destination product. session (object): SQLAlchemy database session (optional). exclude_fields (list): list of fields which shouldn't be copied. track_list (list): list of track ids which should be copied. Returns: response.Response: Contains status code and payload """ exclude_fields = exclude_fields or [] results = [] dest_product_tracks = TrackQuery.get_all_by_product_id( dest_product_id, session, with_for_update=True, eager_loading=False).all() if dest_product_tracks: return response.create_error_response( code=error.COPY_TRACKS_ERROR_CODE, message=error.COPY_TRACKS_ERROR_MESSAGE) src_product_tracks = TrackQuery.get_all_by_product_id( src_product_id, session) if track_list is not None: src_product_tracks = list(TrackQuery.get_all_by_product_id( src_product_id, session).filter(Track.tuid.in_(track_list))) for index, src_track in enumerate(src_product_tracks): src_track.to_dict() # lazy load data before expunging session session.expunge(src_track) src_track.track_number = index + 1 src_track.volume_number = 1 # Import tracks try: for src_track in src_product_tracks: results.append(cls._import_track( src_track, dest_product_id, dest_upc, src_track.volume_number, src_track.track_number, session, exclude_fields=exclude_fields)) except Exception as ex: # Abort on failure (likely due to source track data issue) capture_exception(ex) session.rollback() return response.create_fatal_response('Failed to clone track') # Save before creating response session.commit() return api_utils.create_get_list_response([ {'source': source.to_dict(), 'destination': dest.to_dict()} for source, dest in results ]) @classmethod @mysql.db_session_wrap def claim_new_isrc(cls, session): """Claim new ISRC. Args: session (object): SQLAlchemy database session (optional). Returns: response.Response: Contains status code and payload with ISRC (str). """ isrc_result = TrackQuery.claim_new_isrcs(session, number_of_isrcs=1) if isrc_result: return response.Response( message={'isrc': isrc_result[0]}, status=response_code.OK) error_response = response.create_error_response( code=error.INTERNAL_ERROR, message='Unable to claim an ISRC.', status=response_code.INTERNAL_ERROR) send_to_sentry(error_response, {}, 500, error.ERROR_MESSAGE_UNABLE_TO_CLAIM_ISRC) return error_response @classmethod @mysql.db_session_wrap def claim_new_isrcs(cls, number_of_isrcs, session): """Claim multiple new ISRCs in a single call. Args: number_of_isrcs (int): Number of ISRCs to claim. session (object): SQLAlchemy database session (optional). Returns: response.Response: Contains status code and payload with list of ISRCs. """ if number_of_isrcs <= 0: return response.Response( message={'isrcs': []}, status=response_code.OK) isrc_result = TrackQuery.claim_new_isrcs(session, number_of_isrcs=number_of_isrcs) if isrc_result and len(isrc_result) == number_of_isrcs: return response.Response( message={'isrcs': isrc_result}, status=response_code.OK) error_response = response.create_error_response( code=error.INTERNAL_ERROR, message=f'Unable to claim {number_of_isrcs} ISRCs.', status=response_code.INTERNAL_ERROR) send_to_sentry(error_response, {}, 500, error.ERROR_MESSAGE_UNABLE_TO_CLAIM_ISRC) return error_response @classmethod def _import_track( cls, src_track, dest_product_id, dest_upc, dest_volume_number, dest_track_number, session, exclude_fields=None): """Import an existing track and all related metadata. Args: src_track (Track): Track model object for source track. dest_product_id (int): product_id for the destination track. dest_upc (int): UPC for the destination track. dest_volume_number (int): Volume number for destination track. dest_track_number (int): Track number for cloned track. session (object): SQLAlchemy database session (optional). exclude_fields (list): list of fields which shouldn't be copied. Returns: response.Response: Response object with cloned track. """ exclude_fields = exclude_fields or [] track_data = src_track.to_dict( filter_by_fields=track_field.BASIC_MODEL_FIELDS) track_data.pop(track_field.TUID) # exclude_fields currently works only with basic track data if exclude_fields: for field_name in exclude_fields: track_data.pop(field_name, None) new_track = Track(**track_data) new_track.upc = dest_upc new_track.product_id = dest_product_id new_track.track_number = dest_track_number new_track.volume_number = dest_volume_number new_track.duration_minutes = src_track.duration_minutes new_track.duration_seconds = src_track.duration_seconds session.add(new_track) # One to many relational metadata for artist in src_track.artists: if artist.role_type in track_role_const.SUPPORTED_ARTIST_ROLES: new_track.add_artist( artist.role_type, artist.name) for writer in src_track.writers: new_track.add_writer(writer.name) for publisher in src_track.publishers: new_track.add_publisher(publisher.name) if src_track._spatial: new_track.add_spatial(src_track._spatial.isrc) return (src_track, new_track,) @classmethod def _find_next_track_position(cls, tracks, for_volume=None): """Find the next available track volume and number in tracks list. The reason for passing the tracks is so calling function can make a lock for product tracks, avoiding potential race conditions. Args: tracks (list): List of tracks belonging to a product for_volume (int): Volume Number to check Returns: tuple: (int: volume_number, int: track_number) Raises: ValueError: Only happens when for_volume is out of seq or range """ # Raise exception if for_volume is invalid volume number if for_volume is not None and \ not cls._is_track_volume_in_range(for_volume): raise ValueError(error.VALIDATION_SEQ_RANGE_ERROR_MSG) # Check if empty product if not tracks: if not for_volume or for_volume == 1: # The next track is the first volume and track for product return (1, 1) else: # Volume number higher than 1 for empty product raise ValueError(error.VALIDATION_SEQ_INTEGRITY_ERROR_MSG) # When for_volume isn't specified, just return the next track # for last volume if for_volume is None: return (tracks[-1].volume_number, tracks[-1].track_number + 1) # Otherwise if for_volume is specified, find the next track # in the specified volume if for_volume <= tracks[-1].volume_number: for track in reversed(tracks): if track.volume_number == for_volume: return (track.volume_number, track.track_number + 1) # Check if for_volume is part of a new volume by seeing # if the previous volume exists if tracks[-1].volume_number + 1 == for_volume: return (for_volume, 1) # An out of sequence volume was given raise ValueError(error.VALIDATION_SEQ_INTEGRITY_ERROR_MSG) @classmethod def _validate_tracks_order(cls, tracks): """Validate tracks are properly ordered. Args: tracks (list): List of Track objects. Returns: tuple: (bool: track order is valid, string: error_code if invalid) """ # List of track numbers indexed by volume_number track_volumes_arrgt = defaultdict(list) # Populate track_volumes_arrgt. Should look like this after loop: # {1: [1, 2], 2: [1, 2, 3], 3: [1]} for track in tracks: # Make sure integers are being used if not isinstance(track.volume_number, int) or \ not isinstance(track.track_number, int): return False, error.VALIDATION_ERROR_CODE # Builds volume and track data structure track_volumes_arrgt[track.volume_number].\ append(track.track_number) # Verify Volumes (the keys of the dict) volume_numbers = track_volumes_arrgt.keys() is_valid, error_code = is_list_unique_and_contiguous( volume_numbers, starts_with=1) if not is_valid: return False, error_code # Go through each volume and verify track list for volume_number in volume_numbers: track_list = track_volumes_arrgt[volume_number] is_valid, error_code = is_list_unique_and_contiguous( track_list, starts_with=1) if not is_valid: return False, error_code return True, '' @classmethod def _is_track_volume_in_range(cls, volume_number): """Validate track volume number is in range. Args: volume_number (int): Volume Number to check Returns: bool FIXME: This function belongs to track validation file. Since this isn't yet scoped, this function will need to be moved once that is architected. """ return volume_number >= validation.TRACK_FIELD_MIN_VOL and \ volume_number <= validation.TRACK_FIELD_MAX_VOL @classmethod def _role_type_to_field(cls, role_type): """Get track field for role_type. Args: role_type (str): A valid role type Returns: str: The track field the role belongs to Raises: ValueError: Invalid role type """ if role_type in track_role_const.ARTISTS: return track_field.ARTISTS elif role_type in track_role_const.PUBLISHERS: return track_field.PUBLISHERS elif role_type in track_role_const.WRITERS: return track_field.WRITERS else: raise ValueError(error.INVALID_ROLE_ERROR_MSG.format(role_type)) @classmethod def _apply_track_update( cls, track, data, update_track_volume_numbers=False, artist_info_ids_to_set=None): """Apply data update to track. Args: track (object): Track model object data (dict): Data to apply Return: response.Response: Contains status code and any error messages """ # Validate data doesn't update certain fields if not update_track_volume_numbers: invalid_fields = ( track_field.VOLUME_NUMBER, track_field.TRACK_NUMBER ) else: invalid_fields = tuple() for field, value in data.items(): if field in invalid_fields: return api_utils.create_validation_error_response( error.VALIDATION_ERROR_IMMUTABLE_FIELD_MSG.format(field)) recording_country_id = data.get(track_field.RECORDING_COUNTRY_ID) if recording_country_id and \ track.recording_country_id != recording_country_id: # FIXME: Validate recording_country_id using ows_territories pass # Parse out fields that updates role data track_update_data = {} role_names_update_data = {} # Fix data from legacy systems may not have been persisted correctly if track.us_publishing_obligation == '': track_update_data[track_field.US_PUBLISHING_OBLIGATION] = None if track.third_party_publisher == '': track_update_data[track_field.THIRD_PARTY_PUBLISHER] = 'N' # Create update dictionaries artist_info_ids_for_artists = data.pop( 'artist_info_ids_for_artists', []) artist_info_ids_for_writers = data.pop( 'artist_info_ids_for_writers', []) focus_track = data.pop('focus_track', None) focus_track_state_date = data.pop('focus_track_start_date', None) focus_track_end_date = data.pop('focus_track_end_date', None) user_info = {} if {'user_id', 'user_type'} <= data.keys(): user_info['user_id'] = data.pop('user_id') user_info['user_type'] = data.pop('user_type') for field, value in data.items(): if field.endswith(NAMES_FIELD_MODIFIER): # Field is a list of role names role_type = field[:-len(NAMES_FIELD_MODIFIER)] try: cls._role_type_to_field(role_type) role_names_update_data[role_type] = value except ValueError as e: return response.create_error_response( code=error.INVALID_ROLE_ERROR_CODE, message=str(e)) elif field == track_field.LANGUAGE: # We don't want to update any language data in the language table # because we are only joining on the language table to get the iso codes g.log.warning('An attempt to update track language object was made') continue elif field in track_field.BASIC_MODEL_FIELDS: # Standard field track_update_data[field] = value else: # Invalid field return api_utils.create_validation_error_response( error.VALIDATION_ERROR_INVALID_FIELD_MSG.format(field)) try: track.update(**track_update_data) for role_type, role_names in role_names_update_data.items(): track.sync_role(role_type, role_names) if artist_info_ids_to_set: track.set_artist_info_ids(artist_info_ids_to_set) track.update_focus_track_info( focus_track, focus_track_state_date, focus_track_end_date, user_info ) for track_artist_id, artist_info_id in artist_info_ids_for_artists: track.update_artist_info_id_for_artist( track_artist_id, artist_info_id ) for track_writer_id, artist_info_id in artist_info_ids_for_writers: track.update_artist_info_id_for_writer( track_writer_id, artist_info_id ) return api_utils.create_ok_response() except (TypeError, ValueError) as e: return api_utils.create_validation_error_response(str(e)) except IntegrityError as e: return response.create_error_response( code=error.DUPLICATE_CODE, message=str(e)) except sqlalchemy.exc.IntegrityError as e: return api_utils.create_validation_error_response(str(e.orig)) @classmethod def _make_track_response(cls, track): """Create response object for Track object. Args: track (object): Track to create response object for. Passing None will create a not found response. Returns: response.Response: Contains status code and payload """ if track: status = response_code.OK return response.Response( message=track.to_dict(), status=status) return response.create_not_found_response() @classmethod def _format_artists_for_update(cls, update_data): """Formats artists data for track update.""" if track_field.ARTISTS in update_data: # Initialize all the artists roles with an empty list for role_type in track_role_const.ARTISTS: update_data['{}_names'.format(role_type)] = [] update_data['artist_info_ids_for_artists'] = [] artists = update_data.pop(track_field.ARTISTS) for artist in artists: track_artist_id = artist.get('track_artist_id') artist_info_id = artist.get('artist_info_id') if artist_info_id and not track_artist_id: return api_utils.create_validation_error_response() role_type = artist.get(track_field.TYPE) role_type_key = '{}_names'.format(role_type) name = artist.get(track_field.NAME, '') if role_type_key not in update_data: return api_utils.create_validation_error_response( error.INVALID_ARTIST_TYPE_ERROR_MSG.format(role_type)) update_data[role_type_key].append(name) if track_artist_id and artist_info_id: update_data['artist_info_ids_for_artists'].append( (track_artist_id, artist_info_id,)) return api_utils.create_ok_response() @classmethod def _format_writers_for_update(cls, update_data): """Formats writers data for track update.""" if track_field.WRITERS in update_data: update_data[track_field.WRITER_NAMES] = [] update_data['artist_info_ids_for_writers'] = [] writers = update_data.pop(track_field.WRITERS) for writer in writers: update_data[track_field.WRITER_NAMES].append( writer.get(track_field.NAME, '')) track_writer_id = writer.get('track_writer_id') artist_info_id = writer.get('artist_info_id') if artist_info_id and not track_writer_id: return api_utils.create_validation_error_response() if track_writer_id and artist_info_id: update_data['artist_info_ids_for_writers'].append( (track_writer_id, artist_info_id,)) return api_utils.create_ok_response() @classmethod def _format_artists_and_writers_for_update(cls, update_data): """Formats artists and writers data for track update.""" format_artists_response = cls._format_artists_for_update(update_data) if not format_artists_response: return format_artists_response format_writers_response = cls._format_writers_for_update(update_data) if not format_writers_response: return format_writers_response return api_utils.create_ok_response() @classmethod def _extract_artist_ids(cls, update_data): """Extract artist_info_ids for artists.""" result = {} if track_field.ARTISTS in update_data: artists = update_data[track_field.ARTISTS] for artist in artists: track_artist_id = artist.get('track_artist_id') artist_info_id = artist.get('artist_info_id') if artist_info_id and not track_artist_id: role_type = artist.get(track_field.TYPE) name = artist.get(track_field.NAME, '') del artist['artist_info_id'] key = (role_type, name.strip()) result[key] = artist_info_id return response.Response(message=result) @classmethod def _extract_writer_ids(cls, update_data): """Formats artist_info_ids for writers.""" result = {} if track_field.WRITERS in update_data: writers = update_data[track_field.WRITERS] for writer in writers: track_writer_id = writer.get('track_writer_id') artist_info_id = writer.get('artist_info_id') if artist_info_id and not track_writer_id: name = writer.get(track_field.NAME, '') del writer['artist_info_id'] key = (name.strip(), ) result[key] = artist_info_id return response.Response(message=result) @classmethod def _extract_track_artist_and_writer_ids(cls, update_data): """Extract artist_info_ids for artists and writers.""" result = { 'artists': cls._extract_artist_ids(update_data).message, 'writers': cls._extract_writer_ids(update_data).message } return response.Response(message=result) @classmethod @mysql.db_session_wrap def get_track_by_track_artist_id(cls, track_artist_id, session): """Update a track artist. Args: track_artist_id (int): Track artist id session (object): SQLAlchemy database session Returns: response.Response: Contains status code and payload """ track_artist = session.query(TrackArtist).filter_by( track_artist_id=track_artist_id).one_or_none() if not track_artist: return response.create_not_found_response() track = TrackQuery.get_by_tuid(track_artist.tuid, session) return cls._make_track_response(track) @classmethod @mysql.db_session_wrap def get_track_artist_by_track_id(cls, tuid, session): """Get track artist. Args: tuid (int): Track unique id session (object): SQLAlchemy database session Returns: response.Response: Contains status code and payload """ track_artists = session.query(TrackArtist).filter_by(tuid=tuid).all() if not track_artists: return response.create_not_found_response() # Convert results to array of dicts items = [track_artist.to_dict() for track_artist in track_artists] return response.Response(message=items, status=response.status.OK)