def process_contact_and_account_id()

in workmail-salesforce-python/src/sf_utils.py [0:0]

• 24 lines of code
• 11 McCabe index (conditional complexity)

def process_contact_and_account_id(sf_client, parsed_email, event, from_address):
    # Set the following fields depending on flow direction of email
    contact_address = None
    first_name = 'None'
    last_name = 'None'

    if event['flowDirection'] == 'INBOUND':
        contact_address = from_address
        if parsed_email['From'] is not None:
            first_name, last_name = email_utils.extract_username(parsed_email['From'])
    else:
        # Here we simplify and assume that contact first envelop recipient and `To` header of the email are same.
        contact_address = event['envelope']['recipients'][0]['address']
        if parsed_email['To'] is not None:
            first_name, last_name = email_utils.extract_username(parsed_email['To'])

    # Fetch or create the contact and account in Salesforce
    contact_id = run_sf_query(sf_client, f"SELECT Id FROM Contact WHERE Email='{contact_address}'", 'Id')
    account_id = None

    if contact_id is not None:
        account_id = run_sf_query(sf_client, f"SELECT AccountId FROM Contact WHERE Id='{contact_id}'", 'AccountId')

    if account_id is None:
        domain = contact_address.lower().split('@')[1]
        account_id = sf_client.Account.create({'Name': domain})['id']
        logger.info(f"Created a new account for {domain} with AccountId: {account_id}")

    if contact_id is None:
        contact_id = sf_client.Contact.create({'LastName': last_name, 'FirstName': first_name, 'Email': contact_address, 'AccountId': account_id})['id']
        logger.info(f"Created a new contact for {contact_address} with ContactId: {contact_id}")

    return contact_id, account_id