"""Vendor Model. This vendor model uses sqlalchemy. It provides core functionality for getting vendor information by vendor_id. """ from typing import Any, Dict, List, Tuple, Union from ddtrace import tracer from flask import g from owsresponse import response from pythonfeatures.constants import split as split_constants # noqa from sqlalchemy import ( DATE, Boolean, Column, Enum, Float, ForeignKey, Integer, String, case, func, text, ) from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session, relationship from account.connectors import mysql from account.connectors.sentry import sentry_client from account.constants import constants, error from account.models import CompanyBrand, service_tier from account.models.parent_company import ParentCompany from account.models.service_tier import ServiceTier, vendor_service_tier from account.models.sql.vendor_assigned_reviewer import VENDOR_ASSIGNED_REVIEWER_SQL from account.models.sql.vendor_assigned_to import VENDOR_ASSIGNED_TO_SQL from account.models.sql.vendor_currency_codes import VENDOR_CURRENCY_CODES_SQL from account.models.sql.vendor_document import ( VENDOR_DOCUMENT_SQL, VENDOR_DOCUMENT_WITH_TENANT_UUIDS_SQL, ) from account.models.sql.vendor_label_info import ( COMPANY_BRAND_ID_BY_NAME_SQL, COMPANY_BRAND_ID_SQL, DELETE_PRODUCT_MANAGER_MAPPING_SQL, PRODUCT_MANAGER_SQL, UPDATE_COMPANY_BRAND_ID, UPDATE_SOUNDSCAN_CODE, VENDOR_CONTRACT_CURRENCY_SQL, VENDOR_LABEL_INFO_SQL, ) from account.models.sql.vendor_secondary_internal_contact import ( VENDOR_SECONDARY_INTERNAL_CONTACT_SQL, ) from account.models.types import ( CompanyBrand as CompanyBrandType, DeletedVendor, UpdatedVendor, UpdatedVendorClosers, Vendor as VendorType, ) from account.utils.exception import VendorUpdateException class OrchAdminUser(mysql.BaseModel): """OrchadminUser DB Model.""" 'used for vendor foreign key validation.' __tablename__ = 'orchadmin_users' id = Column(Integer, primary_key=True) f_name = Column(String) l_name = Column(String) is_product_manager = Column(Integer) active = Column(Enum('Y', 'N')) class VendorClosers(mysql.BaseModel): """VendorClosers DB Model""" __tablename__ = 'vendor_closers' id = Column(Integer, primary_key=True) vendor_id = Column(Integer, ForeignKey('vendor.vendor_id'), default=None) orchadmin_user_id = Column(Integer, ForeignKey('orchadmin_users.id'), default=None) last_modified_by = Column(Integer, ForeignKey('orchadmin_users.id'), default=None) class SoundScanCode(mysql.BaseModel): """Soundscan_codes DB Model.""" __tablename__ = 'soundscan_codes' id = Column(Integer, primary_key=True) vendor_id = Column(Integer) country_id = Column(Integer) soundscan_code = Column(String) subaccount_id = Column(Integer) class Vendor(mysql.BaseModel): """Vendor DB Model.""" __tablename__ = 'vendor' vendor_id = Column(Integer, primary_key=True) is_distributor = Column(Enum('Y', 'N'), default='N') migrated_to_abacus = Column(Boolean, default=False) # Relationship Manager: orchadmin user responsible for the client relationship and success. # Receives Content Review escalations. Required for vendors with status='signed'. assigned_to = Column(Integer, ForeignKey('orchadmin_users.id'), default=None) # Content Review assignee: identifies who approves client product before delivery. # Not shown in OA UI — API-only field. assigned_reviewer = Column(Integer, ForeignKey('orchadmin_users.id'), default=None) # Secondary Relationship Manager: backup point of contact for the label. # Feature-flagged in OA UI via `quarterback_label_managers_in_oa`. quarterback_label_manager = Column(Integer, ForeignKey('orchadmin_users.id'), default=None) owner = Column(String) name = Column(String) company = Column(String) company_brand_id = Column(Integer, ForeignKey('company_brand.id'), default=None) contact_email = Column(String) country_id = Column('country', Integer, default=None) date_created = Column(DATE, default=func.current_date()) status = Column(Enum(*constants.VENDOR_STATUS), default='signed') label_identifier = Column(Enum(*constants.LABEL_IDENTIFIER), default=None) label_summary = Column(String) show_release_builder = Column(Enum(*constants.BOOLEAN_IDENTIFIER), default=None) priority = Column(Integer) primary_genre = Column(Integer) region = Column(Integer) support_contact_email = Column(String) transfer_pricing_country = Column(Enum(*constants.TRANSFER_PRICING_COUNTRY), default=None) est_total_releases = Column('est_total_releases', Integer) est_total_tracks = Column('est_total_tracks', Integer) projected_first_year_revenue = Column(Float(18, True, 6)) is_owned = Column('is_owned', String, default='No') website = Column('website', String) last_update = Column('last_update', DATE) sap_vendor_id = Column('sap_vendor_id', String) # Welcome Email Sender: orchadmin user who triggered the welcome email. # Write-once audit field; OA shows it read-only once set. wel_email_sender = Column(Integer, ForeignKey('orchadmin_users.id'), default=None) wel_email_send_date = Column(DATE) external_identifier_1 = Column('external_identifier_1', String) vendor_uuid = Column(String, default=None) last_modified_by = Column(Integer, ForeignKey('orchadmin_users.id'), default=None) service_tier = relationship('ServiceTier', secondary='vendor_service_tier', uselist=False) first_statement_period = Column(Integer, default=1) relationship_notes = Column(String) newsletter = Column(Enum(*constants.BOOLEAN_IDENTIFIER), default='Y') date_signed = Column(DATE) company_brand = relationship( 'CompanyBrand', primaryjoin='Vendor.company_brand_id == CompanyBrand.id', ) country = relationship( 'Country', primaryjoin='Vendor.country_id == Country.id', foreign_keys='[Vendor.country_id]', uselist=False, ) genre = relationship( 'Genre', primaryjoin='Vendor.primary_genre == Genre.genre_id', foreign_keys='[Vendor.primary_genre]', uselist=False, ) def to_dict(self): """Get a dict representation of Vendor.""" return { 'vendor_id': self.vendor_id, 'is_distributor': self.is_distributor, 'migrated_to_abacus': self.migrated_to_abacus, 'assigned_to': self.assigned_to, 'assigned_reviewer': self.assigned_reviewer, 'quarterback_label_manager': self.quarterback_label_manager, 'owner': self.owner, 'name': self.name, 'company': self.company, 'company_brand_id': self.company_brand_id, 'contact_email': self.contact_email, 'status': self.status, 'label_identifier': self.label_identifier, 'show_release_builder': self.show_release_builder, 'priority': self.priority, 'primary_genre': self.primary_genre, 'region': self.region, 'support_contact_email': self.support_contact_email, 'transfer_pricing_country': self.transfer_pricing_country, 'first_statement_period': self.first_statement_period, 'relationship_notes': self.relationship_notes, 'country_id': self.country_id, 'is_owned': self.is_owned, 'vendor_uuid': self.vendor_uuid, 'last_modified_by': self.last_modified_by, 'service_tier': {'uuid': self.service_tier.uuid, 'name': self.service_tier.name} if self.service_tier else None, 'wel_email_sender': self.wel_email_sender, } DEFAULT_FIELDS = ( Vendor.vendor_id, Vendor.is_distributor, Vendor.owner, Vendor.name, Vendor.company, Vendor.country_id, Vendor.status, Vendor.label_identifier, Vendor.contact_email, Vendor.migrated_to_abacus, Vendor.assigned_to, Vendor.assigned_reviewer, Vendor.quarterback_label_manager, Vendor.vendor_uuid, Vendor.last_modified_by, ) CONDITION_IS_DISTRIBUTOR = Vendor.is_distributor == 'Y' # noqa def get_distributor(vendor_id): """Get distributor information for a vendor. Gets distributor information for a vendor that is a distributor. Args: vendor_id (int): unique identifier for the vendor. Returns: response.Response: containing dict of distributor or error. """ with mysql.session_scope(read_only=True) as session: row = ( session.query(*DEFAULT_FIELDS) .filter(Vendor.vendor_id == vendor_id, CONDITION_IS_DISTRIBUTOR) .one_or_none() ) if row: return response.Response(row._asdict()) return response.create_not_found_response() def get_owner(vendor_id): """Get owner information for a vendor given vendor_id. Args: vendor_id (int): unique identifier for the vendor. Returns: response.Response: containing dict of distributor or error. """ with mysql.session_scope(read_only=True) as session: row = session.query('owner').filter(Vendor.vendor_id == vendor_id).one_or_none() if row: return response.Response(row._asdict()) return response.create_not_found_response() def get_vendor(vendor_id): """Get vendor information for a vendor given vendor_id. Args: vendor_id (int): unique identifier for the vendor. Returns: response.Response: containing dict of vendor or error. """ with mysql.session_scope(read_only=True) as session: row = session.query(*DEFAULT_FIELDS).filter(Vendor.vendor_id == vendor_id).one_or_none() if row: return response.Response(row._asdict()) return response.create_not_found_response('Invalid vendor_id.') def get_vendors(vendor_uuids: list[str]): """Get vendor information for a vendors given vendor_uuids. Args: vendor_uuids (list[str]): unique identifiers for each vendor. Returns: list[dict]: list of dicts of vendors. """ try: with mysql.session_scope(read_only=True) as session: select_entities = [ Vendor.vendor_id, Vendor.vendor_uuid, Vendor.is_distributor, Vendor.owner, func.coalesce( case( [(Vendor.company == '', None)], # Replace empty string with None else_=Vendor.company, ), Vendor.name, ).label('name'), # Using coalesce for two fields Vendor.status, Vendor.company_brand_id, ] vendors = ( session.query(*select_entities).filter(Vendor.vendor_uuid.in_(vendor_uuids)).all() ) vendors_data = [v._asdict() for v in vendors] return vendors_data except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) raise Exception(e.args) def get_vendor_names(vendor_uuids: list[str]): """Get vendor names for a vendors given vendor_uuids. may raise exception. Args: vendor_uuids (list[str]): unique identifiers for each vendor. Returns: list[dict]: list of dicts of vendors. """ try: with mysql.session_scope(read_only=True) as session: select_entities = [ Vendor.vendor_id, Vendor.vendor_uuid, func.coalesce( case( [(Vendor.company == '', None)], # Replace empty string with None else_=Vendor.company, ), Vendor.name, ).label('name'), # Using coalesce for two fields ] vendors = ( session.query(*select_entities).filter(Vendor.vendor_uuid.in_(vendor_uuids)).all() ) vendors_data = [v._asdict() for v in vendors] return vendors_data except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) raise Exception(e.args) def get_vendor_company_brands(vendor_uuids: list[str]) -> list[dict[str, str]]: """Get vendor company_brands for a vendors given vendor_uuids. may raise exception. Args: vendor_uuids (list[str]): unique identifiers for each vendor. Returns: list[dict]: list of dicts of vendors. """ try: with mysql.session_scope(read_only=True) as session: select_entities = [ Vendor.vendor_id, Vendor.vendor_uuid.label('uuid'), Vendor.company_brand_id, CompanyBrand.uuid.label('company_brand_uuid'), ] join_entities = [Vendor.company_brand] vendors = ( session.query(*select_entities) .join(*join_entities) .filter(Vendor.vendor_uuid.in_(vendor_uuids)) .all() ) vendors_data = [v._asdict() for v in vendors] return vendors_data except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) raise Exception(e.args) def get_vendor_service_tier(vendor_uuids: list[str]) -> list[dict[str, Union[str, None]]]: """Get vendor service_tier for a set of vendors given vendor_uuids. may raise exception. Args: vendor_uuids (list[str]): unique identifiers for each vendor. Returns: list[dict]: list of dicts of vendors. """ try: with mysql.session_scope(read_only=True) as session: select_entities = [ Vendor.vendor_id, Vendor.vendor_uuid.label('uuid'), ServiceTier.uuid.label('service_tier_uuid'), ServiceTier.name.label('service_tier_name'), ServiceTier.display_name.label('service_tier_display_name'), ] query = ( session.query(*select_entities) .join(vendor_service_tier, Vendor.vendor_id == vendor_service_tier.c.vendor_id) .join(ServiceTier, vendor_service_tier.c.service_tier_uuid == ServiceTier.uuid) .filter(Vendor.vendor_uuid.in_(vendor_uuids)) ) vendors = query.all() vendors_data = [v._asdict() for v in vendors] return vendors_data except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) raise Exception(e.args) def get_vendor_label_info(vendor_id): """Get all vendor information for a vendor given vendor_id. Args: vendor_id (int): unique identifier for the vendor. Returns: response.Response: containing a dict of vendor or error. """ try: with mysql.session_scope(read_only=True) as session: params = {'vendor_id': vendor_id} vendor = session.execute(VENDOR_LABEL_INFO_SQL, params).fetchone() if not vendor: return response.create_not_found_response( message='Vendor : {} not found.'.format(vendor_id) ) vendor_data = dict(vendor) if vendor_data.get('oa_admin_first') and vendor_data.get('oa_admin_last'): vendor_data['assigned_to'] = ( vendor_data.pop('oa_admin_first') + ' ' + vendor_data.pop('oa_admin_last') ) else: vendor_data['assigned_to'] = None if vendor_data.get('wel_email_send_date'): vendor_data['wel_email_send_date'] = vendor_data.get( 'wel_email_send_date' ).strftime('%Y-%m-%d') return response.Response(vendor_data) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def get_vendor_closers(vendor_uuids: list[str]) -> list[Dict[str, Union[str, List[int]]]]: """Get list of closers associated with vendor. Args: vendor_uuids (list[str]): unique identifier for the vendor. Returns: list[Dict[str, Union[str, List[int]]]]: a list of dict containing uuid and its closers. """ try: with mysql.session_scope(read_only=True) as session: closers = ( session.query( Vendor.vendor_uuid, func.group_concat(VendorClosers.orchadmin_user_id).label('closers'), ) .join(Vendor, Vendor.vendor_id == VendorClosers.vendor_id) .filter(Vendor.vendor_uuid.in_(vendor_uuids)) .group_by(Vendor.vendor_uuid) .all() ) vendor_closers = [ { 'uuid': str(closer.vendor_uuid), 'closers': list(map(int, closer.closers.split(','))), } for closer in closers ] return vendor_closers except SQLAlchemyError as e: g.log.exception(e.args) raise Exception(e.args) def get_vendors_first_statement_period( vendor_uuids: list[str], ) -> list[dict[str, str]]: """Retrieve first statement period data for the given vendor UUIDs. Args: vendor_uuids (list[str]): List of vendor UUIDs. Returns: list[dict[str, str]]: Each dict includes 'uuid' and 'first_statement_period'. """ try: with mysql.session_scope(read_only=True) as session: rows = ( session.query(Vendor.vendor_uuid, Vendor.first_statement_period) .filter(Vendor.vendor_uuid.in_(vendor_uuids)) .all() ) result = [ { 'uuid': row.vendor_uuid, # graphql keys are strings 'first_statement_period': str(row.first_statement_period), } for row in rows ] return result except SQLAlchemyError as e: g.log.exception(e.args) raise Exception(e.args) def get_recurring_payment_threshold(vendor_id): """Get currency symbol and currency code for recurring payment threshold field. Args: vendor_id (int): unique identifier for the vendor. Returns: response.Response: a dict containing currency_symbol and currency_code. """ try: with mysql.session_scope(read_only=True) as session: params = {'vendor_id': vendor_id} result = session.execute(VENDOR_CONTRACT_CURRENCY_SQL, params).fetchone() if not result: return response.create_not_found_response( message='Vendor contract currency for {} not found.'.format(vendor_id) ) return response.Response(dict(result)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def get_product_manager_id(vendor_id): """Get the product_manager_id for vendor. Args: vendor_id (int): unique identifier for the vendor. Returns: response.Response: product_manager """ try: with mysql.session_scope(read_only=True) as session: params = {'vendor_id': vendor_id} result = session.execute(PRODUCT_MANAGER_SQL, params).fetchone() if not result: return response.create_not_found_response( message='Product manager for vendor {} not found.'.format(vendor_id) ) return response.Response(dict(result)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def _get_company_brand_id(company_brand_uuid): """Get the company_brand_id for vendor. Args: company_brand_uuid (str): company_brand uuid. Returns: response.Response: company_brand_id """ try: with mysql.session_scope(read_only=True) as session: params = {'company_brand_uuid': company_brand_uuid} result = session.execute(COMPANY_BRAND_ID_SQL, params).fetchone() if not result: return response.create_not_found_response( message='Company_brand_id for company_brand_uuid {} not found.'.format( company_brand_uuid ) ) return response.Response(dict(result)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def get_company_brand_id_by_name(company_brand_name): """Get the company_brand_id for vendor. Args: company_brand_name (str): company_brand name. Returns: response.Response: company_brand_id """ try: with mysql.session_scope(read_only=True) as session: params = {'company_brand_name': company_brand_name} result = session.execute(COMPANY_BRAND_ID_BY_NAME_SQL, params).fetchone() if not result: return response.create_not_found_response( message='Company_brand_id for company_brand_name {} not found.'.format( company_brand_name ) ) return response.Response(dict(result)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def get_soundscan_code(vendor_id, country_id): """Get soundscan_code. Args: vendor_id (int): unique identifier for the vendor. country_id (int): country id. Returns: response.Response """ try: with mysql.session_scope(read_only=True) as session: params = {'vendor_id': vendor_id, 'country_id': country_id} query = session.query(SoundScanCode).filter( SoundScanCode.vendor_id == vendor_id, SoundScanCode.country_id == country_id, SoundScanCode.subaccount_id == None, ) result = session.execute(query, params).fetchone() if not result: return response.create_not_found_response( message='Soundscan code for vendor {} not found.'.format(vendor_id) ) return response.Response(dict(result)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def update_product_manager_id(vendor_id, pm_id): """Update product_manager_id. Args: vendor_id (int): unique identifier for the vendor. pm_id (int): unique identifier for the product_manager_id. Returns: response.Response: """ try: with mysql.session_scope() as session: params = {'vendor_id': vendor_id, 'pm_id': pm_id} session.execute( text( 'INSERT INTO product_manager_mapping_vendor(\ product_manager_id, vendor_id, updated_at)\ VALUES (:pm_id, :vendor_id, current_timestamp)\ ON DUPLICATE KEY UPDATE product_manager_id = :pm_id,\ updated_at = current_timestamp;' ), params, ) return response.Response( message=f'product manager {pm_id} assigned to vendor {vendor_id}.' ) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def delete_product_manager_id(vendor_id): """Unassign product manager from this vendor. Args: vendor_id(int): unique identifier for the vendor Returns: response.Response: success message. """ try: with mysql.session_scope() as session: params = {'vendor_id': vendor_id} result = session.execute(text(DELETE_PRODUCT_MANAGER_MAPPING_SQL), params) if not result: return response.create_not_found_response( message=f'Unable to remove product manager for vendor {vendor_id}.' ) return response.Response(f'product manager removed from vendor {vendor_id}') except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def update_soundscan_code(vendor_id, sc_code, country_id): """Update soundscan codes. Args: vendor_id (int): unique identifier for the vendor. sc_code (str): unique identifier for soundscan code. country_id (int): identifier of country (country_id 1 == USA, country_id 2 == Canada) Returns: response.Response: """ with mysql.session_scope() as session: params = {'vendor_id': vendor_id, 'sc_code': sc_code, 'country_id': country_id} result = session.execute(UPDATE_SOUNDSCAN_CODE, params) return response.Response(dict(result)) def update_company_brand_art_relations(vendor_id, company_brand_uuid): """Update company_brand_id in vendor table in AR. Args: vendor_id (int): unique identifier for the vendor. company_brand_uuid (str): unique identifier as uuid for company_brand. Returns: response.Response: """ company_brand_id = _get_company_brand_id(company_brand_uuid).message['company_brand_id'] with mysql.session_scope() as session: params = {'vendor_id': vendor_id, 'company_brand_id': company_brand_id} result = session.execute(UPDATE_COMPANY_BRAND_ID, params) return response.Response(dict(result)) def map_soundscan_code(vendor_id, sc_code, country_id): """Insert map row to assign soundscan code to vendor.""" if sc_code == 'null': sc_code = '' with mysql.session_scope() as session: params = { 'vendor_id': vendor_id, 'sc_code': sc_code, 'country_id': country_id, 'subaccount_id': None, } insert = session.execute( text( 'INSERT INTO soundscan_codes( \ soundscan_code, vendor_id, country_id, subaccount_id) \ VALUES (:sc_code, :vendor_id, :country_id, :subaccount_id)' ), params, ) return response.Response(dict(insert)) def sc_map_exists(vendor_id, country_id): """Checks if a soundscan code is assigned to this vendor_id. Args: vendor_id(int): unique identifier for the vendor country_id(int): country identifier Returns: int (0, 1): count of rows existing in table. if 0, a new map/row will need to be created. If 1, info will be updated. """ try: with mysql.session_scope(read_only=True) as session: params = {'vendor_id': vendor_id, 'country_id': country_id} result = session.execute( text( 'SELECT count(*) FROM soundscan_codes sc \ WHERE sc.vendor_id=:vendor_id AND sc.country_id=:country_id \ AND sc.subaccount_id is NULL' ), params, ).fetchone() return result[0] except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def create_vendor_with_optional_service_tier(details: dict, service_tier_uuid: str | None = None): """Create a new vendor. Args: details (dict): details for vendor. Returns: response.Response: containing dict of vendor. """ try: with mysql.session_scope() as session: vendor = Vendor(**details) session.add(vendor) if service_tier_uuid: tier = service_tier.ServiceTier.get_by_uuid(uuid=service_tier_uuid, session=session) if tier: vendor.service_tier = tier else: message = f'Invalid service tier uuid {service_tier_uuid} given' g.log.error(message) return response.create_error_response(error.ERROR_CODE_INVALID_INPUT, message) session.commit() return response.Response(vendor.to_dict()) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() return response.create_fatal_response(e.args) def update_vendor(vendor_id, details): """Update an existing vendor. Args: vendor_id (int): Vendor id. details (dict): details for vendor. Returns: response.Response: containing success message. """ try: with mysql.session_scope() as session: result = session.query(Vendor).filter(Vendor.vendor_id == vendor_id).update(details) session.commit() if result < 1: return response.create_not_found_response( f'Vendor update unsuccessful: No vendor found for vendor_id {vendor_id}.' ) updated_vendor = session.query(Vendor).filter(Vendor.vendor_id == vendor_id).first() return response.Response( {'vendor_id': updated_vendor.vendor_id, 'vendor_uuid': updated_vendor.vendor_uuid} ) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() return response.create_fatal_response(e.args) def upsert_vendor_closers( vendor_uuid: str, orchadmin_user_ids: list[int], last_modified_by: Union[int, None] = None ) -> UpdatedVendorClosers: """ Upsert vendor closers: Ensures only the given orchadmin_user_ids exist for the vendor. Args: vendor_uuid (str): UUID of the vendor. orchadmin_user_ids (List[int]): List of orchadmin_user_id values to keep. last_modified_by (Union[int, None]): orchadmin_user_id who requested the update. Null if kicked off by step function. Returns: UpdatedVendorClosers with vendor_id, vendor_uuid, and the updated closers list. Raises: VendorUpdateException: 404 when no vendor matches, 500 on DB error. """ try: with mysql.session_scope() as session: vendor = ( session.query(Vendor.vendor_id) .filter(Vendor.vendor_uuid == vendor_uuid) .one_or_none() ) if not vendor: raise VendorUpdateException( code=error.ERROR_CODE_NOT_FOUND, message=f'Vendor with UUID {vendor_uuid} not found', status=404, ) vendor_id = vendor.vendor_id # Delete all existing closers for the vendor session.query(VendorClosers).filter(VendorClosers.vendor_id == vendor_id).delete( synchronize_session=False ) # Insert new closers if list is not empty if orchadmin_user_ids: closer_records = [ { 'vendor_id': vendor_id, 'orchadmin_user_id': user_id, 'last_modified_by': last_modified_by, } for user_id in orchadmin_user_ids ] session.bulk_insert_mappings(VendorClosers, closer_records) return UpdatedVendorClosers( vendor_id=vendor_id, vendor_uuid=vendor_uuid, closers=orchadmin_user_ids, ) except SQLAlchemyError as e: g.log.exception(e.args) raise VendorUpdateException( code=error.ERROR_CODE_INVALID_REQUEST, message=str(e.args), status=500, ) def update_vendor_first_statement_period( vendor_uuid: str, first_statement_period: int, ) -> dict[str, str]: """ Updates the first_statement_period identified by its UUID. Args: vendor_uuid (str): UUID of the vendor. first_statement_period (int): value to set for vendor's first statement period. Returns: Response: Contains vendor_id, vendor_uuid, and updated first statement period or error message. """ try: with mysql.session_scope() as session: result = ( session.query(Vendor) .filter(Vendor.vendor_uuid == vendor_uuid) .update({'first_statement_period': first_statement_period}) ) session.commit() if result < 1: raise VendorUpdateException( code=error.ERROR_CODE_INVALID_REQUEST, message=f'Vendor update unsuccessful: ' f'No vendor found for vendor_uuid {vendor_uuid}.', status=400, ) updated_vendor = session.query(Vendor).filter(Vendor.vendor_uuid == vendor_uuid).first() return { 'vendor_id': updated_vendor.vendor_id, 'vendor_uuid': updated_vendor.vendor_uuid, 'first_statement_period': str(updated_vendor.first_statement_period), } except Exception as e: raise e def update_vendor_relationship_notes(vendor_uuid: str, notes: str) -> dict[str, str]: """Update the relationship_notes field for a vendor by vendor_uuid. Args: vendor_uuid (str): Unique identifier for the vendor. notes (str): The relationship notes to update for the vendor. Returns: dict[str, str]: Dictionary containing vendor_id, vendor_uuid, and the updated relationship_notes. """ try: with mysql.session_scope() as session: result = ( session.query(Vendor) .filter(Vendor.vendor_uuid == vendor_uuid) .update({'relationship_notes': notes}) ) session.commit() if result < 1: raise VendorUpdateException( code=error.ERROR_CODE_INVALID_REQUEST, message=f'Vendor update unsuccessful: ' f'No vendor found for vendor_uuid {vendor_uuid}.', status=400, ) updated_vendor = session.query(Vendor).filter(Vendor.vendor_uuid == vendor_uuid).first() return { 'vendor_id': updated_vendor.vendor_id, 'vendor_uuid': updated_vendor.vendor_uuid, 'relationship_notes': updated_vendor.relationship_notes, } except Exception as e: raise e def get_vendor_document_by_id(vendor_id, with_tenant_uuids=False): """Get the document as defined in cloudsearch corpus by vendor_id id.""" try: query = VENDOR_DOCUMENT_WITH_TENANT_UUIDS_SQL if with_tenant_uuids else VENDOR_DOCUMENT_SQL with mysql.session_scope(read_only=True) as session: params = {'vendor_id': vendor_id} vendor = session.execute(query, params).fetchone() if not vendor: return response.create_not_found_response( message='Vendor : {} not found.'.format(vendor_id) ) return response.Response(dict(vendor)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def get_assigned_to_by_id(vendor_id): """Get the assigned-to field by vendor_id.""" try: with mysql.session_scope(read_only=True) as session: vendor = session.execute(VENDOR_ASSIGNED_TO_SQL, {'vendor_id': vendor_id}).fetchone() if not vendor: return response.create_not_found_response( message=f'Vendor : {vendor_id} not found.' ) return response.Response(dict(vendor)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def get_assigned_reviewer_by_id(vendor_id): """Get the assigned-reviewer field by vendor_id.""" try: with mysql.session_scope(read_only=True) as session: vendor = session.execute( VENDOR_ASSIGNED_REVIEWER_SQL, {'vendor_id': vendor_id} ).fetchone() if not vendor: return response.create_not_found_response( message=f'Vendor : {vendor_id} not found.' ) return response.Response(dict(vendor)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def get_secondary_internal_contact_by_id(vendor_id): """Get the secondary internal contact by vendor id.""" try: with mysql.session_scope(read_only=True) as session: vendor = session.execute( VENDOR_SECONDARY_INTERNAL_CONTACT_SQL, {'vendor_id': vendor_id} ).fetchone() if not vendor: return response.create_not_found_response( message=f'Vendor {vendor_id}: Secondary Internal Contact not found.' ) return response.Response(dict(vendor)) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def get_all_vendor_currency_codes(): """Get the ISO currency codes for all vendors. Returns: response.Response: containing dict of vendor currency codes. """ try: with mysql.session_scope(read_only=True) as session: rows = session.execute(VENDOR_CURRENCY_CODES_SQL).fetchall() items = [] for row in rows: items.append({'vendor_id': row[0], 'currency_code': row[1]}) return response.Response({'items': items}) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) return response.create_fatal_response(e.args) def update_is_distributor_in_vendor(vendor_id): """Update is_distributor field from N to Y. Args: vendor_id (int): Vendor id. Returns: response.Response: containing dict of vendor. """ with mysql.session_scope() as session: row = session.query(Vendor).filter(Vendor.vendor_id == vendor_id).one_or_none() if not row: return response.create_not_found_response(f'No vendor found for vendor_id {vendor_id}.') if row.is_distributor == 'N': row.is_distributor = 'Y' row.label_identifier = 'D3' session.commit() return response.Response(constants.VENDOR_UPDATED) return response.Response(constants.VENDOR_ALREADY_DISTRIBUTOR) def get_vendors_by_external_identifier(external_identifier_1, owner): """Get all vendors by external_identifier_1 and owner. Args: external_identifier_1 (str): external_identifier_1. owner (str): Vendor owner. Returns: response.Response: containing dict of vendor's ids. """ with mysql.session_scope(read_only=True) as session: result = ( session.query(Vendor.vendor_id, Vendor.vendor_uuid) .filter( Vendor.external_identifier_1 == external_identifier_1, Vendor.owner == owner, ) .all() ) return response.Response({'items': [row._asdict() for row in result]}) def lookup_vendors_by_uuids( vendor_uuids: list[str], fetch_flags: list[str], ) -> response.Response: """Lookup vendors by uuids. Args: vendor_uuids: list of vendor uuids fetch_flags: list of additional identifier attributes to fetch Return: response.Response.message: list(obj). obj shape: { vendor_id: number uuid: str company_brand_uuid: str (when fetch flags are requested) parent_company_uuid: str (when fetch flags are requested && FF is enabled) } """ if not vendor_uuids: return response.Response([]) select_entities = [Vendor.vendor_id, Vendor.vendor_uuid.label('uuid')] join_entities = [] if constants.FETCH_TENANT_HIERARCHY in fetch_flags: select_entities.append(CompanyBrand.uuid.label('company_brand_uuid')) join_entities.append(Vendor.company_brand) select_entities.append(ParentCompany.uuid.label('parent_company_uuid')) join_entities.append(CompanyBrand.parent_company) if constants.FETCH_IS_DISTRIBUTOR in fetch_flags: select_entities.append( case( [ (Vendor.is_distributor == 'Y', True), ], else_=False, ).label('is_distributor'), ) with mysql.session_scope(read_only=True) as session: query = session.query(*select_entities) if join_entities: query = query.join(*join_entities) result = query.filter(Vendor.vendor_uuid.in_(vendor_uuids)) if not result: return response.create_not_found_response(message='vendor_uuids not found.') return response.Response([row._asdict() for row in result]) def lookup_vendors_by_vendor_ids( vendor_ids: list[str], fetch_flags: list[str], ) -> response.Response: """Lookup vendors by uuids. Args: vendor_ids: list of vendor uuids fetch_flags: list of additional identifier attributes to fetch Return: response.Response.message: list(obj). obj shape: { vendor_id: number uuid: str company_brand_uuid: str (when fetch flags are requested) parent_company_uuid: str (when fetch flags are requested && FF is enabled) } """ if not vendor_ids: return response.Response([]) select_entities = [Vendor.vendor_id, Vendor.vendor_uuid.label('uuid')] join_entities = [] if constants.FETCH_TENANT_HIERARCHY in fetch_flags: select_entities.append(CompanyBrand.uuid.label('company_brand_uuid')) join_entities.append(Vendor.company_brand) select_entities.append(ParentCompany.uuid.label('parent_company_uuid')) join_entities.append(CompanyBrand.parent_company) if constants.FETCH_IS_DISTRIBUTOR in fetch_flags: select_entities.append( case( [ (Vendor.is_distributor == 'Y', True), ], else_=False, ).label('is_distributor'), ) with mysql.session_scope(read_only=True) as session: query = session.query(*select_entities) if join_entities: query = query.join(*join_entities) result = query.filter(Vendor.vendor_id.in_(vendor_ids)) if not result: return response.create_not_found_response(message='vendor_ids not found.') return response.Response([row._asdict() for row in result]) def lookup_vendor_and_company_brand_by_uuid( uuid: str, session: Session ) -> Tuple[VendorType, CompanyBrandType] | None: """Lookup vendor by uuid. Args: uuid: vendor uuid session: sqlalchemy session Return: Tuple[VendorType, CompanyBrandType]: vendor and company brand. """ select_entities = [ Vendor.vendor_id, Vendor.migrated_to_abacus, Vendor.owner, Vendor.vendor_uuid.label('uuid'), CompanyBrand.id.label('company_brand_id'), CompanyBrand.name.label('company_brand_name'), ] join_entities = [Vendor.company_brand] query = session.query(*select_entities).join(*join_entities).filter(Vendor.vendor_uuid == uuid) result = query.first() if not result: return None result_dict = result._asdict() return ( VendorType( vendor_id=result_dict['vendor_id'], uuid=result_dict['uuid'], migrated_to_abacus=result_dict['migrated_to_abacus'], owner=result_dict['owner'], ), CompanyBrandType( id=result_dict['company_brand_id'], name=result_dict['company_brand_name'], ), ) def lookup_vendors_and_company_brands_by_uuids( vendor_uuids: list[str], session: Session ) -> dict[str, Tuple[VendorType, CompanyBrandType]]: """Bulk lookup vendors and company brands by UUIDs. Args: vendor_uuids: list of vendor UUIDs session: sqlalchemy session Return: dict[str, Tuple[VendorType, CompanyBrandType]]: mapping of UUIDs to vendor and company brand tuples. """ if not vendor_uuids: return {} select_entities = [ Vendor.vendor_id, Vendor.migrated_to_abacus, Vendor.owner, Vendor.vendor_uuid.label('uuid'), CompanyBrand.id.label('company_brand_id'), CompanyBrand.name.label('company_brand_name'), ] join_entities = [Vendor.company_brand] query = ( session.query(*select_entities) .join(*join_entities) .filter(Vendor.vendor_uuid.in_(vendor_uuids)) ) results = query.all() result = { row.uuid: ( VendorType( vendor_id=row.vendor_id, uuid=row.uuid, migrated_to_abacus=row.migrated_to_abacus, owner=row.owner, ), CompanyBrandType( id=row.company_brand_id, name=row.company_brand_name, ), ) for row in results } return result @tracer.wrap() def delete_vendor_by_uuid(vendor_uuid: str, last_modified_by: int) -> DeletedVendor: """Soft-delete a vendor by setting its status to 'deletion'. Returns: DeletedVendor: vendor_id, vendor_uuid, and status. status is 'deletion' if this call performed the soft-delete; None if the vendor was already deleted or was not found (in which case vendor_id is also None). Raises: SQLAlchemyError: on database error. """ try: with mysql.session_scope() as session: existing = session.query(Vendor).filter(Vendor.vendor_uuid == vendor_uuid).first() vendor_status = None if existing and existing.status != 'deletion': vendor_status = 'deletion' existing.status = vendor_status existing.last_modified_by = last_modified_by session.commit() return DeletedVendor( vendor_id=existing.vendor_id if existing else None, vendor_uuid=vendor_uuid, status=vendor_status, last_modified_by=last_modified_by, ) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() g.log.exception(e.args) raise @tracer.wrap() def update_vendor_by_uuid(vendor_uuid: str, details: dict[str, Any]) -> UpdatedVendor: """Update an existing vendor, keying off vendor_uuid. Returns: UpdatedVendor with vendor_id and vendor_uuid on success. Raises: VendorUpdateException: 404 when no vendor matches, 500 on DB error. """ try: with mysql.session_scope() as session: result = session.query(Vendor).filter(Vendor.vendor_uuid == vendor_uuid).update(details) session.commit() if result < 1: raise VendorUpdateException( code=error.ERROR_CODE_INVALID_INPUT, message=( f'Vendor update unsuccessful: No vendor found for ' f'vendor_uuid {vendor_uuid}.' ), status=404, ) updated_vendor = session.query(Vendor).filter(Vendor.vendor_uuid == vendor_uuid).first() return UpdatedVendor( vendor_id=updated_vendor.vendor_id, vendor_uuid=updated_vendor.vendor_uuid, ) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() raise VendorUpdateException( code=error.ERROR_CODE_INVALID_REQUEST, message=str(e.args), status=500, ) @tracer.wrap() def update_vendor_service_tier_by_uuid( vendor_uuid: str, service_tier_uuid: str, last_modified_by: int, ) -> UpdatedVendor: """Replace the vendor_service_tier row for a vendor and stamp last_modified_by. Vendor existence is enforced upstream by the PP authorization check; if the vendor is missing, the SQLAlchemy load below raises and is wrapped as 500. The schema's OneOf validator gates service_tier_uuid before this is called, but tier.get_by_uuid still defensively returns None for unknown uuids — we surface that as a 400. """ try: with mysql.session_scope() as session: vendor = session.query(Vendor).filter(Vendor.vendor_uuid == vendor_uuid).one() tier = service_tier.ServiceTier.get_by_uuid(uuid=service_tier_uuid, session=session) if tier is None: raise VendorUpdateException( code=error.ERROR_CODE_INVALID_INPUT, message=f'Invalid service tier uuid {service_tier_uuid} given.', status=400, ) vendor.service_tier = tier vendor.last_modified_by = last_modified_by session.commit() return UpdatedVendor(vendor_id=vendor.vendor_id, vendor_uuid=vendor.vendor_uuid) except SQLAlchemyError as e: if sentry_client: sentry_client.capture_exception() raise VendorUpdateException( code=error.ERROR_CODE_INVALID_REQUEST, message=str(e.args), status=500, ) def get_relationship_notes(vendor_uuids: list[str]) -> list[dict]: try: with mysql.session_scope(read_only=True) as session: vendors = ( session.query(Vendor.vendor_id, Vendor.vendor_uuid, Vendor.relationship_notes) .filter(Vendor.vendor_uuid.in_(vendor_uuids)) .all() ) return [ { 'vendor_id': vendor.vendor_id, 'vendor_uuid': vendor.vendor_uuid, 'relationship_notes': vendor.relationship_notes, } for vendor in vendors ] except SQLAlchemyError as e: g.log.exception(e) raise Exception(f'Database error fetching relationship_notes: {str(e)}')