fetch_example.py

from prefect.filesystems import S3
s3_block = S3.load('chatbot20231012')
from prefect.blocks.system import Secret
from sellpath_test import ClientMgr
client = ClientMgr('5d143acf-2812-4fed-b4d4-6b48bd8eca4b')

from prefect import task, flow
from sellpath_test import ClientMgr
from sellpath_test.http_client import SellPathHttpClient


@task(name="Connect to Apollo IO")
def connect_to_apollo(env_tag) -> SellPathHttpClient:
    apollo_client = client.get_client("apollo", env_tag)
    return apollo_client

@task(name="Search Contact in Apollo IO")
def search_contact(apollo_client, email):
    path = "/contacts/search"
    params = {"q_keyword": email}
    response = apollo_client.get(path, params=params)
    return response

@task(name="Extract Contact Details")
def extract_contact_details(response):
    if response and 'contacts' in response:
        contacts = response['contacts']
        if contacts:
            contact_details = contacts[0]  # Assuming the first match is the desired contact
            return {
                "name": contact_details.get("name"),
                "email": contact_details.get("email"),
                "phone": contact_details.get("phone"),
                "title": contact_details.get("title"),
                "last_activity_date": contact_details.get("last_activity_date")
            }
    return {}

@flow(name="main", persist_result=True, result_storage=s3_block)
def main(env_tag="production", email="[email protected]"):
    output = {}
    try:
        apollo_client = connect_to_apollo(env_tag)
        contact_response = search_contact(apollo_client, email)
        contact_info = extract_contact_details(contact_response)
        output = {
            "contact_details": contact_info
        }
        return output

    except Exception as e:
        output["status"] = "failed"
        output["error"] = str(e)
        return output