added e2e testsuite
This commit is contained in:
220
custom/scripts/espocrm_api_client.py
Normal file
220
custom/scripts/espocrm_api_client.py
Normal file
@@ -0,0 +1,220 @@
|
||||
#!/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
|
||||
Reference in New Issue
Block a user