Spaces:
Running
Running
| """ | |
| 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", {}) | |