"""Logic layer for unsafe methods.""" from oto import response from sqlalchemy.orm import Session from product_digital.connectors import mysql from product_digital.constants.product_statuses import IN_CONTENT from product_digital.models import mkt_priority from product_digital.models import ows_product from product_digital.models import release from product_digital.models import release_artist from product_digital.models import release_grid from product_digital.models import release_status as release_status_model from product_digital.models import release_subgenre from product_digital.models import track from product_digital.config import DELETE_PRODUCT_NFD_VALUE @mysql.db_session_wrap def UNSAFE_hard_delete_product(upc: str, session: Session) -> response.Response: """Hard delete a product by upc. This method is "unsafe" because it has not been tested with products having NFD values other than the provided DELETE_PRODUCT_NFD_VALUE. At the time of writing, it is only a good idea to delete `SMEAnalyticsDummy` NFD values. Args: upc: str The upc of the product to be deleted. session: Session The database session to use to ensure the operation is transactional. This is provided by the db_session_wrap decorator if unspecified. Returns: response.Response A response object with a status code of 200 if the product was successfully deleted, 403 if the product does not match the DELETE_PRODUCT_NFD_VALUE, or 404 if the product does not exist. """ product_response = ows_product.get_product_by_upc(upc) if not product_response: return response.Response(status=404, message=f"No product found for upc {upc}") # If the product does not match the supplied nfd value, the # request is forbidden. product_nfd_value = product_response.message["not_for_distribution"] if product_nfd_value != DELETE_PRODUCT_NFD_VALUE: return response.Response( status=403, message=f"Product with nfd value {product_nfd_value} is not permitted to be deleted", ) product_id = product_response.message["product_id"] track_response = track.delete_all_tracks(product_id) if not track_response: return track_response release_artist.delete(product_id, session) release_status_model.delete(product_id, session) release_subgenre.delete_release_subgenre(product_id, session=session) mkt_priority.bulk_delete_mkt_priority_by_release_id(product_id, session) release_grid.delete(product_id, session) release.delete(product_id, session) return response.Response(status=200) def UNSAFE_update_display_upc(product_id, display_upc) -> response.Response: """Update product display upc. This method is "unsafe" because it not safe to have upc != display_upc. Display upc update is only allowed if : 1. Product is SMEAnalyticsDummy 2. There is another product with upc = display_upc 3. The other product nfd is not SMEAnalyticsDummy 4. The other product status is not IN_CONTENT or product is deleted Args: product_id: int The product_id to be updated. session: Session The database session to use to ensure the operation is transactional. This is provided by the db_session_wrap decorator if unspecified. Returns: response.Response A response object with a status code of 200 if the product was successfully updated, 403 if the product does not match the DELETE_PRODUCT_NFD_VALUE, or 404 if the product does not exist. """ product_response = ows_product.get_product(product_id) if not product_response: return response.Response(status=404, message=f"No product found for product_id {product_id}") # If the product does not match the supplied nfd value, the # request is forbidden. product_nfd_value = product_response.message["not_for_distribution"] if product_nfd_value != 'SMEAnalyticsDummy': return response.Response( status=403, message=f"Product with nfd value {product_nfd_value} is not permitted to have its display upc updated", ) display_upc_product = ows_product.get_product_by_upc(display_upc) if not display_upc_product: return response.Response(status=403, message=f"There is no product with upc {display_upc}, display_upc update is denied") display_upc_nfd_value = display_upc_product.message["not_for_distribution"] if display_upc_nfd_value == 'SMEAnalyticsDummy': return response.Response( status=403, message=f"Substituted product (the product with upc {display_upc})" " nfd value cannot be 'SMEAnalyticsDummy'." ) display_upc_release_status = display_upc_product.message["status"] display_upc_release_is_deleted = display_upc_product.message[ "deletions"] == 'Y' if display_upc_release_status == IN_CONTENT and not display_upc_release_is_deleted: return response.Response( status=403, message=f"Substituted product (the product with upc {display_upc}) " "product has to be either not in_content or marked as deleted." ) product_id = product_response.message["product_id"] return release.UNSAFE_update_display_upc(product_id, display_upc) def UNSAFE_soft_delete(product_id) -> response.Response: """Soft delete for specified SMEAnalyticsDummy product.""" product_response = ows_product.get_product(product_id) if not product_response: return response.Response(status=404, message=f"No product found for product_id {product_id}") product_nfd_value = product_response.message["not_for_distribution"] if product_nfd_value != 'SMEAnalyticsDummy': return response.Response( status=403, message="It is only allowed to soft delete products with nfd value 'SMEAnalyticsDummy'.", ) product_release_status = product_response.message["status"] if product_release_status != IN_CONTENT: return response.Response( status=403, message=f"It is only allowed to soft delete products with release status {IN_CONTENT}.", ) return release.UNSAFE_soft_delete(product_id)