Spaces:
Sleeping
Sleeping
File size: 4,359 Bytes
2358888 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
"""
Base simulator class for all service integrations.
All service-specific simulators inherit from this class.
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
import json
class BaseSimulator(ABC):
"""Base class for all service simulators."""
def __init__(self, service_name: str):
self.service_name = service_name
self.mock_responses = {}
self.load_mock_responses()
@abstractmethod
def load_mock_responses(self):
"""Load mock responses for this service."""
pass
@abstractmethod
def get_required_permissions(self, action: str) -> set[str]:
"""Get required permissions for an action."""
pass
@abstractmethod
def validate_params(self, action: str, params: dict) -> tuple[bool, Optional[str]]:
"""
Validate parameters for an action.
Returns:
(is_valid, error_message)
"""
pass
@abstractmethod
def generate_mock_response(self, action: str, params: dict) -> dict:
"""Generate a realistic mock response for the action."""
pass
def execute(self, action: str, params: dict, mock: bool = True, credentials: Optional[dict] = None) -> dict:
"""
Execute an API action (mock or real).
Args:
action: The action to perform (e.g., 'get_pull_request')
params: Parameters for the action
mock: If True, return mock response. If False, call real API.
credentials: API credentials (required for real mode)
Returns:
Response dict with data or error
"""
# Validate parameters
is_valid, error_msg = self.validate_params(action, params)
if not is_valid:
return {
'success': False,
'error': 'INVALID_PARAMS',
'message': error_msg,
'service': self.service_name,
'action': action
}
if mock:
return self._execute_mock(action, params)
else:
if not credentials:
return {
'success': False,
'error': 'MISSING_CREDENTIALS',
'message': 'Credentials required for real API calls',
'service': self.service_name,
'action': action
}
return self._execute_real(action, params, credentials)
def _execute_mock(self, action: str, params: dict) -> dict:
"""Execute in mock mode."""
try:
response_data = self.generate_mock_response(action, params)
return {
'success': True,
'mode': 'mock',
'service': self.service_name,
'action': action,
'data': response_data,
'note': '⚡ This was a simulated response - no real API call made'
}
except Exception as e:
return {
'success': False,
'error': 'MOCK_GENERATION_ERROR',
'message': str(e),
'service': self.service_name,
'action': action
}
def _execute_real(self, action: str, params: dict, credentials: dict) -> dict:
"""
Execute real API call.
Subclasses should override this to implement real API calls.
"""
return {
'success': False,
'error': 'NOT_IMPLEMENTED',
'message': f'Real API mode not yet implemented for {self.service_name}.{action}',
'service': self.service_name,
'action': action,
'note': 'Override _execute_real() in your simulator to enable real API calls'
}
def estimate_tokens(self, response: dict) -> int:
"""
Estimate token count for a response.
Uses simple heuristic: ~4 characters per token for JSON.
Will be improved with tiktoken in later phases.
"""
response_str = json.dumps(response)
return len(response_str) // 4
def get_action_list(self) -> list[str]:
"""Get list of available actions for this service."""
return list(self.mock_responses.keys())
|