Maxim Snesarev
33d6bb7ebd
- Updated database configuration to switch from SQLite to PostgreSQL, including changes to alembic.ini, Docker Compose, and environment settings. - Refactored application code to utilize PostgreSQL database adapters, ensuring compatibility with the new database structure. - Enhanced API routes and data handling to support the new database, including adjustments in data models and query logic. - Introduced new job processing mechanisms for full synchronization of AMO CRM entities, leveraging FastStream for background tasks. - Improved logging and error handling across the application to facilitate better monitoring and debugging. - Removed obsolete SQLite adapter files and migrations, streamlining the project structure for PostgreSQL integration.
192 lines
7.7 KiB
Python
192 lines
7.7 KiB
Python
"""
|
|
AMO CRM API Client
|
|
Simple client for fetching data from AMO CRM API
|
|
"""
|
|
import httpx
|
|
import asyncio
|
|
from typing import Dict, Any, List, Optional
|
|
from utils.config import settings
|
|
|
|
|
|
class AmoCRMClient:
|
|
def __init__(self, domain: Optional[str] = None, access_token: Optional[str] = None):
|
|
self.domain = domain or settings.AMO_CRM_DOMAIN
|
|
self.access_token = access_token or settings.AMO_CRM_ACCESS_TOKEN
|
|
self.base_url = f"https://{self.domain}/api/v4"
|
|
|
|
self.headers = {
|
|
"Authorization": f"Bearer {self.access_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def _make_request(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Make HTTP request to AMO CRM API"""
|
|
async with httpx.AsyncClient() as client:
|
|
url = f"{self.base_url}/{endpoint}"
|
|
response = await client.get(url, headers=self.headers, params=params or {})
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_users(self, limit: int = 250) -> Dict[str, Any]:
|
|
"""Get users (пользователи)"""
|
|
return await self._make_request("users", {"limit": limit})
|
|
|
|
async def get_pipelines(self) -> Dict[str, Any]:
|
|
"""Get pipelines (воронки)"""
|
|
return await self._make_request("leads/pipelines")
|
|
|
|
async def get_companies(self, limit: int = 250, page: int = 1) -> Dict[str, Any]:
|
|
"""Get companies (компании)"""
|
|
params = {"limit": limit, "page": page, "with": "custom_fields_values"}
|
|
return await self._make_request("companies", params)
|
|
|
|
async def get_contacts(self, limit: int = 250, page: int = 1) -> Dict[str, Any]:
|
|
"""Get contacts (контакты)"""
|
|
params = {"limit": limit, "page": page, "with": "custom_fields_values"}
|
|
return await self._make_request("contacts", params)
|
|
|
|
async def get_deals(self, limit: int = 250, page: int = 1) -> Dict[str, Any]:
|
|
"""Get deals/leads (сделки)"""
|
|
params = {"limit": limit, "page": page, "with": "contacts,companies,custom_fields_values"}
|
|
return await self._make_request("leads", params)
|
|
|
|
async def get_events(self, limit: int = 250, page: int = 1, event_types: Optional[List[str]] = None) -> Dict[str, Any]:
|
|
"""Get events (события)"""
|
|
params: Dict[str, Any] = {"limit": limit, "page": page}
|
|
|
|
# Filter by event types
|
|
if event_types:
|
|
params["filter[type]"] = ",".join(event_types)
|
|
else:
|
|
# Default to our supported event types
|
|
params["filter[type]"] = ",".join(["incoming_call", "outgoing_call", "lead_status_changed"])
|
|
|
|
return await self._make_request("events", params)
|
|
|
|
async def get_custom_fields(self, entity_type: str = "leads") -> Dict[str, Any]:
|
|
"""Get custom fields metadata for entity type"""
|
|
return await self._make_request(f"{entity_type}/custom_fields")
|
|
|
|
async def fetch_entity_data(
|
|
self,
|
|
entity_type: str,
|
|
limit: Optional[int] = None,
|
|
page: Optional[int] = None,
|
|
updated_at: Optional[int] = None,
|
|
**kwargs
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Generic method to fetch any entity type from AMO CRM.
|
|
|
|
Args:
|
|
entity_type: Type of entity (deals, contacts, companies, users, pipelines, events)
|
|
limit: Maximum number of records to fetch
|
|
page: Page number for pagination
|
|
updated_at: Unix timestamp for incremental updates
|
|
**kwargs: Additional query parameters
|
|
|
|
Returns:
|
|
List of entity records from AMO CRM
|
|
"""
|
|
# Map entity types to their API endpoints and response keys
|
|
entity_config = {
|
|
"deals": {"endpoint": "leads", "key": "leads", "with": "contacts,companies,custom_fields_values"},
|
|
"contacts": {"endpoint": "contacts", "key": "contacts", "with": "custom_fields_values"},
|
|
"companies": {"endpoint": "companies", "key": "companies", "with": "custom_fields_values"},
|
|
"users": {"endpoint": "users", "key": "users", "with": None},
|
|
"pipelines": {"endpoint": "leads/pipelines", "key": "pipelines", "with": None},
|
|
"events": {"endpoint": "events", "key": "events", "with": None},
|
|
}
|
|
|
|
if entity_type not in entity_config:
|
|
raise ValueError(f"Unsupported entity type: {entity_type}. Must be one of {list(entity_config.keys())}")
|
|
|
|
config = entity_config[entity_type]
|
|
params: Dict[str, Any] = {}
|
|
|
|
# Add pagination parameters
|
|
if limit:
|
|
params["limit"] = limit
|
|
if page:
|
|
params["page"] = page
|
|
|
|
# Add incremental update filter
|
|
if updated_at:
|
|
params["filter[updated_at][from]"] = updated_at
|
|
|
|
# Add entity-specific 'with' parameter
|
|
if config["with"]:
|
|
params["with"] = config["with"]
|
|
|
|
# Add any additional parameters
|
|
params.update(kwargs)
|
|
|
|
# Special handling for events - add default type filter
|
|
if entity_type == "events" and "filter[type]" not in params:
|
|
params["filter[type]"] = ",".join(["incoming_call", "outgoing_call", "lead_status_changed"])
|
|
|
|
# Fetch data from API
|
|
response = await self._make_request(config["endpoint"], params)
|
|
|
|
# Extract entities from response
|
|
embedded = response.get("_embedded", {})
|
|
entities = embedded.get(config["key"], [])
|
|
|
|
return entities
|
|
|
|
async def fetch_all_data(self) -> Dict[str, Any]:
|
|
"""Fetch all data from AMO CRM for testing"""
|
|
print("Fetching AMO CRM data...")
|
|
|
|
results = {}
|
|
|
|
try:
|
|
print("- Fetching users...")
|
|
results["users"] = await self.get_users()
|
|
print(f" Found {len(results['users'].get('_embedded', {}).get('users', []))} users")
|
|
|
|
print("- Fetching pipelines...")
|
|
results["pipelines"] = await self.get_pipelines()
|
|
print(f" Found {len(results['pipelines'].get('_embedded', {}).get('pipelines', []))} pipelines")
|
|
|
|
print("- Fetching companies...")
|
|
results["companies"] = await self.get_companies(limit=10) # Limit for testing
|
|
print(f" Found {len(results['companies'].get('_embedded', {}).get('companies', []))} companies")
|
|
|
|
print("- Fetching contacts...")
|
|
results["contacts"] = await self.get_contacts(limit=10) # Limit for testing
|
|
print(f" Found {len(results['contacts'].get('_embedded', {}).get('contacts', []))} contacts")
|
|
|
|
print("- Fetching deals...")
|
|
results["deals"] = await self.get_deals(limit=10) # Limit for testing
|
|
print(f" Found {len(results['deals'].get('_embedded', {}).get('leads', []))} deals")
|
|
|
|
print("- Fetching events...")
|
|
results["events"] = await self.get_events(limit=10) # Limit for testing
|
|
print(f" Found {len(results['events'].get('_embedded', {}).get('events', []))} events")
|
|
|
|
print("- Fetching custom fields...")
|
|
results["custom_fields_deals"] = await self.get_custom_fields("leads")
|
|
results["custom_fields_contacts"] = await self.get_custom_fields("contacts")
|
|
results["custom_fields_companies"] = await self.get_custom_fields("companies")
|
|
|
|
print("✅ All data fetched successfully!")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error fetching data: {e}")
|
|
raise
|
|
|
|
return results
|
|
|
|
|
|
# Convenience function for testing
|
|
async def fetch_amocrm_data():
|
|
"""Fetch AMO CRM data for testing"""
|
|
client = AmoCRMClient()
|
|
return await client.fetch_all_data()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Test the client
|
|
asyncio.run(fetch_amocrm_data())
|