"""Track Modeling.""" from connector_neo4j import get_session from sqlalchemy import BigInteger from sqlalchemy import Column from sqlalchemy import Enum from sqlalchemy import ForeignKey from sqlalchemy import Integer from sqlalchemy import orm from sqlalchemy import SmallInteger from sqlalchemy import String from backend.connectors import mysql from backend.constants import error from backend.constants import ownership_type from backend.constants import track_field as tf from backend.constants import track_role from backend.constants import us_publishing_obligation from backend.cypher import tracks as cypher from backend.exceptions import IntegrityError from backend.models.focus_track import FocusTrack from backend.models.language import Language # noqa from backend.models.track_additional_isrc import TrackAdditionalIsrc from backend.models.track_additional_isrc import TrackAdditionalIsrcType from backend.models.track_spatial import TrackSpatial # noqa from backend.utils import validation as validation_util class Track(mysql.BaseModel): """Encapsulates Track data. Represents track table in the art_relations database. """ __tablename__ = 'track' tuid = Column( 'id', Integer, primary_key=True, autoincrement=True, nullable=False) product_id = Column('release_id', Integer, nullable=False) # Basic Metadata track_name = Column(String(255)) version = Column(String(255)) meta_language_code = Column('meta_language', String( 8), ForeignKey('language.language_code')) isrc = Column(String(16)) p_info = Column('p_line', String(255)) explicit = Column( 'explicit_lyrics', Enum('Y', 'N', 'C', ''), default='N', nullable=False) upc = Column('upc', BigInteger, nullable=False) preview_start_time = Column( 'preview_start_time', Integer, default=None, nullable=True) # Track Duration - Legacy fields that duplicate asset info duration_minutes = Column('length_minute', SmallInteger) duration_seconds = Column('length_seconds', SmallInteger) # Track Ordering volume_number = Column('cd', SmallInteger, nullable=False) track_number = Column('track_id', SmallInteger, nullable=False) track_type = Column( 'track_type', Enum(tf.TRACK_TYPE_MUSIC, tf.TRACK_TYPE_VIDEO), default=tf.TRACK_TYPE_MUSIC ) offer_type = Column( 'offer_type', Enum( tf.OFFER_TYPE_ALL, tf.OFFER_TYPE_ALBUM_DOWNLOAD_ONLY, tf.OFFER_TYPE_TRACK_DOWNLOAD_ONLY, tf.OFFER_TYPE_TRACK_DOWNLOAD_STREAM, tf.OFFER_TYPE_ALBUM_TRACK_DOWNLOAD, tf.OFFER_TYPE_ALBUM_DOWNLOAD_STREAM, tf.OFFER_TYPE_STREAM_ONLY, tf.OFFER_TYPE_NONE, ), default=tf.OFFER_TYPE_ALL, nullable=True ) # Master Rights Info (including next 2 relationships) recording_country_id = Column('recording_country', SmallInteger) # Use ownership_rights getter/setter _master_rights = orm.relationship( 'TrackMasterRights', backref='track', uselist=False, cascade='all, delete-orphan') # Use original_rights_holder_country_id getter/setter _producer_nationality = orm.relationship( 'TrackProducerNationality', backref='track', uselist=False, cascade='all, delete-orphan') # Related langauge data language = orm.relationship( Language, foreign_keys='Track.meta_language_code', backref='track') # Related artist data artists = orm.relationship( 'TrackArtist', order_by='TrackArtist.track_artist_id', backref='track', cascade='all, delete-orphan') # Related writer data writers = orm.relationship( 'TrackWriter', order_by='TrackWriter.track_writer_id', backref='track', cascade='all, delete-orphan') # Publishing Obligation us_publishing_obligation = Column( 'us_publishing_obligation', Enum('', *us_publishing_obligation.OBLIGATION_ENUM), nullable=True) third_party_publisher = Column( 'third_party_publisher', Enum('', 'Y', 'N'), default='N', nullable=True) publishers = orm.relationship( 'TrackPublisher', order_by='TrackPublisher.track_publisher_id', backref='track', cascade='all, delete-orphan') instant_grats = orm.relationship( 'InstantGrat', order_by='InstantGrat.tuid', backref='track', cascade='all, delete-orphan') # Focus track data _focus_track = orm.relationship( 'FocusTrack', backref='track', uselist=False, cascade='all, delete-orphan') _spatial = orm.relationship( 'TrackSpatial', backref='track', uselist=False, cascade='all, delete-orphan') # Lets Track.add_spatial (the clone path) cascade-insert the dual-write # mirror alongside _spatial; normal writes mirror via create_track_spatial. _additional_isrcs = orm.relationship( 'TrackAdditionalIsrc', backref='track', uselist=True, cascade='all, delete-orphan') @property def ownership_rights(self): """Indicate ownership rights of track. Returns: string: The ownership type. None returned if not set. """ if self._master_rights: return self._master_rights.ownership_rights return None @ownership_rights.setter def ownership_rights(self, value): """Set the ownership type. The ownership_type constants are valid values. Args: value (string): String indicating ownership type, None to unset. Raises: ValueError: When invalid ownership type is passed in. """ if value: if not self._master_rights: self._master_rights = TrackMasterRights(tuid=self.tuid) self._master_rights.ownership_rights = value elif self._master_rights: del self._master_rights @property def is_owner(self): """Indicate track has an owner. Returns: string: The ownership type. None returned if not set. """ if self._master_rights: return self._master_rights.is_owner return None @is_owner.setter def is_owner(self, value): """Set the is_owner type. The is_owner can be y, n or None. Args: value (string): String indicating is_owner field, None to unset. Raises: ValueError: When invalid is_owner type is passed in. """ if value: if not self._master_rights: self._master_rights = TrackMasterRights(tuid=self.tuid) self._master_rights.is_owner = value elif self._master_rights: del self._master_rights @property def original_rights_holder_country_id(self): """Indicate country_id of producer nationality. Returns: int: country_id if set, otherwise None """ if self._producer_nationality: return self._producer_nationality.country_id return None @original_rights_holder_country_id.setter def original_rights_holder_country_id(self, value): """Set the original rights holder country_id. Args: value (int): Foreign key to country table. If None, will delete the row in track_producer_nationality Raises: sqlalchemy.exc.IntegrityError """ if value: if not self._producer_nationality: self._producer_nationality = TrackProducerNationality( tuid=self.tuid) self._producer_nationality.country_id = value elif self._producer_nationality: del self._producer_nationality @property def focus_track(self): """Indicate if track is a focus track. Returns: string: Y or N based on if track is a focus track. """ if self._focus_track: return 'Y' return 'N' @focus_track.setter def focus_track(self, value): """Delete or create focus track. Args: value (str): Y or N indicating if track is focus track """ if value == 'N' and self._focus_track: del self._focus_track elif value == 'Y' and not self._focus_track: self._focus_track = FocusTrack(tuid=self.tuid) @property def focus_track_start_date(self): """Indicate focus track start date. Returns: datetime.date: Date value. None if not a focus track. """ if self._focus_track: return self._focus_track.start_date return None @focus_track_start_date.setter def focus_track_start_date(self, value): """Set focus track start_date. Args: value (datetime.date): Date value """ if value: if not self._focus_track: self._focus_track = FocusTrack(tuid=self.tuid) self._focus_track.start_date = value elif self._focus_track: del self._focus_track @property def focus_track_end_date(self): """Indicate focus track end date. Returns: datetime.date: Date value. None if not set or not a focus track. """ if self._focus_track: return self._focus_track.end_date return None @focus_track_end_date.setter def focus_track_end_date(self, value): """Set focus track end_date. Args: value (datetime.date): date """ if self._focus_track: self._focus_track.end_date = value def update_focus_track_info(self, focus_track, start_date, end_date, user_info): """Set focus tracks values. Args: focus_track (str): 'Y', 'N', or None start_date (datetime.date): date value or None end_date (datetime.date): date value or None """ if focus_track == 'N' and self._focus_track: del self._focus_track elif focus_track == 'Y': if not self._focus_track: self._focus_track = FocusTrack( tuid=self.tuid, product_id=self.product_id, created_by_user_id=user_info.get('user_id'), created_by_user_type=user_info.get('user_type')) self._focus_track.start_date = start_date self._focus_track.end_date = end_date if {'user_id', 'user_type'} <= user_info.keys(): self._focus_track.updated_by_user_id = user_info.get('user_id') self._focus_track.updated_by_user_type = user_info.get( 'user_type') def get_roles_by_type(self, role_type): """Get list of track roles filtered by type. Args: role_type (str): Role type to filter by Returns: list: List of role objects Raises: ValueError: When invalid role type is specified """ if role_type in track_role.ARTISTS: return [ artist for artist in self.artists if artist.role_type == role_type] elif role_type in track_role.PUBLISHERS: return self.publishers elif role_type in track_role.WRITERS: return self.writers else: raise ValueError(error.INVALID_ROLE_ERROR_MSG.format(role_type)) def add_role(self, role_type, name): """Add role to track. Args: role_type (str): Type of role to add name (str): Name for new role Returns: object: New role object Raises: ValueError: When invalid role type is specified """ if role_type in track_role.ARTISTS: return self.add_artist(role_type, name) elif role_type in track_role.PUBLISHERS: return self.add_publisher(name) elif role_type in track_role.WRITERS: return self.add_writer(name) else: raise ValueError(error.INVALID_ROLE_ERROR_MSG.format(role_type)) def remove_role(self, role_type, role): """Remove role from track. Args: role_type (str): Type of role role (object): Role object to remove Returns: bool: Artist was successfully removed Raises: ValueError: When invalid role type is specified """ if role_type in track_role.ARTISTS: return self.remove_artist(role.track_artist_id, role_type) elif role_type in track_role.PUBLISHERS: return self.remove_publisher(role.track_publisher_id) elif role_type in track_role.WRITERS: return self.remove_writer(role.track_writer_id) else: raise ValueError(error.INVALID_ROLE_ERROR_MSG.format(role_type)) def sync_role(self, role_type, role_names): """Sync track to have roles in same order as passed in role_names. Args: role_type (str): Role to sync role_names (list): List of role names """ if role_names is None: role_names = [] existing_track_roles = self.get_roles_by_type(role_type) role_names = [name.strip() for name in role_names] if len(role_names) != len(set(role_names)): raise IntegrityError( error.DUPLICATE_MSG.format(role_type + 's')) # Go through existing roles and update, or add new roles as needed keep_track_roles = existing_track_roles[:len(role_names)] index = 0 for name in role_names: if index < len(keep_track_roles): keep_track_roles[index].name = name else: self.add_role(role_type, name) index += 1 # Remove any leftover roles remove_track_roles = existing_track_roles[len(role_names):] for role in remove_track_roles: self.remove_role(role_type, role) def add_artist(self, artist_type, artist_name): """Add artist to track. Makes sure arist doesn't already exist before adding. Args: artist_type (string): Artist type (performer, featured, etc...) artist_name (string): Artist Name Returns: TrackArtist object Raises: ValueError: When invalid artist type is specified """ if not self._is_valid_artist_type(artist_type): raise ValueError( error.INVALID_ARTIST_TYPE_ERROR_MSG.format(artist_type)) artist_name = artist_name.strip() for artist in self.artists: if artist_type == artist.artist_type.lower() and \ artist_name.lower() == artist.artist_name.lower(): raise IntegrityError( error.DUPLICATE_MSG.format(tf.ARTISTS)) artist = TrackArtist( tuid=self.tuid, artist_type=artist_type, artist_name=artist_name) self.artists.append(artist) return artist def update_artist_info_id_for_artist(self, track_artist_id, artist_info_id): # noqa """Update artist info ID. Handles updating the artist info ID column on a track artist. Args: track_artist_id (str): Track artist ID artist_info_id (int): Artist info ID """ for artist in self.artists: if str(artist.track_artist_id) == track_artist_id: artist.artist_info_id = artist_info_id def update_artist_info_id_for_writer(self, track_writer_id, artist_info_id): # noqa """Update artist info ID. Handles updating the artist info ID column on a track writer. Args: track_writer_id (str): Track artist ID artist_info_id (int): Artist info ID """ for writer in self.writers: if str(writer.track_writer_id) == track_writer_id: writer.artist_info_id = artist_info_id def remove_artist(self, track_artist_id, artist_type=None): """Remove artist from track. Args: track_artist_id (int): Primary key of track_artist artist_type (string): Optional check to validates if field matches Returns: bool: Artist was successfully removed Raises: ValueError: When invalid artist type is specified (except None) """ if not self._is_valid_artist_type(artist_type): raise ValueError( error.INVALID_ARTIST_TYPE_ERROR_MSG.format(artist_type)) for artist in self.artists: if artist.track_artist_id == track_artist_id: if artist_type and artist.artist_type != artist_type: return False self.artists.remove(artist) return True return False def add_publisher(self, publisher_name): """Add publisher to track. Makes sure publisher doesn't already exist before adding. Args: publisher_name (string): Publisher Name Returns: TrackPublisher object """ publisher_name = publisher_name.strip() for publisher in self.publishers: if publisher_name.lower() == publisher.publisher_name.lower(): raise IntegrityError( error.DUPLICATE_MSG.format(tf.PUBLISHERS)) publisher = TrackPublisher( tuid=self.tuid, publisher_name=publisher_name, upc=self.upc) self.publishers.append(publisher) return publisher def remove_publisher(self, publisher_id): """Remove publisher from track. Args: publisher_id (id): Primary key of track_publisher Returns: bool: Publisher was successfully removed """ for publisher in self.publishers: if publisher.track_publisher_id == publisher_id: self.publishers.remove(publisher) return True return False def add_spatial(self, spatial_isrc): """Add track_spatial data and its track_additional_isrc mirror.""" self._spatial = TrackSpatial(track_id=self.tuid, isrc=spatial_isrc) self._additional_isrcs.append(TrackAdditionalIsrc( track_id=self.tuid, type=TrackAdditionalIsrcType.ATMOS, isrc=spatial_isrc)) return self._spatial def add_writer(self, writer_name): """Add writer to track. Makes sure writer doesn't already exist before adding. Args: writer_name (string): Writer Name Returns: TrackWriter object """ writer_name = writer_name.strip() for writer in self.writers: if writer_name.lower() == writer.writer_name.lower(): raise IntegrityError( error.DUPLICATE_MSG.format(tf.WRITERS)) writer = TrackWriter( tuid=self.tuid, writer_name=writer_name, upc=self.upc, volume_number=self.volume_number, track_number=self.track_number) self.writers.append(writer) return writer def remove_writer(self, writer_id): """Remove writer from track. Args: writer_id (id): Primary key of track_writer Returns: bool: Writer was successfully removed """ for writer in self.writers: if writer.track_writer_id == writer_id: self.writers.remove(writer) return True return False def update(self, **kwargs): """Update track data. Handles updating all the track data and foreign relationships. Args: **kwargs: Fields to update, passed as named parameters Raises: TypeError: Invalid field type ValueError: If trying to overwrite read-only fields """ for key, val in kwargs.items(): if key not in tf.BASIC_MODEL_FIELDS: raise TypeError( error.VALIDATION_ERROR_SUPERFLUOUS_FIELD_MSG.format(key)) # Make sure data doesn't overwrite tuid, product_id, or UPC if key in (tf.TUID, tf.PRODUCT_ID,): if getattr(self, key) != val: raise ValueError( error.VALIDATION_ERROR_IMMUTABLE_FIELD_MSG.format( key)) setattr(self, key, val) def to_dict(self, filter_by_fields=None): """Convert the track data to dict. Args: filter_by_fields (list): Returned dict only contains the listed fields and the tuid Returns: dict: Dictionary of track data """ result = {} basic_model_fields = tf.BASIC_MODEL_FIELDS for field in basic_model_fields: result[field] = getattr(self, field) # Some fields could be set to blank that should be NULL if result[tf.US_PUBLISHING_OBLIGATION] == '': result[tf.US_PUBLISHING_OBLIGATION] = None if result[tf.THIRD_PARTY_PUBLISHER] == '': result[tf.THIRD_PARTY_PUBLISHER] = None # Add artists result[tf.ARTISTS] = [artist.to_dict() for artist in self.artists] # Add publishers result[tf.PUBLISHERS] = [pub.to_dict() for pub in self.publishers] # Add writers result[tf.WRITERS] = [writer.to_dict() for writer in self.writers] # Add language result[tf.LANGUAGE] = self.language.to_dict() if self.language else None if filter_by_fields: filtered_result = { key: result[key] for key in filter_by_fields} filtered_result[tf.TUID] = self.tuid return filtered_result return result def to_light_dict(self): """Convert the track data to light dict. Returns: dict: Dictionary of track data """ result = {} for field in tf.LIGHT_BASIC_MODEL_FIELDS: result[field] = getattr(self, field) return result def to_medium_dict(self): """Convert the track data to medium dict. Returns: dict: Dictionary of track data """ result = {} for field in tf.MEDIUM_BASIC_MODEL_FIELDS: result[field] = getattr(self, field) return result def overview_to_dict(self): """Convert the track data to dict for is_overview param. Returns: dict: Dictionary of track data """ result = {} for field in tf.OVERVIEW_MODEL_FIELDS: result[field] = getattr(self, field) # Add artists result[tf.ARTISTS] = [artist.to_dict() for artist in self.artists] return result def columns_values_to_dict(self, fields): """Convert the track data to dict. Returns: dict: Dictionary of track data. """ return {field: getattr(self, field) for field in fields} @orm.validates(tf.DURATION_MINUTES) def validate_duration_minutes(self, key, value): """Validate duration_minutes. Args: key (string): Name of column value (None, int): New duration for minutes part Raises: ValueError """ if value is None or value >= 0: return value raise ValueError('Must be null, or an integer of 0 or higher') @orm.validates(tf.DURATION_SECONDS) def validate_duration_seconds(self, key, value): """Validate duration_seconds. Args: key (string): Name of column value (None, int): New duration for seconds part Raises: ValueError """ if value is None or 0 <= value < 60: return value raise ValueError( 'Must be null, or an integer equal to or between 0 and 59') def _is_valid_artist_type(self, artist_type): """Return True if artist_type is valid. Args: artist_type (str): The artist type to test Returns: bool: Is valid artist_type """ return artist_type in track_role.ARTISTS def set_artist_info_ids(self, artist_info_ids_to_set): """Update track artists and writers with artist_info_id. Args: artist_info_ids_to_set (dict): artist_info_ids mapping. """ if artist_info_ids_to_set.get('artists', []): artists_mapping = artist_info_ids_to_set['artists'] for artist in self.artists: key = (artist.role_type, artist.name, ) if artists_mapping.get(key): artist_info_id = artists_mapping[key] artist.artist_info_id = artist_info_id if artist_info_ids_to_set.get('writers', []): writers_mapping = artist_info_ids_to_set['writers'] for writer in self.writers: key = (writer.name,) if writers_mapping.get(key): artist_info_id = writers_mapping[key] writer.artist_info_id = artist_info_id class TrackArtist(mysql.BaseModel): """Encapsulates Track Artist data. Represents track_artist table in the art_relations database. This table has a one to many relationship to the Track table. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'track_artist' track_artist_id = Column( 'id', Integer, primary_key=True, autoincrement=True, nullable=False) tuid = Column( 'track_id', ForeignKey('track.id'), nullable=False) artist_type = Column('type', String(50), nullable=False) artist_name = Column('name', String(255), nullable=False) artist_info_id = Column('artist_info_id', Integer, nullable=True) @property def name(self): """Property to get artist_name.""" return self.artist_name @name.setter def name(self, value): """Set the name. Args: value (str): Name to set """ self.artist_name = value @property def role_type(self): """Property to make uniform API for related track data.""" return self.artist_type @orm.validates('artist_type', 'artist_name') def validate_artist_name(self, key, value): """Validate artist name. Since artist_type can be set before artist_name, we have to check when either of these fields are set. Args: key (string): Name of column value (string): Value of column Raises: ValueError """ if key == 'artist_name' and self.role_type: _raise_if_invalid_role_name(value, self.role_type) elif key == 'artist_type' and self.name is not None: _raise_if_invalid_role_name(self.name, value) return value def to_dict(self, include_tuid=False): """Convert the track artist data to dict. Args: include_tuid (bool): Adds track tuid to dict """ data = { 'track_artist_id': self.track_artist_id, 'type': self.artist_type, 'name': self.artist_name, 'artist_info_id': self.artist_info_id } if include_tuid: data[tf.TUID] = self.tuid return data class TrackMasterRights(mysql.BaseModel): """Encapsulates Track Master Rights data. Represents track_master_rights table in the art_relations database. This table has a one to one relationship to the Track table. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'track_master_rights' master_rights_id = Column( Integer, primary_key=True, autoincrement=True, nullable=False) tuid = Column('track_id', ForeignKey('track.id'), nullable=False) is_owner = Column(Enum('y', 'n')) rights_level = Column(Enum( ownership_type.ORIGINAL_OWNER, ownership_type.ACQUIRED_RIGHTS, ownership_type.EXCLUSIVE_LICENSEE, ownership_type.NON_EXCLUSIVE_LICENSEE, ownership_type.NO_RIGHTS)) @property def ownership_rights(self): """Indicate ownership rights of track. Returns: string: Indicates ownership type. None returned if not set. """ return self.rights_level @ownership_rights.setter def ownership_rights(self, value): """Set the ownership type. Args: value (string): String indicating ownership type, None to unset. Raises: ValueError: When invalid ownership type is passed in. """ if value not in ownership_type.OWNERSHIP_TYPES: raise ValueError(error.INVALID_OWNERSHIP_TYPE_MSG.format( ', '.join(ownership_type.OWNERSHIP_TYPES))) self.rights_level = value self.is_owner = 'y' if ownership_type.IS_OWNER_MAP[value] else 'n' class TrackProducerNationality(mysql.BaseModel): """Encapsulates Track Producer Nationality data. Represents track_producer_nationality table in the art_relations database. This table has a one to one relationship to the Track table. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'track_producer_nationality' track_producer_nationality_id = Column( 'id', Integer, primary_key=True, autoincrement=True, nullable=False) tuid = Column('track_id', ForeignKey('track.id'), nullable=False) # This has a foreign key constraint into the country table, so invalid # values will just throw an exception. country_id = Column( 'nationality_country_id', SmallInteger, nullable=False) class TrackPublisher(mysql.BaseModel): """Encapsulates Track Publisher data. Represents track_publisher table in the art_relation database. This table has a one to many relationship to the Track table. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'track_publisher' track_publisher_id = Column( Integer, primary_key=True, autoincrement=True, nullable=False) tuid = Column( 'unique_track_id', ForeignKey('track.id'), nullable=False) publisher_name = Column(String(255), nullable=False) # It is unknown if a legacy process is using the UPC column here upc = Column('upc', BigInteger) @property def name(self): """Property to get publisher_name.""" return self.publisher_name @name.setter def name(self, value): """Set the name. Args: value (str): Name to set """ self.publisher_name = value @property def role_type(self): """Property to make uniform API for related track data.""" return 'publisher' @orm.validates('publisher_name') def validate_publisher_name(self, key, name): """Validate artist name. Args: key (string): Name of column name (string): New publisher name Raises: ValueError """ _raise_if_invalid_role_name(name, self.role_type) return name def to_dict(self): """Convert the track artist data to dict.""" return { 'track_publisher_id': self.track_publisher_id, 'type': 'publisher', 'name': self.publisher_name } class TrackWriter(mysql.BaseModel): """Encapsulates Track Writer data. Represents track_writer table in the art_relation database. This table has a one to many relationship to the Track table. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'track_writer' track_writer_id = Column( Integer, primary_key=True, autoincrement=True, nullable=False) tuid = Column('unique_track_id', ForeignKey('track.id'), nullable=False) writer_name = Column(String(255), nullable=False) artist_info_id = Column('artist_info_id', Integer, nullable=True) # Redundant fields that are already populated in the track table. # Unfortunately, these fields need to be included or the database will # throw a fit since these columns cannot be null. It is also unknown if a # legacy process is using the UPC field. upc = Column('upc', BigInteger, nullable=False) volume_number = Column('cd', SmallInteger, nullable=False) track_number = Column('track_id', SmallInteger, nullable=False) @property def name(self): """Property to get writer_name.""" return self.writer_name @name.setter def name(self, value): """Set the name. Args: value (str): Name to set """ self.writer_name = value @property def role_type(self): """Property to make uniform API for related track data.""" return 'writer' @orm.validates('writer_name') def validate_writer_name(self, key, name): """Validate writer name. Args: key (string): Name of column name (string): New writer name Raises: ValueError """ _raise_if_invalid_role_name(name, self.role_type) return name def to_dict(self): """Convert the track writer data to dict.""" return { 'track_writer_id': self.track_writer_id, 'type': 'writer', 'name': self.writer_name, 'artist_info_id': self.artist_info_id } class Releases(mysql.BaseModel): """Encapsulates Releases data. Represents releases table in the art_relation database. This table has a one to many relationship to the Track table. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'releases' release_id = Column( Integer, primary_key=True, nullable=False) release_name = Column(String(255), nullable=True) label = Column(String(255), nullable=True) release_status = Column(Enum('orchard_processing', 'label_confirmation', 'transfer_to_content', 'label_processing', 'in_content'), nullable=False) project_id = Column(Integer, nullable=True) distribution_format_id = Column(Integer, nullable=True) genre_id = Column(SmallInteger, nullable=True) def to_dict(self): """Convert the releases data to dict.""" return { 'release_id': self.release_id, 'release_name': self.release_name, 'release_status': self.release_status, 'project_id': self.project_id, 'distribution_format_id': self.distribution_format_id, 'genre_id': self.genre_id } class Project(mysql.BaseModel): """Encapsulates Project data. Represents project table in the art_relation database. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'project' project_id = Column(Integer, primary_key=True, nullable=False) vendor_id = Column(Integer, nullable=False) subaccount_id = Column(Integer, nullable=True) artist_id = Column(Integer, nullable=True) def to_dict(self): """Convert the project data to dict.""" return { 'project_id': self.project_id, 'vendor_id': self.vendor_id, 'subaccount_id': self.subaccount_id } class ArtistInfo(mysql.BaseModel): """Encapsulates ArtistInfo data. Represents artist_info table in the art_relations database. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'artist_info' artist_id = Column(Integer, primary_key=True, nullable=False) name = Column(String(255), nullable=True) class ReleaseArtist(mysql.BaseModel): """Encapsulates ReleaseArtist data. Represents release_artist table in the art_relations database. This class shouldn't be referenced outside this file except in the case of running unit tests. """ __tablename__ = 'release_artist' release_artist_id = Column(Integer, primary_key=True, nullable=False) release_id = Column(Integer, nullable=False) role = Column(String(50), nullable=True) artist_name = Column(String(255), nullable=True) def _raise_if_invalid_role_name(role_name, role_type): """Validate name of role for role_type. Args: role_name (string): Role name role_type (string): Type of role Raises: ValueError """ if not role_name.strip(): raise ValueError( error.VALIDATION_ERROR_BLANK_NAME_MSG.format( role_type.title())) if role_type in track_role.ARTISTS \ and not validation_util.is_valid_artist_name(role_name): raise ValueError( error.INVALID_ARTIST_NAME_ERROR_MSG.format(role_name)) def get_tracks_by_osrid(osr_id, limit, offset): """Get a paginated list of tracks by OrchardSoundRecording ID. When offset is 0, the primary track is returned first. Args: osr_id (str): The OrchardSoundRecording to search by. limit (int): The number of tracks to return per page. offset (int): The offset to return a page of results from. Returns: dict: A standard pagination object with total_records and items[]. """ neo4j_session = get_session() results = neo4j_session.run( cypher.GET_TRACKS_BY_OSRID, osr_id=osr_id, limit=limit, offset=offset ) total_count = 0 tracks = [] for result in results: total_count = result['totalCount'] tracks = result['tracks'] break formatted_tracks = [] for track in tracks: formatted_tracks.append({ 'id': track.get('id'), 'isrc': track.get('isrc'), }) return { 'total_records': total_count, 'items': formatted_tracks }