""" Contact Model ============= Model function for getting contact information. """ from ows_accounting import response from ows_accounting.utils import mysql def _fetchone(sql, **args): """Query database and fetch one result. Args: sql (str): sql query. args (dict): dictionary of parameters. Example: dict(vendor_id=vendor_id) Returns: tuple: one row of tuple of results. None if no result found. """ with mysql.db_session() as session: result = session.execute(sql, args) return result.fetchone() def get_email_by_vend_contact_id(vend_contact_id): """Get transaction types that have sales. Args: vend_contact_id (int): primary key of vend_contact table. Returns: Response object: Response object with message equals email for vend_contact_id. 404 response if not found. """ sql = """ SELECT c.contact_email FROM vend_contact vc INNER JOIN contact c ON c.contact_id = vc.contact_id WHERE vc.id = :vend_contact_id """ result = _fetchone(sql, vend_contact_id=vend_contact_id) if result: return response.Response(result[0]) return response.create_not_found_response() def get_email_by_subaccount_id(subaccount_id): """Get email for subaccount_id. Args: subaccount_id (int): subaccount_id. Returns: Response object: Response object with message equals email for subaccount_id. 404 response if not found. """ sql = """ SELECT c.contact_email FROM vend_contact vc INNER JOIN contact c ON c.contact_id = vc.contact_id WHERE vc.subaccount_id = :subaccount_id """ result = _fetchone(sql, subaccount_id=subaccount_id) if result: return response.Response(result[0]) return response.create_not_found_response() def get_email_by_vendor_id(vendor_id): """Get email for vendor_id. Args: vendor_id (int): vendor_id. Returns: Response object: Response object with message equals email for vendor_id. 404 response if not found. """ sql = """ SELECT c.contact_email FROM vend_contact vc INNER JOIN contact c ON c.contact_id = vc.contact_id WHERE vc.master = 'Y' AND vc.vendor_id = :vendor_id """ result = _fetchone(sql, vendor_id=vendor_id) if result: return response.Response(result[0]) return response.create_not_found_response()