74 lines
No EOL
2.6 KiB
Python
74 lines
No EOL
2.6 KiB
Python
from enum import Enum
|
|
from typing import Dict, List, Optional, Any, Literal
|
|
from datetime import datetime, UTC
|
|
from pydantic import BaseModel, Field
|
|
|
|
class AuthSource(str, Enum):
|
|
LOCAL = "local"
|
|
MSFT = "msft"
|
|
GOOGLE = "google"
|
|
# Add more auth sources as needed
|
|
|
|
class MethodParameter(BaseModel):
|
|
"""Model for method parameters"""
|
|
name: str
|
|
type: str
|
|
required: bool
|
|
validation: Optional[callable] = None
|
|
description: str
|
|
|
|
class MethodResult(BaseModel):
|
|
"""Model for method results"""
|
|
success: bool
|
|
data: Dict[str, Any]
|
|
metadata: Dict[str, Any] = Field(default_factory=dict)
|
|
validation: List[str] = Field(default_factory=list)
|
|
|
|
class MethodBase:
|
|
"""Base class for all methods"""
|
|
|
|
def __init__(self):
|
|
self.name: str
|
|
self.description: str
|
|
self.auth_source: AuthSource = AuthSource.LOCAL # Default to local auth
|
|
|
|
@property
|
|
def actions(self) -> Dict[str, Dict[str, Any]]:
|
|
"""Available actions and their parameters"""
|
|
raise NotImplementedError
|
|
|
|
async def execute(self, action: str, parameters: Dict[str, Any], auth_data: Optional[Dict[str, Any]] = None) -> MethodResult:
|
|
"""Execute method action with authentication data"""
|
|
raise NotImplementedError
|
|
|
|
async def validate_parameters(self, action: str, parameters: Dict[str, Any]) -> bool:
|
|
"""Validate action parameters"""
|
|
if action not in self.actions:
|
|
return False
|
|
|
|
action_def = self.actions[action]
|
|
required_params = {k for k, v in action_def['parameters'].items() if v['required']}
|
|
return all(param in parameters for param in required_params)
|
|
|
|
async def rollback(self, action: str, parameters: Dict[str, Any], auth_data: Optional[Dict[str, Any]] = None) -> None:
|
|
"""Rollback action if needed"""
|
|
pass
|
|
|
|
def _validate_auth(self, auth_data: Optional[Dict[str, Any]] = None) -> bool:
|
|
"""Validate authentication data"""
|
|
if self.auth_source == AuthSource.LOCAL:
|
|
return True
|
|
return bool(auth_data and auth_data.get('source') == self.auth_source)
|
|
|
|
def _create_result(self, success: bool, data: Dict[str, Any], metadata: Optional[Dict[str, Any]] = None) -> MethodResult:
|
|
"""Create a method result"""
|
|
return MethodResult(
|
|
success=success,
|
|
data=data,
|
|
metadata=metadata or {},
|
|
validation=[]
|
|
)
|
|
|
|
def _add_validation_message(self, result: MethodResult, message: str) -> None:
|
|
"""Add a validation message to the result"""
|
|
result.validation.append(message) |