File size: 3,595 Bytes
4b9d59b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """
Apollo Integration Adapter
Provides a unified interface for Apollo.io services within the IntegrationFactory.
"""
import logging
import os
import httpx
from typing import Dict, Any, List, Optional
from datetime import datetime, timezone
logger = logging.getLogger(__name__)
class ApolloAdapter:
"""
Adapter for Apollo.io API integration.
"""
def __init__(self, db=None, workspace_id: str = None):
self.db = db
self.workspace_id = workspace_id
self.service_name = "apollo"
self.base_url = "https://api.apollo.io/v1"
# In a real scenario, we would fetch the API key from the database
# using workspace_id and service_name from UserConnection table.
# For now, we'll assume it's provided or handled by the factory/caller context.
self._api_key: Optional[str] = os.getenv("APOLLO_API_KEY")
async def test_connection(self) -> bool:
"""Test the Apollo API connection"""
if not self._api_key:
return False
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/auth/health",
params={"api_key": self._api_key}
)
return response.status_code == 200
except Exception as e:
logger.error(f"Apollo connection test failed: {e}")
return False
async def get_data(self, data_type: str, query: str = None, **kwargs) -> Dict[str, Any]:
"""
Generic method to fetch data from Apollo.
Supported data_types:
- people: Search for people
- search: Search for people (alias)
- enrichment: Enrich a person by email
"""
if not self._api_key:
raise ValueError("Apollo API key not configured")
async with httpx.AsyncClient() as client:
if data_type in ["people", "search"]:
# Search people
search_query = query or kwargs.get("q_description", "")
response = await client.post(
f"{self.base_url}/mixed_people/search",
params={"api_key": self._api_key},
json={"q_description": search_query}
)
response.raise_for_status()
result = response.json()
return {"ok": True, "data": result.get("people", [])}
elif data_type == "enrichment":
# Enrich person
email = query or kwargs.get("email")
if not email:
raise ValueError("Email is required for enrichment")
response = await client.get(
f"{self.base_url}/people/match",
params={"api_key": self._api_key, "email": email}
)
response.raise_for_status()
result = response.json()
return {"ok": True, "data": result.get("person")}
else:
raise ValueError(f"Unsupported data type for Apollo: {data_type}")
async def search_people(self, query: str) -> List[Dict[str, Any]]:
"""Search people in Apollo"""
result = await self.get_data("people", query=query)
return result.get("data", [])
async def enrich_person(self, email: str) -> Dict[str, Any]:
"""Enrich person by email"""
result = await self.get_data("enrichment", query=email)
return result.get("data", {})
|