espocrm/custom/scripts/espocrm_api_client.py

#!/usr/bin/env python3
"""
EspoCRM API Client
Provides a clean interface to the EspoCRM REST API
"""

import requests
import json
from typing import Dict, List, Optional, Any
from urllib.parse import urljoin


class EspoCRMAPIClient:
    """Client for EspoCRM REST API"""
    
    def __init__(self, base_url: str, api_key: str, username: str):
        """
        Initialize API client
        
        Args:
            base_url: Base URL of EspoCRM (e.g., 'https://crm.bitbylaw.com')
            api_key: API authentication token
            username: Username for API access
        """
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.username = username
        self.session = requests.Session()
        self.session.headers.update({
            'X-Api-Key': api_key,
            'Content-Type': 'application/json'
        })
        
    def _make_request(self, method: str, endpoint: str, data: Optional[Dict] = None, 
                     params: Optional[Dict] = None) -> Dict:
        """
        Make HTTP request to API
        
        Args:
            method: HTTP method (GET, POST, PUT, DELETE)
            endpoint: API endpoint (e.g., 'Contact')
            data: Request body data
            params: URL query parameters
            
        Returns:
            Response data as dict
            
        Raises:
            requests.HTTPError: If request fails
        """
        url = urljoin(f"{self.base_url}/api/v1/", endpoint)
        
        try:
            response = self.session.request(
                method=method,
                url=url,
                json=data,
                params=params
            )
            response.raise_for_status()
            
            if response.status_code == 204:  # No content
                return {}
            
            return response.json() if response.content else {}
            
        except requests.exceptions.HTTPError as e:
            print(f"❌ API Error: {method} {url}")
            print(f"   Status: {e.response.status_code}")
            if e.response.content:
                try:
                    error_data = e.response.json()
                    print(f"   Error: {json.dumps(error_data, indent=2)}")
                except:
                    print(f"   Response: {e.response.text}")
            raise
    
    # ============================================================================
    # CRUD Operations
    # ============================================================================
    
    def create(self, entity_type: str, data: Dict) -> Dict:
        """
        Create a new record
        
        Args:
            entity_type: Entity type (e.g., 'Contact', 'CMietobjekt')
            data: Record data
            
        Returns:
            Created record with ID
        """
        return self._make_request('POST', entity_type, data=data)
    
    def read(self, entity_type: str, record_id: str) -> Dict:
        """
        Read a record by ID
        
        Args:
            entity_type: Entity type
            record_id: Record ID
            
        Returns:
            Record data
        """
        return self._make_request('GET', f"{entity_type}/{record_id}")
    
    def update(self, entity_type: str, record_id: str, data: Dict) -> Dict:
        """
        Update a record
        
        Args:
            entity_type: Entity type
            record_id: Record ID
            data: Updated fields
            
        Returns:
            Updated record
        """
        return self._make_request('PUT', f"{entity_type}/{record_id}", data=data)
    
    def delete(self, entity_type: str, record_id: str) -> Dict:
        """
        Delete a record
        
        Args:
            entity_type: Entity type
            record_id: Record ID
            
        Returns:
            Empty dict on success
        """
        return self._make_request('DELETE', f"{entity_type}/{record_id}")
    
    def list(self, entity_type: str, params: Optional[Dict] = None) -> List[Dict]:
        """
        List records
        
        Args:
            entity_type: Entity type
            params: Query parameters (offset, maxSize, where, etc.)
            
        Returns:
            List of records
        """
        response = self._make_request('GET', entity_type, params=params)
        return response.get('list', [])
    
    # ============================================================================
    # Relationship Operations
    # ============================================================================
    
    def link(self, entity_type: str, record_id: str, link_name: str, 
             foreign_id: str) -> Dict:
        """
        Link two records
        
        Args:
            entity_type: Source entity type
            record_id: Source record ID
            link_name: Relationship name
            foreign_id: Target record ID
            
        Returns:
            Empty dict on success
        """
        endpoint = f"{entity_type}/{record_id}/{link_name}/{foreign_id}"
        return self._make_request('POST', endpoint)
    
    def unlink(self, entity_type: str, record_id: str, link_name: str, 
               foreign_id: str) -> Dict:
        """
        Unlink two records
        
        Args:
            entity_type: Source entity type
            record_id: Source record ID
            link_name: Relationship name
            foreign_id: Target record ID
            
        Returns:
            Empty dict on success
        """
        endpoint = f"{entity_type}/{record_id}/{link_name}/{foreign_id}"
        return self._make_request('DELETE', endpoint)
    
    def get_linked(self, entity_type: str, record_id: str, link_name: str) -> List[Dict]:
        """
        Get linked records
        
        Args:
            entity_type: Source entity type
            record_id: Source record ID
            link_name: Relationship name
            
        Returns:
            List of linked records
        """
        endpoint = f"{entity_type}/{record_id}/{link_name}"
        response = self._make_request('GET', endpoint)
        return response.get('list', [])
    
    # ============================================================================
    # Utility Methods
    # ============================================================================
    
    def check_connection(self) -> bool:
        """
        Test API connection
        
        Returns:
            True if connection successful
        """
        try:
            # Try to get current user info
            self._make_request('GET', 'User')
            return True
        except Exception as e:
            print(f"❌ Connection failed: {e}")
            return False