"""Database Persister. This module runs the abstracted, low level database operations needed by the API at the HTTP handler level. The structure of these table configs are: { 'name': , 'columns': { : || None } } If a database column is mapped to "None", that means it may only be set explicitly in code, rather than auto-mapped from incoming json fields. """ import datetime import decimal from flask import g from oto.response import create_error_response from oto.response import create_not_found_response from oto.response import Response import sentry_sdk from sqlalchemy import exc from ows_product_physical.connector.mysql import db_session from ows_product_physical.constant import error from ows_product_physical.constant import field from ows_product_physical.constant.store import ALL_STORE_IDS from ows_product_physical.models import ows_marketing from ows_product_physical.models import release_status as ReleaseStatus # noqa from ows_product_physical.models import release_subgenre as ReleaseSubgenre # noqa from ows_product_physical.models import release_artist as ReleaseArtist # noqa from ows_product_physical.models import release_approval_queue as ReleaseApprovalQueue # noqa from ows_product_physical.models import product_physical as ProductPhysical # noqa from ows_product_physical.models import product_physical_change_history \ as ProductPhysicalChangeHistory from ows_product_physical.models import releases as Releases # noqa from ows_product_physical.models import track as Track # noqa from ows_product_physical.models.ows_product import get_provisioned_upc from ows_product_physical.models import ows_carveouts from ows_product_physical.models.product_physical_packaging import \ ProductPhysicalPackaging from ows_product_physical.models import product_physical_supply_chain_metadata\ as ProductPhysicalSupplyChainMetadata from ows_product_physical.models.releases import Releases as ReleasesModel from ows_product_physical.models.sql import product_physical def _map_releases_fields_to_dict(fields): """Map releases fields to database column names. Args: fields (dict): dictionary of physical product data """ values = {key: fields[key] for key in field.RELEASES_FIELDS if key in fields.keys()} product_type = fields.get(field.PRODUCT_TYPE) if product_type: values['new_release'] = product_type return values def _create_release(session, fields, subaccount_id, project_code, artist_id): """Create the release db record. Args: session (sqlalchemy.orm.session.Session): db session. fields: (dict) product to insert. subaccount_id (string|int): the subaccount_id for this product. project_code (string): project code of the related project Returns: Response: the result of the query. """ values = _map_releases_fields_to_dict(fields) values.update({ 'subaccount_id': subaccount_id, 'vendor_catalog_number': project_code, 'release_status': field.DEFAULT_RELEASE_STATUS_STATUS, 'artist_id': artist_id }) try: release = Releases.create(values, session) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) return Response( status=201, message={'release_id': release.get('release_id')}) def _create_release_approval_queue(session, release_id, vendor_id): """Create the release_approval_queue db record. Args: session (sqlalchemy.orm.session.Session): db session. release_id (int): the release this record is keyed to. vendor_id (int): the vendor id the release belongs to. Returns: Response: the result of the query. """ try: ReleaseApprovalQueue.create(release_id, vendor_id, session) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error release_status', status=500) return Response(status=201) def _create_release_status( session, release_id, vendor_id, release_status=None): """Create the release_status db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): product to insert. release_id (int): the release this record is keyed to. vendor_id (int): the vendor id the release belongs to. Returns: Response: the result of the query. """ try: ReleaseStatus.create({ 'release_id': release_id, 'changed_by': vendor_id, 'status': release_status or field.DEFAULT_RELEASE_STATUS_STATUS, 'changed_by_type': field.CHANGED_BY_TYPE_VENDOR }, session) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error release_status', status=500) return Response(status=201) def _insert_release_primary_artist(session, fields, release_id): """Create the release_artist db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): product to insert. release_id (int): the release this record is keyed to. Returns: Response: the result of the query. """ try: ReleaseArtist.create({ 'release_id': release_id, 'artist_name': fields[field.PRIMARY_ARTIST], 'upc': fields[field.UPC] }, session) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) return Response(status=201) def _update_release_primary_artist(session, fields, release_id): """Create the release_artist db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): product to insert. release_id (int): the release this record is keyed to. Returns: Response: the result of the query. """ update_values = {} if fields.get(field.UPC): update_values.update({field.UPC: fields[field.UPC]}) if fields.get(field.PRIMARY_ARTIST): update_values.update( {'artist_name': fields[field.PRIMARY_ARTIST]}) if update_values: ReleaseArtist.update(update_values, release_id, session) def _map_physical_product_fields_to_dict(fields, release_id=None): """Map physical product fields to database column names. Args: fields (dict): dictionary of physical product data """ values = {key: fields[key] for key in field.PRODUCT_PHYSICAL_FIELDS if fields.get(key)} artist_is_individual = fields.get(field.ARTIST_IS_INDIVIDUAL) if artist_is_individual: values.update({'individual': artist_is_individual}) if release_id: values.update({'release_id': release_id}) return values def _create_product_physical(session, fields, release_id): """Create the product physical db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): product to insert. release_id (int): the release this record is keyed to. Returns: Response: the result of the query. """ values = _map_physical_product_fields_to_dict(fields, release_id) try: ProductPhysical.create(values, session) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) return Response(status=201) def _update_product_physical(session, fields, release_id): """Update the product physical db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): product to insert. """ values = _map_physical_product_fields_to_dict(fields) if values: ProductPhysical.update(values, release_id, session) def _update_release(session, fields, release_id): """Update the product physical db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): dictionary of fields to update for product. release_id (int): id of release to update """ values = _map_releases_fields_to_dict(fields) if field.PROJECT_CODE in fields: values[field.VENDOR_CATALOG_NUMBER] = fields[field.PROJECT_CODE] if values: Releases.update(values, release_id, session) def _update_release_status(session, release_status, release_id): """Update the product physical db record. Args: session (sqlalchemy.orm.session.Session): db session. release_status (str): updated release status for the product release_id (int): id of release to update """ try: Releases.update( {'release_status': release_status}, release_id, session) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) return Response(status=200) def _insert_release_subgenre(session, fields, release_id): """Create the release subgenre db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): product to insert. release_id (int): the release this record is keyed to. Returns: Response: the result of the query. """ try: ReleaseSubgenre.create({ 'release_id': release_id, 'upc': fields[field.UPC], 'subgenre_id': fields[field.SUBGENRE_ID] }, session) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error release_subgenre', status=500) return Response(status=201) def _update_release_subgenre(session, fields, product_id): """Create the release subgenre db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): product to insert. product_id (int): the product this record is keyed to. Returns: Response: the result of the query. """ values_to_update = {} if fields.get(field.UPC): values_to_update.update({field.UPC: fields[field.UPC]}) if fields.get(field.SUBGENRE_ID): values_to_update.update({field.SUBGENRE_ID: fields[field.SUBGENRE_ID]}) if values_to_update: ReleaseSubgenre.update(values_to_update, product_id, session) def _fetch_product_by_release_id(session, release_id): """Fetch the product. Args: session (sqlalchemy.orm.session.Session): db session. release_id (int): the release id of the product to fetch. Returns: Response: the result of the query. """ product_select_params = {field.RELEASE_ID: release_id} try: resp = session.execute( product_physical.SELECT_PRODUCT_BY_RELEASE_ID, product_select_params).fetchone() if not resp: return create_not_found_response() product = resp._mapping # string format the date fields product = _convert_dates_to_string(product) # string format the decimal fields product = _convert_decimal_to_string(product) return Response(message=product) except Exception as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) def _convert_dates_to_string(product_dict): """Convert datetime fields to string. Args: product_dict (dict): the product dictionary. Returns: product_dict (dict): the product dictionary. """ return { field: str(value) if isinstance(value, datetime.date) else value for field, value in product_dict.items()} def _convert_decimal_to_string(product_dict): """Convert decimal fields to string. Args: product_dict (dict): the product dictionary. Returns: product_dict (dict): the product dictionary. """ return { field: str(value) if isinstance(value, decimal.Decimal) else value for field, value in product_dict.items()} def get_product_physical_packaging(): """Return a list of options from product_physical_packaging. Returns: Response: Response containing the options list. """ with db_session() as session: try: results = session.query(ProductPhysicalPackaging).all() return Response( message=[result.to_dict() for result in results], status=200) except (exc.SQLAlchemyError, exc.IntegrityError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) def _assign_display_upc(session, fields): """Set display_upc field if assign_display_upc is set. Set the display_upc field in fields to an orchard assigned value if the assign_display_upc field is set. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): Product fields containing display_upc. """ if not fields.pop('assign_display_upc', False): return g.ows.log.info('Assigning display_upc...') provisioner_response = get_provisioned_upc(mark_used=True) if provisioner_response: fields[field.DISPLAY_UPC] = provisioner_response.message g.ows.log.info( 'Assigned display_upc {} from upc_provisioner'.format( fields[field.DISPLAY_UPC] ) ) return fields[field.DISPLAY_UPC] = _reserve_display_upc(session) g.ows.log.info( 'Assigned display_upc {} from SP'.format(fields['display_upc']) ) def create_product(fields, vendor_id, subaccount_id, project_code, artist_id): """Take the HTTP POST body and persists to the product db schema. Args: fields (dict): product to insert. vendor_id (string|int): the vendor_id for this product. subaccount_id (string|int): the subaccount_id for this product. project_code (string): project code of the related project Returns: Response: Response containing the created product. """ with db_session() as session: try: _assign_display_upc(session, fields) # `release` INSERT resp = _create_release( session, fields, subaccount_id, project_code, artist_id) if resp.status != 201: return resp product_id = resp.message.get('release_id') # `product_physical` INSERT resp = _create_product_physical( session, fields, product_id) if resp.status != 201: return resp # `release_status` INSERT resp = _create_release_status(session, product_id, vendor_id) if resp.status != 201: return resp # `release_artist` INSERT resp = _insert_release_primary_artist( session, fields, product_id) if resp.status != 201: return resp # subgenre? if fields.get(field.SUBGENRE_ID) is not None: # `release_subgenre` INSERT resp = _insert_release_subgenre( session, fields, product_id) if resp.status != 201: return resp g.ows.log.info( 'Create new physical fields, product_id: {}'.format( product_id)) return Response(message={'product_id': product_id}, status=201) except Exception as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) def update_product(product_id, fields): """Update product. Args: product_id: id fields (dict): updated fields. Returns: Response: Response containing the updated product. """ with db_session() as session: try: _assign_display_upc(session, fields) _update_release(session, fields, product_id) _update_product_physical(session, fields, product_id) _update_release_subgenre(session, fields, product_id) _update_release_primary_artist(session, fields, product_id) g.ows.log.info( 'Update physical product fields, product_id: {}'.format( product_id)) return Response(status=200) except (exc.SQLAlchemyError, exc.IntegrityError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) def update_product_highlights(product_id, product_highlights_description): """Update product highlights. Args: product_id: id product_highlights_description: text description Returns: Response: Response containing the updated product highlights. """ product_highlights = \ ows_marketing.get_product_highlight_by_product_id( product_id) if product_highlights.status not in [200, 404]: return product_highlights if product_highlights.status == 200: resp = ows_marketing.update_product_highlight( product_highlights.message.get('highlight_id'), product_highlights_description) elif product_highlights.status == 404: resp = ows_marketing.create_product_highlight( product_id, product_highlights_description) return resp def update_product_status(product_id, product_status, vendor_id): """Update product. Args: product_id: id product_status: the status to update to in the approval flow. vendor_id: id of the vendor that owns this project/product. Returns: Response: Response containing the updated product. """ with db_session() as session: release_response = _update_release_status( session, product_status, product_id) if not release_response: return release_response release_status_response = _create_release_status( session, product_id, vendor_id, product_status) if release_status_response.status != 201: return release_status_response if product_status == 'transfer_to_content': release_approval_queue_response = _create_release_approval_queue( session, product_id, vendor_id) if release_approval_queue_response.status != 201: return release_approval_queue_response g.ows.log.info( 'Update physical product status, product_id: {}'.format( product_id)) return release_status_response def get_product_by_id(product_id): """Fetch a product by primary key. Args: product_id (int): the id of the product to fetch. Returns: Response: Response containing the created product. """ with db_session() as session: try: return _fetch_product_by_release_id(session, product_id) except (exc.SQLAlchemyError, exc.IntegrityError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) def get_release_by_product_code(product_code, product_id=''): """Fetch a release id by a product code. Args: product_code(string|int): product code of the product. product_id (string|int): the id of the product to fetch if exists. Returns: Response: Response containing the result of fetching. """ filters = [ (ReleasesModel.product_code == product_code), (ReleasesModel.release_id != product_id) ] with db_session() as session: try: response = session.query( ReleasesModel.release_id, ReleasesModel.subaccount_id).filter(*filters).all() zip_with = ['release_id', 'subaccount_id'] return Response( [dict(zip(zip_with, row)) for row in response]) except Exception as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) def delete_product(release_id): """Delete a physical product by its release id. Args: release_id (int): release id of the product to be deleted Returns: Response: Response containing the result of deleting the product. """ try: with db_session() as session: ReleaseStatus.delete_all(release_id, session) ReleaseSubgenre.delete(release_id, session) ReleaseArtist.delete(release_id, session) ProductPhysical.delete(release_id, session) Track.delete_all_tracks(release_id, session) Releases.delete(release_id, session) except Exception as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500, ) return Response(message={'status': 'ok'}) def get_physical_release_count_by_product_code(product_code, product_id=''): """Fetch physical releases by product code. Args: product_code (str): product_code of the releases to retrieve Returns: Response: Response containing errors or the releases. """ with db_session() as session: try: select_params = { 'product_code': product_code, 'product_id': product_id } count = session.execute( product_physical.SELECT_PHYSICAL_RELEASE_COUNT_BY_PRODUCT_CODE, select_params).fetchone()[0] return Response(message=count) except Exception as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) def create_history(**fields): """Helper to create new row in `product_physical_change_history`. Args: fields: dictionary of possible """ with db_session() as session: try: ProductPhysicalChangeHistory.create(session, **fields) except Exception as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error create_history', status=500) return Response(status=201) def get_upc_by_product_id(product_id): """Fetch a release's upc by product id. Args: product_id (id): product_id of the upc to retrieve Returns: Response: Response containing errors or the releases upc. """ with db_session() as session: try: select_params = { 'product_id': product_id } result = session.execute( product_physical.SELECT_UPC_BY_PRODUCT_ID, select_params).fetchone() return Response(message=result and result[0]) except Exception as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) def _reserve_display_upc(session): return str(session.execute(product_physical.RESERVE_UPC).scalar()) def create_product_physical_supply_chain_metadata(fields, release_id, upc=None): """Create the product physical supply chain metadata db record. Args: session (sqlalchemy.orm.session.Session): db session. fields (dict): product to insert. release_id (int): the release this record is keyed to. upc (string): UPC of the product Returns: Response: the result of the query. """ sale_start_date = fields.get(field.SALE_START_DATE) if not upc: upc = fields.get(field.UPC) carve_in_stores = get_carveins_for_upc(upc) values = { 'metadata': [ { 'store_id': store_id, 'sale_start_date': sale_start_date, 'product_id': release_id, } for store_id in carve_in_stores ] } try: ProductPhysicalSupplyChainMetadata.set_physical_supply_chain_metadata( release_id, values) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) return Response(status=201) def get_carveins_for_upc(upc): """Get carve-in store IDs for a UPC. Args: upc (str): The UPC used to fetch carveout store IDs. Returns: set[int]: The store IDs remaining after carveouts are excluded. """ carveout_ids_for_upc = ows_carveouts.get_release_carveouts_for_upc(upc) if not carveout_ids_for_upc: stores = [] if carveout_ids_for_upc: stores = list(map(int, dict(carveout_ids_for_upc.message).keys())) carveins = set(ALL_STORE_IDS) - set(stores) return carveins def update_product_physical_supply_chain_metadata(fields, release_id, upc=None): """Update the product physical supply chain metadata db record. This replaces existing supply chain metadata for the given release by deleting current rows and inserting the newly derived set. Args: fields (dict): product fields used to build metadata. release_id (int): the release this record is keyed to. Returns: Response: the result of the query. """ sale_start_date = fields.get(field.SALE_START_DATE) if not upc: upc = fields.get(field.UPC) carve_in_stores = get_carveins_for_upc(upc) values = { 'metadata': [ { 'store_id': store_id, 'sale_start_date': sale_start_date, 'product_id': release_id, } for store_id in carve_in_stores ] } try: ProductPhysicalSupplyChainMetadata.delete_supply_chain_metadata_by_product( release_id ) ProductPhysicalSupplyChainMetadata.set_physical_supply_chain_metadata( release_id, values ) except (exc.SQLAlchemyError, exc.DBAPIError) as exception: sentry_sdk.capture_exception(exception) return create_error_response( code=error.INTERNAL_ERROR, message='mysql error', status=500) return Response(status=200)