| """
|
| CDMS (Common Data Model Schema) API client for label access.
|
| """
|
|
|
| from typing import Dict, List, Any, Optional
|
| import json
|
| from datetime import datetime
|
|
|
| from .base_client import BaseAPIClient, HTTPMethod, APIResponse
|
| from config.settings import settings
|
| from src.utils.logging_config import logger
|
|
|
|
|
| class CDMSClient(BaseAPIClient):
|
| """Client for CDMS label access API."""
|
|
|
| def __init__(
|
| self,
|
| api_url: Optional[str] = None,
|
| api_key: Optional[str] = None,
|
| timeout: int = None
|
| ):
|
| """
|
| Initialize CDMS client.
|
|
|
| Args:
|
| api_url: CDMS API base URL
|
| api_key: API key for authentication
|
| timeout: Request timeout
|
| """
|
| self.api_url = api_url or settings.cdms_api_url
|
| self.api_key = api_key or settings.cdms_api_key
|
|
|
| if not self.api_url:
|
| raise ValueError("CDMS API URL is required")
|
|
|
| if not self.api_key:
|
| logger.warning("CDMS API key not provided. Some operations may fail.")
|
|
|
|
|
| auth_headers = {}
|
| if self.api_key:
|
| auth_headers['Authorization'] = f'Bearer {self.api_key}'
|
| auth_headers['X-API-Key'] = self.api_key
|
|
|
| super().__init__(
|
| base_url=self.api_url,
|
| timeout=timeout or settings.cdms_timeout,
|
| auth_headers=auth_headers
|
| )
|
|
|
| self._authenticated = False
|
|
|
| def authenticate(self) -> bool:
|
| """
|
| Authenticate with CDMS API.
|
|
|
| Returns:
|
| True if authentication successful
|
| """
|
| if not self.api_key:
|
| logger.error("Cannot authenticate without API key")
|
| return False
|
|
|
| try:
|
|
|
| response = self.get('/health')
|
|
|
| if response.success:
|
| self._authenticated = True
|
| logger.info("CDMS authentication successful")
|
| return True
|
| else:
|
| logger.error(f"CDMS authentication failed: {response.error_message}")
|
| return False
|
|
|
| except Exception as e:
|
| logger.error(f"CDMS authentication error: {e}")
|
| return False
|
|
|
| def get_api_info(self) -> Dict[str, Any]:
|
| """Get CDMS API information."""
|
| try:
|
| response = self.get('/info')
|
|
|
| if response.success:
|
| return response.data
|
| else:
|
| logger.warning(f"Failed to get API info: {response.error_message}")
|
| return {
|
| 'name': 'CDMS API',
|
| 'version': 'unknown',
|
| 'status': 'error',
|
| 'error': response.error_message
|
| }
|
|
|
| except Exception as e:
|
| logger.error(f"Error getting API info: {e}")
|
| return {
|
| 'name': 'CDMS API',
|
| 'version': 'unknown',
|
| 'status': 'error',
|
| 'error': str(e)
|
| }
|
|
|
| def get_labels(
|
| self,
|
| dataset_id: Optional[str] = None,
|
| label_type: Optional[str] = None,
|
| limit: int = 100,
|
| offset: int = 0
|
| ) -> APIResponse:
|
| """
|
| Retrieve CDMS labels.
|
|
|
| Args:
|
| dataset_id: Filter by dataset ID
|
| label_type: Filter by label type
|
| limit: Maximum number of labels to return
|
| offset: Offset for pagination
|
|
|
| Returns:
|
| APIResponse with label data
|
| """
|
| params = {
|
| 'limit': limit,
|
| 'offset': offset
|
| }
|
|
|
| if dataset_id:
|
| params['dataset_id'] = dataset_id
|
|
|
| if label_type:
|
| params['label_type'] = label_type
|
|
|
| response = self.get('/labels', params=params)
|
|
|
| if response.success:
|
| logger.info(f"Retrieved {len(response.data.get('labels', []))} labels")
|
| else:
|
| logger.error(f"Failed to retrieve labels: {response.error_message}")
|
|
|
| return response
|
|
|
| def get_label_by_id(self, label_id: str) -> APIResponse:
|
| """
|
| Get a specific label by ID.
|
|
|
| Args:
|
| label_id: Label identifier
|
|
|
| Returns:
|
| APIResponse with label data
|
| """
|
| response = self.get(f'/labels/{label_id}')
|
|
|
| if response.success:
|
| logger.info(f"Retrieved label: {label_id}")
|
| else:
|
| logger.error(f"Failed to retrieve label {label_id}: {response.error_message}")
|
|
|
| return response
|
|
|
| def search_labels(
|
| self,
|
| query: str,
|
| filters: Optional[Dict[str, Any]] = None,
|
| limit: int = 50
|
| ) -> APIResponse:
|
| """
|
| Search labels by query.
|
|
|
| Args:
|
| query: Search query
|
| filters: Additional filters
|
| limit: Maximum results
|
|
|
| Returns:
|
| APIResponse with search results
|
| """
|
| data = {
|
| 'query': query,
|
| 'limit': limit
|
| }
|
|
|
| if filters:
|
| data['filters'] = filters
|
|
|
| response = self.post('/labels/search', json_data=data)
|
|
|
| if response.success:
|
| results_count = len(response.data.get('results', []))
|
| logger.info(f"Search returned {results_count} results for query: '{query}'")
|
| else:
|
| logger.error(f"Label search failed: {response.error_message}")
|
|
|
| return response
|
|
|
| def get_datasets(self, limit: int = 100, offset: int = 0) -> APIResponse:
|
| """
|
| Get available datasets.
|
|
|
| Args:
|
| limit: Maximum number of datasets
|
| offset: Offset for pagination
|
|
|
| Returns:
|
| APIResponse with dataset list
|
| """
|
| params = {
|
| 'limit': limit,
|
| 'offset': offset
|
| }
|
|
|
| response = self.get('/datasets', params=params)
|
|
|
| if response.success:
|
| logger.info(f"Retrieved {len(response.data.get('datasets', []))} datasets")
|
| else:
|
| logger.error(f"Failed to retrieve datasets: {response.error_message}")
|
|
|
| return response
|
|
|
| def get_dataset_labels(self, dataset_id: str) -> APIResponse:
|
| """
|
| Get labels for a specific dataset.
|
|
|
| Args:
|
| dataset_id: Dataset identifier
|
|
|
| Returns:
|
| APIResponse with dataset labels
|
| """
|
| response = self.get(f'/datasets/{dataset_id}/labels')
|
|
|
| if response.success:
|
| labels_count = len(response.data.get('labels', []))
|
| logger.info(f"Retrieved {labels_count} labels for dataset: {dataset_id}")
|
| else:
|
| logger.error(f"Failed to retrieve labels for dataset {dataset_id}: {response.error_message}")
|
|
|
| return response
|
|
|
| def create_label(self, label_data: Dict[str, Any]) -> APIResponse:
|
| """
|
| Create a new label.
|
|
|
| Args:
|
| label_data: Label data dictionary
|
|
|
| Returns:
|
| APIResponse with created label
|
| """
|
|
|
| if 'created_at' not in label_data:
|
| label_data['created_at'] = datetime.utcnow().isoformat()
|
|
|
| response = self.post('/labels', json_data=label_data)
|
|
|
| if response.success:
|
| label_id = response.data.get('id', 'unknown')
|
| logger.info(f"Created label: {label_id}")
|
| else:
|
| logger.error(f"Failed to create label: {response.error_message}")
|
|
|
| return response
|
|
|
| def update_label(self, label_id: str, label_data: Dict[str, Any]) -> APIResponse:
|
| """
|
| Update an existing label.
|
|
|
| Args:
|
| label_id: Label identifier
|
| label_data: Updated label data
|
|
|
| Returns:
|
| APIResponse with updated label
|
| """
|
|
|
| label_data['updated_at'] = datetime.utcnow().isoformat()
|
|
|
| response = self.put(f'/labels/{label_id}', json_data=label_data)
|
|
|
| if response.success:
|
| logger.info(f"Updated label: {label_id}")
|
| else:
|
| logger.error(f"Failed to update label {label_id}: {response.error_message}")
|
|
|
| return response
|
|
|
| def delete_label(self, label_id: str) -> APIResponse:
|
| """
|
| Delete a label.
|
|
|
| Args:
|
| label_id: Label identifier
|
|
|
| Returns:
|
| APIResponse confirming deletion
|
| """
|
| response = self.delete(f'/labels/{label_id}')
|
|
|
| if response.success:
|
| logger.info(f"Deleted label: {label_id}")
|
| else:
|
| logger.error(f"Failed to delete label {label_id}: {response.error_message}")
|
|
|
| return response
|
|
|
| def get_label_metadata(self, label_id: str) -> APIResponse:
|
| """
|
| Get metadata for a label.
|
|
|
| Args:
|
| label_id: Label identifier
|
|
|
| Returns:
|
| APIResponse with label metadata
|
| """
|
| response = self.get(f'/labels/{label_id}/metadata')
|
|
|
| if response.success:
|
| logger.info(f"Retrieved metadata for label: {label_id}")
|
| else:
|
| logger.error(f"Failed to retrieve metadata for label {label_id}: {response.error_message}")
|
|
|
| return response
|
|
|
| def bulk_import_labels(self, labels: List[Dict[str, Any]]) -> APIResponse:
|
| """
|
| Import multiple labels in bulk.
|
|
|
| Args:
|
| labels: List of label data dictionaries
|
|
|
| Returns:
|
| APIResponse with import results
|
| """
|
| data = {
|
| 'labels': labels,
|
| 'import_timestamp': datetime.utcnow().isoformat()
|
| }
|
|
|
| response = self.post('/labels/bulk-import', json_data=data)
|
|
|
| if response.success:
|
| imported_count = response.data.get('imported_count', 0)
|
| logger.info(f"Bulk imported {imported_count} labels")
|
| else:
|
| logger.error(f"Bulk import failed: {response.error_message}")
|
|
|
| return response
|
|
|
| def export_labels(
|
| self,
|
| dataset_id: Optional[str] = None,
|
| format: str = 'json'
|
| ) -> APIResponse:
|
| """
|
| Export labels in specified format.
|
|
|
| Args:
|
| dataset_id: Optional dataset filter
|
| format: Export format (json, csv, xml)
|
|
|
| Returns:
|
| APIResponse with exported data
|
| """
|
| params = {
|
| 'format': format
|
| }
|
|
|
| if dataset_id:
|
| params['dataset_id'] = dataset_id
|
|
|
| response = self.get('/labels/export', params=params)
|
|
|
| if response.success:
|
| logger.info(f"Exported labels in {format} format")
|
| else:
|
| logger.error(f"Export failed: {response.error_message}")
|
|
|
| return response
|
|
|
| def get_statistics(self) -> APIResponse:
|
| """
|
| Get CDMS statistics.
|
|
|
| Returns:
|
| APIResponse with statistics
|
| """
|
| response = self.get('/statistics')
|
|
|
| if response.success:
|
| logger.info("Retrieved CDMS statistics")
|
| else:
|
| logger.error(f"Failed to retrieve statistics: {response.error_message}")
|
|
|
| return response
|
|
|
| def health_check(self) -> bool:
|
| """
|
| Check if CDMS API is healthy.
|
|
|
| Returns:
|
| True if API is healthy
|
| """
|
| try:
|
| response = self.get('/health')
|
| return response.success and response.data.get('status') == 'healthy'
|
| except Exception as e:
|
| logger.error(f"Health check failed: {e}")
|
| return False
|
|
|
