diff --git a/app/adapters/__init__.py b/app/adapters/__init__.py new file mode 100644 index 00000000..539bef8e --- /dev/null +++ b/app/adapters/__init__.py @@ -0,0 +1,18 @@ +""" +FloTorch adapters package for external service integration. +""" +from .opensearch_adapter import OpenSearchAdapter +from .retriever_adapter import RetrieverAdapter +from .indexer_adapter import IndexerAdapter +from .eval_adapter import EvalAdapter +from .http_service_adapter import HTTPServiceAdapter, ServiceConfig + +__all__ = [ + 'OpenSearchAdapter', + 'RetrieverAdapter', + 'IndexerAdapter', + 'EvalAdapter', + 'HTTPServiceAdapter', + 'ServiceConfig' +] + diff --git a/app/adapters/dual_write_adapter.py b/app/adapters/dual_write_adapter.py new file mode 100644 index 00000000..72838a78 --- /dev/null +++ b/app/adapters/dual_write_adapter.py @@ -0,0 +1,363 @@ +""" +Dual write adapter that writes to both DynamoDB and PostgreSQL during migration. +""" +import os +import logging +from typing import Dict, Any, Optional, List +from app.adapters.postgres_adapter import PostgresAdapter +from flotorch_core.storage.db.db_storage import DBStorage +import boto3 + +logger = logging.getLogger(__name__) + +class DualWriteAdapter: + """ + Adapter that writes to both DynamoDB and PostgreSQL for gradual migration. + Reads can be configured to use either database as primary. + """ + + def __init__(self, table_name: str, aws_region: str = "us-east-1", **kwargs): + self.table_name = table_name + self.aws_region = aws_region + + # Configuration flags + self.dual_write_enabled = os.getenv("ENABLE_DUAL_WRITES", "false").lower() == "true" + self.read_from_postgres = os.getenv("READ_FROM_POSTGRES", "false").lower() == "true" + + # Initialize adapters + self.postgres_adapter = PostgresAdapter(table_name=table_name, **kwargs) + self.dynamodb_adapter = None + + if self.dual_write_enabled: + self._init_dynamodb_adapter() + + logger.info(f"DualWriteAdapter initialized for {table_name}: " + f"dual_write={self.dual_write_enabled}, " + f"read_from_postgres={self.read_from_postgres}") + + def _init_dynamodb_adapter(self): + """Initialize DynamoDB adapter for dual writes.""" + try: + # Map table names to DynamoDB table names + table_mapping = { + "experiments": os.getenv("experiment_table", "flotorch_experiment"), + "executions": os.getenv("execution_table", "flotorch_execution"), + "question_metrics": os.getenv("experiment_question_metrics_table", "flotorch_question_metrics"), + "model_invocations": os.getenv("execution_model_invocations_table", "flotorch_model_invocations") + } + + dynamodb_table = table_mapping.get(self.table_name) + if not dynamodb_table: + logger.warning(f"No DynamoDB table mapping for {self.table_name}") + return + + # Create a lightweight wrapper around boto3 client to mimic the previous API + ddb_client = boto3.client("dynamodb", region_name=self.aws_region) + + class _TableWrapper: + def __init__(self, client, table_name): + self._client = client + self._name = table_name + + def put_item(self, **kwargs): + return self._client.put_item(TableName=self._name, **kwargs) + + def get_item(self, **kwargs): + return self._client.get_item(TableName=self._name, **kwargs) + + def update_item(self, **kwargs): + return self._client.update_item(TableName=self._name, **kwargs) + + def delete_item(self, **kwargs): + return self._client.delete_item(TableName=self._name, **kwargs) + + def scan(self, **kwargs): + return self._client.scan(TableName=self._name, **kwargs) + + class _DynAdapter: + def __init__(self, client, table_name): + self.table = _TableWrapper(client, table_name) + + self.dynamodb_adapter = _DynAdapter(ddb_client, dynamodb_table) + logger.info(f"DynamoDB adapter initialized for {self.table_name} -> {dynamodb_table}") + + except Exception as e: + logger.error(f"Failed to initialize DynamoDB adapter: {e}") + self.dynamodb_adapter = None + + def create_item(self, item_data: Dict[str, Any]) -> Dict[str, Any]: + """Create item in both databases if dual write is enabled.""" + results = {} + primary_result = None + + try: + # Write to PostgreSQL (always primary for writes) + primary_result = self.postgres_adapter.create_item(item_data) + results["postgresql"] = {"status": "success", "data": primary_result} + + # Write to DynamoDB if dual write is enabled + if self.dual_write_enabled and self.dynamodb_adapter: + try: + dynamodb_result = self._create_dynamodb_item(item_data) + results["dynamodb"] = {"status": "success", "data": dynamodb_result} + except Exception as e: + logger.error(f"DynamoDB dual write failed: {e}") + results["dynamodb"] = {"status": "failed", "error": str(e)} + + return primary_result + + except Exception as e: + logger.error(f"Primary write (PostgreSQL) failed: {e}") + results["postgresql"] = {"status": "failed", "error": str(e)} + raise + + def get_item(self, key: str, value: Any) -> Optional[Dict[str, Any]]: + """Get item from configured primary database.""" + if self.read_from_postgres: + # Read from PostgreSQL + try: + result = self.postgres_adapter.get_item(key, value) + if result: + return result + except Exception as e: + logger.error(f"PostgreSQL read failed: {e}") + # Fallback to DynamoDB + if self.dynamodb_adapter: + return self._get_dynamodb_item(key, value) + else: + # Read from DynamoDB (legacy behavior) + if self.dynamodb_adapter: + result = self._get_dynamodb_item(key, value) + if result: + return result + + # Fallback to PostgreSQL + try: + return self.postgres_adapter.get_item(key, value) + except Exception as e: + logger.error(f"PostgreSQL fallback read failed: {e}") + + return None + + def update_item(self, key: str, value: Any, update_data: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Update item in both databases if dual write is enabled.""" + results = {} + primary_result = None + + try: + # Update PostgreSQL (always primary for writes) + primary_result = self.postgres_adapter.update_item(key, value, update_data) + results["postgresql"] = {"status": "success", "data": primary_result} + + # Update DynamoDB if dual write is enabled + if self.dual_write_enabled and self.dynamodb_adapter: + try: + dynamodb_result = self._update_dynamodb_item(key, value, update_data) + results["dynamodb"] = {"status": "success", "data": dynamodb_result} + except Exception as e: + logger.error(f"DynamoDB dual write update failed: {e}") + results["dynamodb"] = {"status": "failed", "error": str(e)} + + return primary_result + + except Exception as e: + logger.error(f"Primary update (PostgreSQL) failed: {e}") + results["postgresql"] = {"status": "failed", "error": str(e)} + raise + + def delete_item(self, key: str, value: Any) -> bool: + """Delete item from both databases if dual write is enabled.""" + results = {} + primary_result = False + + try: + # Delete from PostgreSQL (always primary for writes) + primary_result = self.postgres_adapter.delete_item(key, value) + results["postgresql"] = {"status": "success", "deleted": primary_result} + + # Delete from DynamoDB if dual write is enabled + if self.dual_write_enabled and self.dynamodb_adapter: + try: + dynamodb_result = self._delete_dynamodb_item(key, value) + results["dynamodb"] = {"status": "success", "deleted": dynamodb_result} + except Exception as e: + logger.error(f"DynamoDB dual write delete failed: {e}") + results["dynamodb"] = {"status": "failed", "error": str(e)} + + return primary_result + + except Exception as e: + logger.error(f"Primary delete (PostgreSQL) failed: {e}") + results["postgresql"] = {"status": "failed", "error": str(e)} + raise + + def query_items(self, filters: Optional[Dict[str, Any]] = None, + limit: Optional[int] = None, + offset: Optional[int] = None) -> List[Dict[str, Any]]: + """Query items from configured primary database.""" + if self.read_from_postgres: + # Query PostgreSQL + try: + return self.postgres_adapter.query_items(filters, limit, offset) + except Exception as e: + logger.error(f"PostgreSQL query failed: {e}") + # Fallback to DynamoDB + if self.dynamodb_adapter: + return self._query_dynamodb_items(filters, limit, offset) + else: + # Query DynamoDB (legacy behavior) + if self.dynamodb_adapter: + result = self._query_dynamodb_items(filters, limit, offset) + if result: + return result + + # Fallback to PostgreSQL + try: + return self.postgres_adapter.query_items(filters, limit, offset) + except Exception as e: + logger.error(f"PostgreSQL fallback query failed: {e}") + + return [] + + def _create_dynamodb_item(self, item_data: Dict[str, Any]) -> Dict[str, Any]: + """Create item in DynamoDB.""" + # Convert PostgreSQL format to DynamoDB format + dynamodb_item = self._convert_to_dynamodb_format(item_data) + + response = self.dynamodb_adapter.table.put_item(Item=dynamodb_item) + return {"status": "created", "item": item_data} + + def _get_dynamodb_item(self, key: str, value: Any) -> Optional[Dict[str, Any]]: + """Get item from DynamoDB.""" + try: + response = self.dynamodb_adapter.table.get_item( + Key={key: {"S": str(value)}} + ) + + if "Item" in response: + return self._convert_from_dynamodb_format(response["Item"]) + return None + except Exception as e: + logger.error(f"DynamoDB get_item failed: {e}") + return None + + def _update_dynamodb_item(self, key: str, value: Any, update_data: Dict[str, Any]) -> Dict[str, Any]: + """Update item in DynamoDB.""" + # Convert update data to DynamoDB format + dynamodb_update = self._convert_to_dynamodb_format(update_data) + + # Build update expression + update_expression = "SET " + ", ".join([f"{k} = :{k}" for k in update_data.keys()]) + expression_values = {f":{k}": dynamodb_update[k] for k in update_data.keys()} + + response = self.dynamodb_adapter.table.update_item( + Key={key: {"S": str(value)}}, + UpdateExpression=update_expression, + ExpressionAttributeValues=expression_values, + ReturnValues="ALL_NEW" + ) + + return {"status": "updated", "data": update_data} + + def _delete_dynamodb_item(self, key: str, value: Any) -> bool: + """Delete item from DynamoDB.""" + try: + response = self.dynamodb_adapter.table.delete_item( + Key={key: {"S": str(value)}} + ) + return True + except Exception as e: + logger.error(f"DynamoDB delete_item failed: {e}") + return False + + def _query_dynamodb_items(self, filters: Optional[Dict[str, Any]] = None, + limit: Optional[int] = None, + offset: Optional[int] = None) -> List[Dict[str, Any]]: + """Query items from DynamoDB.""" + try: + # Simple scan implementation - in practice, you'd want to use query with indexes + scan_params = {} + if limit: + scan_params["Limit"] = limit + + response = self.dynamodb_adapter.table.scan(**scan_params) + items = response.get("Items", []) + + return [self._convert_from_dynamodb_format(item) for item in items] + except Exception as e: + logger.error(f"DynamoDB query failed: {e}") + return [] + + def _convert_to_dynamodb_format(self, item_data: Dict[str, Any]) -> Dict[str, Any]: + """Convert PostgreSQL item format to DynamoDB format.""" + dynamodb_item = {} + + for key, value in item_data.items(): + if value is None: + continue + elif isinstance(value, str): + dynamodb_item[key] = {"S": value} + elif isinstance(value, (int, float)): + dynamodb_item[key] = {"N": str(value)} + elif isinstance(value, bool): + dynamodb_item[key] = {"BOOL": value} + elif isinstance(value, dict): + dynamodb_item[key] = {"M": self._convert_dict_to_dynamodb(value)} + else: + dynamodb_item[key] = {"S": str(value)} + + return dynamodb_item + + def _convert_from_dynamodb_format(self, dynamodb_item: Dict[str, Any]) -> Dict[str, Any]: + """Convert DynamoDB item format to PostgreSQL format.""" + item = {} + + for key, value in dynamodb_item.items(): + if "S" in value: + item[key] = value["S"] + elif "N" in value: + try: + item[key] = float(value["N"]) if "." in value["N"] else int(value["N"]) + except: + item[key] = value["N"] + elif "BOOL" in value: + item[key] = value["BOOL"] + elif "M" in value: + item[key] = self._convert_dynamodb_map(value["M"]) + else: + item[key] = value + + return item + + def _convert_dict_to_dynamodb(self, data: Dict[str, Any]) -> Dict[str, Any]: + """Convert nested dictionary to DynamoDB format.""" + result = {} + for key, value in data.items(): + if isinstance(value, str): + result[key] = {"S": value} + elif isinstance(value, (int, float)): + result[key] = {"N": str(value)} + elif isinstance(value, bool): + result[key] = {"BOOL": value} + elif isinstance(value, dict): + result[key] = {"M": self._convert_dict_to_dynamodb(value)} + else: + result[key] = {"S": str(value)} + return result + + def _convert_dynamodb_map(self, dynamodb_map: Dict[str, Any]) -> Dict[str, Any]: + """Convert DynamoDB map to regular dictionary.""" + result = {} + for key, value in dynamodb_map.items(): + if "S" in value: + result[key] = value["S"] + elif "N" in value: + try: + result[key] = float(value["N"]) if "." in value["N"] else int(value["N"]) + except: + result[key] = value["N"] + elif "BOOL" in value: + result[key] = value["BOOL"] + elif "M" in value: + result[key] = self._convert_dynamodb_map(value["M"]) + return result diff --git a/app/adapters/eval_adapter.py b/app/adapters/eval_adapter.py new file mode 100644 index 00000000..c967739b --- /dev/null +++ b/app/adapters/eval_adapter.py @@ -0,0 +1,83 @@ +""" +Evaluation service adapter with feature flag to switch between local and external service. +""" +import os +import logging +from typing import Dict, Any, List, Optional +from app.adapters.http_service_adapter import HTTPServiceAdapter, ServiceConfig + +logger = logging.getLogger(__name__) + +class EvalAdapter: + """ + Adapter that can use either local evaluation implementation or external HTTP service. + Controlled by USE_EXTERNAL_SERVICES environment variable. + """ + + def __init__(self, config: Dict[str, Any], experimental_config: Dict[str, Any]): + self.use_external = os.getenv("USE_EXTERNAL_SERVICES", "false").lower() == "true" + + if self.use_external: + logger.info("Using external evaluation service") + self._init_external_service() + else: + logger.info("Using local evaluation implementation") + self._init_local() + + def _init_external_service(self): + """Initialize external evaluation service.""" + eval_url = os.getenv("EVAL_SERVICE_URL", "http://localhost:8003") + service_config = ServiceConfig( + base_url=eval_url, + timeout=int(os.getenv("EVAL_TIMEOUT", "120")), + max_retries=int(os.getenv("EVAL_MAX_RETRIES", "2")) + ) + self.client = HTTPServiceAdapter(service_config) + + def _init_local(self): + """Initialize local evaluation implementation.""" + from evaluation.eval import evaluate + self.evaluate_func = evaluate + + def evaluate(self, config: Any, experimental_config: Any) -> None: + """ + Execute evaluation process. + + Args: + config: Global configuration object + experimental_config: Experiment-specific configuration + """ + if self.use_external: + return self._evaluate_external(config, experimental_config) + else: + return self._evaluate_local(config, experimental_config) + + def _evaluate_external(self, config: Any, experimental_config: Any) -> None: + """Execute evaluation using external service.""" + try: + # Convert config objects to dictionaries for JSON serialization + config_dict = self._config_to_dict(config) + experimental_config_dict = self._config_to_dict(experimental_config) + + payload = { + "config": config_dict, + "experimental_config": experimental_config_dict + } + + response = self.client.post("/evaluate", json=payload) + logger.info(f"External evaluator response: {response}") + + except Exception as e: + logger.error(f"External evaluator failed: {e}") + raise + + def _evaluate_local(self, config: Any, experimental_config: Any) -> None: + """Execute evaluation using local implementation.""" + return self.evaluate_func(config, experimental_config) + + def _config_to_dict(self, config_obj: Any) -> Dict[str, Any]: + """Convert config object to dictionary for JSON serialization.""" + if hasattr(config_obj, '__dict__'): + return {k: v for k, v in config_obj.__dict__.items() + if not k.startswith('_')} + return config_obj diff --git a/app/adapters/http_service_adapter.py b/app/adapters/http_service_adapter.py new file mode 100644 index 00000000..e6d1aba6 --- /dev/null +++ b/app/adapters/http_service_adapter.py @@ -0,0 +1,82 @@ +""" +Base HTTP service adapter for external FloTorch services. +""" +import os +import requests +import logging +from typing import Dict, Any, Optional +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +@dataclass +class ServiceConfig: + """Configuration for external service endpoints.""" + base_url: str + timeout: int = 30 + max_retries: int = 3 + retry_delay: float = 1.0 + +class HTTPServiceAdapter: + """ + Base adapter for HTTP-based external services with retry logic and error handling. + """ + + def __init__(self, config: ServiceConfig): + self.config = config + self.session = requests.Session() + + # Configure retry strategy + from requests.adapters import HTTPAdapter + from urllib3.util.retry import Retry + + retry_strategy = Retry( + total=config.max_retries, + backoff_factor=config.retry_delay, + status_forcelist=[429, 500, 502, 503, 504], + ) + + adapter = HTTPAdapter(max_retries=retry_strategy) + self.session.mount("http://", adapter) + self.session.mount("https://", adapter) + + def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]: + """Make HTTP request with error handling and logging.""" + url = f"{self.config.base_url.rstrip('/')}/{endpoint.lstrip('/')}" + + try: + logger.info(f"Making {method.upper()} request to {url}") + response = self.session.request( + method=method, + url=url, + timeout=self.config.timeout, + **kwargs + ) + response.raise_for_status() + + # Try to parse JSON response + try: + return response.json() + except ValueError: + return {"data": response.text, "status_code": response.status_code} + + except requests.exceptions.RequestException as e: + logger.error(f"HTTP request failed: {e}") + raise Exception(f"Service request failed: {str(e)}") + + def get(self, endpoint: str, **kwargs) -> Dict[str, Any]: + """Make GET request.""" + return self._make_request("GET", endpoint, **kwargs) + + def post(self, endpoint: str, **kwargs) -> Dict[str, Any]: + """Make POST request.""" + return self._make_request("POST", endpoint, **kwargs) + + def put(self, endpoint: str, **kwargs) -> Dict[str, Any]: + """Make PUT request.""" + return self._make_request("PUT", endpoint, **kwargs) + + def delete(self, endpoint: str, **kwargs) -> Dict[str, Any]: + """Make DELETE request.""" + return self._make_request("DELETE", endpoint, **kwargs) + diff --git a/app/adapters/indexer_adapter.py b/app/adapters/indexer_adapter.py new file mode 100644 index 00000000..48f330c1 --- /dev/null +++ b/app/adapters/indexer_adapter.py @@ -0,0 +1,83 @@ +""" +Indexer service adapter with feature flag to switch between local and external service. +""" +import os +import logging +from typing import Dict, Any, List, Optional +from app.adapters.http_service_adapter import HTTPServiceAdapter, ServiceConfig + +logger = logging.getLogger(__name__) + +class IndexerAdapter: + """ + Adapter that can use either local indexer implementation or external HTTP service. + Controlled by USE_EXTERNAL_SERVICES environment variable. + """ + + def __init__(self, config: Dict[str, Any], experimental_config: Dict[str, Any]): + self.use_external = os.getenv("USE_EXTERNAL_SERVICES", "false").lower() == "true" + + if self.use_external: + logger.info("Using external indexer service") + self._init_external_service() + else: + logger.info("Using local indexer implementation") + self._init_local() + + def _init_external_service(self): + """Initialize external indexer service.""" + indexer_url = os.getenv("INDEXER_SERVICE_URL", "http://localhost:8002") + service_config = ServiceConfig( + base_url=indexer_url, + timeout=int(os.getenv("INDEXER_TIMEOUT", "300")), # Longer timeout for indexing + max_retries=int(os.getenv("INDEXER_MAX_RETRIES", "2")) + ) + self.client = HTTPServiceAdapter(service_config) + + def _init_local(self): + """Initialize local indexer implementation.""" + from indexing.indexing import chunk_embed_store + self.chunk_embed_store_func = chunk_embed_store + + def chunk_embed_store(self, config: Any, experimental_config: Any) -> None: + """ + Execute chunking, embedding, and storage process. + + Args: + config: Global configuration object + experimental_config: Experiment-specific configuration + """ + if self.use_external: + return self._index_external(config, experimental_config) + else: + return self._index_local(config, experimental_config) + + def _index_external(self, config: Any, experimental_config: Any) -> None: + """Execute indexing using external service.""" + try: + # Convert config objects to dictionaries for JSON serialization + config_dict = self._config_to_dict(config) + experimental_config_dict = self._config_to_dict(experimental_config) + + payload = { + "config": config_dict, + "experimental_config": experimental_config_dict + } + + response = self.client.post("/index", json=payload) + logger.info(f"External indexer response: {response}") + + except Exception as e: + logger.error(f"External indexer failed: {e}") + raise + + def _index_local(self, config: Any, experimental_config: Any) -> None: + """Execute indexing using local implementation.""" + return self.chunk_embed_store_func(config, experimental_config) + + def _config_to_dict(self, config_obj: Any) -> Dict[str, Any]: + """Convert config object to dictionary for JSON serialization.""" + if hasattr(config_obj, '__dict__'): + return {k: v for k, v in config_obj.__dict__.items() + if not k.startswith('_')} + return config_obj diff --git a/app/adapters/opensearch_adapter.py b/app/adapters/opensearch_adapter.py new file mode 100644 index 00000000..523783ba --- /dev/null +++ b/app/adapters/opensearch_adapter.py @@ -0,0 +1,86 @@ +""" +OpenSearch adapter backed by opensearch-py (no dependency on local `core`). +This avoids any 'core' imports and works with Docker OpenSearch out-of-the-box. +""" +import logging +from typing import Dict, Any, List, Optional +from opensearchpy import OpenSearch, RequestsHttpConnection + +logger = logging.getLogger(__name__) + +class OpenSearchAdapter: + """ + Adapter that uses opensearch-py client under the hood. + """ + + def __init__(self, host: str, use_ssl: bool = False, port: int = 9200, + is_serverless: bool = False, region: str = 'us-east-1', + username: str = None, password: str = None): + + # For local/dev Docker OpenSearch, defaults are: http on port 9200, no auth + http_auth = (username, password) if (username and password) else None + verify_certs = True if use_ssl else False + + self.client = OpenSearch( + hosts=[{"host": host, "port": port}], + http_auth=http_auth, + use_ssl=use_ssl, + verify_certs=verify_certs, + connection_class=RequestsHttpConnection, + timeout=30, + max_retries=3, + retry_on_timeout=True, + ) + + # ---------------- Convenience methods ---------------- + + # Delegate all methods to the underlying client + def create_index(self, index_name: str, mapping: Dict[str, Any], algorithm: str) -> None: + return self.client.create_index(index_name, mapping, algorithm) + + def update_index(self, index_name: str, new_mapping: Dict[str, Any]) -> None: + return self.client.update_index(index_name, new_mapping) + + def delete_index(self, index_name: str) -> None: + return self.client.delete_index(index_name) + + def insert_document(self, index_name: str, document: Dict[str, Any]) -> None: + return self.client.insert_document(index_name, document) + + def search(self, index_name: str, query_vector: List[float], k: int) -> List[Dict[str, Any]]: + return self.client.search(index_name, query_vector, k) + + def index_exists(self, index_name: str) -> bool: + return self.client.index_exists(index_name) + + def insert_chunk(self, index_name: str, text: str, embedding: List[float], + chunk_id: str, metadata: Dict = None): + return self.client.insert_chunk(index_name, text, embedding, chunk_id, metadata) + + def batch_insert_chunks(self, index_name: str, chunks: List[str], + chunk_embeddings: List[List[float]], + metadata: Optional[List[Dict]] = None, + batch_size: int = 100): + return self.client.batch_insert_chunks( + index_name, chunks, chunk_embeddings, metadata, batch_size + ) + + def print_opensearch_info(self): + try: + info = self.client.info() + logger.info(f"OpenSearch Version: {info.get('version', {}).get('number')}") + logger.info(f"Cluster Name: {info.get('cluster_name')}") + logger.info(f"Cluster UUID: {info.get('cluster_uuid')}") + return info + except Exception as e: + logger.error(f"Error getting OpenSearch info: {e}") + raise + + def index_chunk_embeddings(self, chunks: List[str], chunk_embeddings: List[List[float]], + indexing_algorithm: str, chunking_algorithm: str, + vector_dimension: int, metadata: List[Dict] = None, + chunk_size: int = 1200): + return self.client.index_chunk_embeddings( + chunks, chunk_embeddings, indexing_algorithm, chunking_algorithm, + vector_dimension, metadata, chunk_size + ) diff --git a/app/adapters/orchestrator_adapter.py b/app/adapters/orchestrator_adapter.py new file mode 100644 index 00000000..e30d00be --- /dev/null +++ b/app/adapters/orchestrator_adapter.py @@ -0,0 +1,139 @@ +""" +Orchestrator adapter with feature flag to switch between Step Functions and direct service calls. +""" +import os +import json +import boto3 +import logging +from typing import Dict, Any +from fastapi import HTTPException +from config.config import get_config +from app.adapters.retriever_adapter import RetrieverAdapter +from app.adapters.indexer_adapter import IndexerAdapter +from app.adapters.eval_adapter import EvalAdapter + +logger = logging.getLogger(__name__) + +class OrchestratorAdapter: + """ + Adapter that can use either Step Functions or direct service calls for orchestration. + Controlled by USE_DIRECT_ORCHESTRATION environment variable. + """ + + def __init__(self): + self.config = get_config() + self.use_direct = os.getenv("USE_DIRECT_ORCHESTRATION", "false").lower() == "true" + + if self.use_direct: + logger.info("Using direct service orchestration") + self._init_direct_orchestration() + else: + logger.info("Using Step Functions orchestration") + self._init_step_functions() + + def _init_direct_orchestration(self): + """Initialize direct service orchestration.""" + # Initialize service adapters + self.retriever = RetrieverAdapter({}, {}) + self.indexer = IndexerAdapter({}, {}) + self.eval = EvalAdapter({}, {}) + + def _init_step_functions(self): + """Initialize Step Functions orchestration.""" + try: + self.step_function_client = boto3.client( + "stepfunctions", + region_name=self.config.aws_region + ) + except Exception as e: + logger.error(f"Failed to initialize Step Function client: {e}") + raise HTTPException( + status_code=500, + detail="Failed to initialize AWS Step Function client" + ) + + def run_experiment_orchestration(self, execution_id: str) -> Dict[str, Any]: + """ + Execute experiment orchestration using either Step Functions or direct calls. + + Args: + execution_id (str): The execution ID + + Returns: + Dict[str, Any]: Response from orchestration + """ + if self.use_direct: + return self._run_direct_orchestration(execution_id) + else: + return self._run_step_functions_orchestration(execution_id) + + def _run_direct_orchestration(self, execution_id: str) -> Dict[str, Any]: + """Execute orchestration using direct service calls.""" + try: + logger.info(f"Starting direct orchestration for execution: {execution_id}") + + # Load execution and experiment configs from database + # This would need to be implemented based on your data access patterns + config = self.config + experimental_config = self._load_experimental_config(execution_id) + + # Execute pipeline steps in sequence + results = {} + + # Step 1: Indexing (if needed) + if experimental_config.get('needs_indexing', False): + logger.info("Executing indexing step") + self.indexer.chunk_embed_store(config, experimental_config) + results['indexing'] = 'completed' + + # Step 2: Retrieval + logger.info("Executing retrieval step") + self.retriever.retrieve(config, experimental_config) + results['retrieval'] = 'completed' + + # Step 3: Evaluation (if needed) + if experimental_config.get('needs_evaluation', False): + logger.info("Executing evaluation step") + self.eval.evaluate(config, experimental_config) + results['evaluation'] = 'completed' + + logger.info(f"Direct orchestration completed for execution: {execution_id}") + return { + 'executionArn': f'direct-{execution_id}', + 'status': 'SUCCEEDED', + 'results': results + } + + except Exception as e: + logger.error(f"Direct orchestration failed: {e}") + raise HTTPException( + status_code=500, + detail=f"Direct orchestration failed: {str(e)}" + ) + + def _run_step_functions_orchestration(self, execution_id: str) -> Dict[str, Any]: + """Execute orchestration using Step Functions.""" + try: + payload = {"execution_id": execution_id} + response = self.step_function_client.start_execution( + stateMachineArn=self.config.step_function_arn, + input=json.dumps(payload) + ) + + logger.info(f"Started Step Function with Execution ARN: {response['executionArn']}") + return response + + except Exception as e: + error_message = f"Failed to execute Step Functions orchestration: {str(e)}" + logger.error(error_message, exc_info=True) + raise HTTPException(status_code=500, detail=error_message) + + def _load_experimental_config(self, execution_id: str) -> Dict[str, Any]: + """Load experimental configuration for execution.""" + # This is a placeholder - you would implement actual database loading here + # For now, return a basic config + return { + 'execution_id': execution_id, + 'needs_indexing': True, + 'needs_evaluation': True + } diff --git a/app/adapters/postgres_adapter.py b/app/adapters/postgres_adapter.py new file mode 100644 index 00000000..22ae4390 --- /dev/null +++ b/app/adapters/postgres_adapter.py @@ -0,0 +1,483 @@ +""" +PostgreSQL adapter implementing the DBStorage interface for database-agnostic operations. +""" +import os +import logging +from typing import Dict, Any, List, Optional +import uuid +from sqlalchemy.orm import Session +from sqlalchemy import text +from app.database.connection import db_manager +from app.database.models import Experiment, Execution, QuestionMetrics, ModelInvocations + +logger = logging.getLogger(__name__) + +class PostgresAdapter: + """ + PostgreSQL adapter that implements database operations compatible with the existing DBStorage interface. + """ + + def __init__(self, table_name: str, **kwargs): + # Normalize legacy logical names (flotorch_*) to app model names + self.table_name = self._normalize_table_name(table_name) + self.logger = logging.getLogger(f"{__name__}.{table_name}") + + # Map table names to model classes + self.model_map = { + "experiments": Experiment, + "executions": Execution, + "question_metrics": QuestionMetrics, + "model_invocations": ModelInvocations + } + + if self.table_name not in self.model_map: + raise ValueError(f"Unknown table name: {self.table_name}") + + self.model_class = self.model_map[self.table_name] + + def _normalize_table_name(self, name: str) -> str: + mapping = { + "flotorch_experiment": "experiments", + "flotorch_execution": "executions", + "flotorch_question_metrics": "question_metrics", + "flotorch_model_invocations": "model_invocations", + } + return mapping.get(name, name) + + def _get_session(self) -> Session: + """Get database session.""" + return db_manager.SessionLocal() + + def create_item(self, item_data: Dict[str, Any]) -> Dict[str, Any]: + """Create a new item in the database.""" + try: + # Legacy compatibility: support simple write() shape used by routes + if self.table_name == "executions": + return self._write_execution(item_data) + if self.table_name == "experiments": + return self._write_experiment(item_data) + + with db_manager.get_session() as session: + model_instance = self.model_class(**item_data) + session.add(model_instance) + session.flush() + + return self._model_to_dict(model_instance) + except Exception as e: + self.logger.error(f"Failed to create item in {self.table_name}: {e}") + raise + + def get_item(self, key: str, value: Any) -> Optional[Dict[str, Any]]: + """Get an item by key-value pair.""" + try: + if self.table_name == "executions" and key == "id": + items = self.read({"id": value}) + return items[0] if items else None + if self.table_name == "experiments" and key == "id": + items = self.read({"id": value}) + return items[0] if items else None + + with db_manager.get_session() as session: + query = session.query(self.model_class) + if hasattr(self.model_class, key): + query = query.filter(getattr(self.model_class, key) == value) + else: + self.logger.warning(f"Key {key} not found in {self.table_name}") + return None + + result = query.first() + if result: + return self._model_to_dict(result) + return None + except Exception as e: + self.logger.error(f"Failed to get item from {self.table_name}: {e}") + raise + + def update_item(self, key: str, value: Any, update_data: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Update an item by key-value pair.""" + try: + if self.table_name == "executions" and key == "id": + return self._update_execution_by_legacy_id(value, update_data) + if self.table_name == "experiments" and key == "id": + return self._update_experiment_by_legacy_id(value, update_data) + + with db_manager.get_session() as session: + query = session.query(self.model_class) + if hasattr(self.model_class, key): + query = query.filter(getattr(self.model_class, key) == value) + else: + self.logger.warning(f"Key {key} not found in {self.table_name}") + return None + + result = query.first() + if result: + for field, new_value in update_data.items(): + if hasattr(result, field): + setattr(result, field, new_value) + + session.flush() + return self._model_to_dict(result) + return None + except Exception as e: + self.logger.error(f"Failed to update item in {self.table_name}: {e}") + raise + + def delete_item(self, key: str, value: Any) -> bool: + """Delete an item by key-value pair.""" + try: + with db_manager.get_session() as session: + query = session.query(self.model_class) + if hasattr(self.model_class, key): + query = query.filter(getattr(self.model_class, key) == value) + else: + self.logger.warning(f"Key {key} not found in {self.table_name}") + return False + + result = query.first() + if result: + session.delete(result) + return True + return False + except Exception as e: + self.logger.error(f"Failed to delete item from {self.table_name}: {e}") + raise + + def query_items(self, filters: Optional[Dict[str, Any]] = None, + limit: Optional[int] = None, + offset: Optional[int] = None) -> List[Dict[str, Any]]: + """Query items with optional filters, limit, and offset.""" + try: + if self.table_name == "executions": + return self._read_executions(filters) + if self.table_name == "experiments": + return self._read_experiments(filters) + + with db_manager.get_session() as session: + query = session.query(self.model_class) + + if filters: + for field, value in filters.items(): + if hasattr(self.model_class, field): + query = query.filter(getattr(self.model_class, field) == value) + + if limit: + query = query.limit(limit) + if offset: + query = query.offset(offset) + + results = query.all() + return [self._model_to_dict(result) for result in results] + except Exception as e: + self.logger.error(f"Failed to query items from {self.table_name}: {e}") + raise + + def _model_to_dict(self, model_instance) -> Dict[str, Any]: + """Convert SQLAlchemy model instance to dictionary.""" + result = {} + for column in self.model_class.__table__.columns: + value = getattr(model_instance, column.name) + # Handle datetime and UUID serialization + if hasattr(value, 'isoformat'): # datetime + value = value.isoformat() + elif hasattr(value, 'hex'): # UUID + value = str(value) + result[column.name] = value + return result + + # --- Legacy DBStorage compatibility layer --- + + # Expected by routes: write(), read(filters_dict | None), update(key_dict, data_dict) + def write(self, item_data: Dict[str, Any]) -> Dict[str, Any]: + if self.table_name == "executions": + return self._write_execution(item_data) + if self.table_name == "experiments": + return self._write_experiment(item_data) + return self.create_item(item_data) + + def read(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: + if self.table_name == "executions": + return self._read_executions(filters) + if self.table_name == "experiments": + return self._read_experiments(filters) + return self.query_items(filters) + + def update(self, key: Dict[str, Any], data: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if self.table_name == "executions" and "id" in key: + return self._update_execution_by_legacy_id(key["id"], data) + if self.table_name == "experiments" and "id" in key: + return self._update_experiment_by_legacy_id(key["id"], data) + # Fallback: single-key update if matches a column + if len(key) == 1: + k, v = next(iter(key.items())) + return self.update_item(k, v, data) + self.logger.warning("Complex key updates not supported for this table") + return None + + def delete(self, key: Dict[str, Any]) -> bool: + """Legacy delete interface accepting a key dict (e.g., {"id": legacy_id}).""" + try: + if self.table_name == "experiments" and "id" in key: + legacy_id = str(key["id"]) + with db_manager.get_session() as session: + obj = session.query(Experiment).filter(text("metadata->>'id' = :legacy_id")).params(legacy_id=legacy_id).first() + if obj: + session.delete(obj) + return True + # Fallback: try by UUID column if it was a real UUID id + try: + obj = session.query(Experiment).filter(Experiment.experiment_id == legacy_id).first() + if obj: + session.delete(obj) + return True + except Exception: + pass + return False + if self.table_name == "executions" and "id" in key: + legacy_id = str(key["id"]) + with db_manager.get_session() as session: + obj = session.query(Execution).filter(text("metadata->>'id' = :legacy_id")).params(legacy_id=legacy_id).first() + if obj: + session.delete(obj) + return True + try: + obj = session.query(Execution).filter(Execution.execution_id == legacy_id).first() + if obj: + session.delete(obj) + return True + except Exception: + pass + return False + + # Fallback: single-key delete via mapped column + if len(key) == 1: + k, v = next(iter(key.items())) + return self.delete_item(k, v) + + self.logger.warning("Complex key deletes not supported for this table") + return False + except Exception as e: + self.logger.error(f"Failed to delete item from {self.table_name}: {e}") + raise + + # --- Execution table legacy helpers --- + def _write_execution(self, item_data: Dict[str, Any]) -> Dict[str, Any]: + """ + Store legacy execution payload into PostgreSQL 'executions' table. + - Uses metadata_json to store opaque fields: id, config, gt_data, kb_data, region, name + - Stores status into the 'status' column when provided + - Stores name into 'execution_name' when provided + Returns a dict compatible with existing routes expectations. + """ + with db_manager.get_session() as session: + # Try to find existing by legacy id in metadata_json + legacy_id = item_data.get("id") + existing = None + try: + existing = session.query(Execution).filter(Execution.metadata_json["id"].astext == str(legacy_id)).first() if legacy_id else None + except Exception: + existing = None + + if existing is None: + model_instance = Execution() + else: + model_instance = existing + + # Map fields + if "name" in item_data: + model_instance.execution_name = item_data.get("name") + if "status" in item_data: + model_instance.status = item_data.get("status") + # Ensure required FK-like field exists (nullable in practice for legacy flow) + if getattr(model_instance, "experiment_id", None) is None: + # Generate a placeholder UUID; experiments are created later in the legacy flow + model_instance.experiment_id = uuid.uuid4() + + # Merge metadata + metadata = dict(model_instance.metadata_json or {}) + for key in ["id", "config", "gt_data", "kb_data", "region", "name"]: + if key in item_data: + metadata[key] = item_data[key] + model_instance.metadata_json = metadata + + session.add(model_instance) + session.flush() + + return self._execution_to_legacy_dict(model_instance) + + def _read_executions(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: + with db_manager.get_session() as session: + results = session.query(Execution).all() + + def matches_filters(obj: Execution) -> bool: + if not filters: + return True + # Support filtering by status (column) and by legacy fields in metadata + for k, v in filters.items(): + if k == "status" and getattr(obj, "status", None) == v: + continue + # Check metadata_json + md = obj.metadata_json or {} + if md.get(k) == v: + continue + return False + return True + + filtered = [self._execution_to_legacy_dict(row) for row in results if matches_filters(row)] + return filtered + + def _update_execution_by_legacy_id(self, legacy_id: Any, update_data: Dict[str, Any]) -> Optional[Dict[str, Any]]: + with db_manager.get_session() as session: + try: + obj = session.query(Execution).filter(Execution.metadata_json["id"].astext == str(legacy_id)).first() + except Exception: + obj = None + if not obj: + return None + + # Update mapped columns when relevant + if "status" in update_data: + obj.status = update_data["status"] + + # Update metadata + metadata = dict(obj.metadata_json or {}) + metadata.update(update_data) + obj.metadata_json = metadata + + session.flush() + return self._execution_to_legacy_dict(obj) + + def _execution_to_legacy_dict(self, obj: Execution) -> Dict[str, Any]: + md = obj.metadata_json or {} + return { + "id": md.get("id") or (str(obj.execution_id) if hasattr(obj, "execution_id") else None), + "date": (obj.created_at.isoformat() if hasattr(obj, "created_at") and obj.created_at else None), + "status": obj.status, + "validation_status": md.get("validation_status"), + "gt_data": md.get("gt_data", ""), + "kb_data": md.get("kb_data", ""), + "region": md.get("region", ""), + "config": md.get("config"), + "name": md.get("name") or getattr(obj, "execution_name", ""), + } + + # --- Experiment table legacy helpers --- + def _write_experiment(self, item_data: Dict[str, Any]) -> Dict[str, Any]: + with db_manager.get_session() as session: + legacy_id = item_data.get("id") + existing = None + try: + # Use JSONB operator for reliable lookup + existing = session.query(Experiment).filter(text("metadata->>'id' = :legacy_id")).params(legacy_id=str(legacy_id)).first() if legacy_id else None + except Exception: + existing = None + + if existing is None: + model_instance = Experiment() + else: + model_instance = existing + + # Map simple columns + if "experiment_status" in item_data: + model_instance.status = item_data.get("experiment_status") + # Derive a name for visibility if available + name_candidate = ( + (item_data.get("config") or {}).get("name") + if isinstance(item_data.get("config"), dict) else None + ) or item_data.get("index_id") or "" + if name_candidate: + model_instance.experiment_name = name_candidate[:255] + + # Merge metadata + metadata = dict(model_instance.metadata_json or {}) + for key in [ + "id", + "execution_id", + "config", + "index_id", + "experiment_status", + "index_status", + "retrieval_status", + "eval_status", + ]: + if key in item_data: + metadata[key] = item_data[key] + model_instance.metadata_json = metadata + + session.add(model_instance) + session.flush() + return self._experiment_to_legacy_dict(model_instance) + + def _read_experiments(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: + with db_manager.get_session() as session: + results = session.query(Experiment).all() + + def matches_filters(obj: Experiment) -> bool: + if not filters: + return True + md = obj.metadata_json or {} + for k, v in filters.items(): + if k == "experiment_status" and getattr(obj, "status", None) == v: + continue + if k == "id" and (md.get("id") == v or str(getattr(obj, "experiment_id", "")) == str(v)): + continue + if md.get(k) == v: + continue + return False + return True + + return [self._experiment_to_legacy_dict(row) for row in results if matches_filters(row)] + + def _update_experiment_by_legacy_id(self, legacy_id: Any, update_data: Dict[str, Any]) -> Optional[Dict[str, Any]]: + with db_manager.get_session() as session: + try: + obj = session.query(Experiment).filter(text("metadata->>'id' = :legacy_id")).params(legacy_id=str(legacy_id)).first() + except Exception: + obj = None + if not obj: + return None + + if "experiment_status" in update_data: + obj.status = update_data["experiment_status"] + + metadata = dict(obj.metadata_json or {}) + metadata.update(update_data) + obj.metadata_json = metadata + + session.flush() + return self._experiment_to_legacy_dict(obj) + + def _experiment_to_legacy_dict(self, obj: Experiment) -> Dict[str, Any]: + md = obj.metadata_json or {} + return { + "id": md.get("id") or (str(obj.experiment_id) if hasattr(obj, "experiment_id") else None), + "experiment_status": obj.status, + "index_status": md.get("index_status", "not_started"), + "retrieval_status": md.get("retrieval_status", "not_started"), + "eval_status": md.get("eval_status", "not_started"), + "execution_id": md.get("execution_id"), + "index_id": md.get("index_id"), + "config": md.get("config"), + } + + def batch_create_items(self, items_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Create multiple items in a batch.""" + try: + with db_manager.get_session() as session: + model_instances = [self.model_class(**item_data) for item_data in items_data] + session.add_all(model_instances) + session.flush() + + return [self._model_to_dict(instance) for instance in model_instances] + except Exception as e: + self.logger.error(f"Failed to batch create items in {self.table_name}: {e}") + raise + + def execute_raw_query(self, query: str, params: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: + """Execute raw SQL query.""" + try: + with db_manager.get_session() as session: + result = session.execute(text(query), params or {}) + return [dict(row._mapping) for row in result] + except Exception as e: + self.logger.error(f"Failed to execute raw query on {self.table_name}: {e}") + raise diff --git a/app/adapters/retriever_adapter.py b/app/adapters/retriever_adapter.py new file mode 100644 index 00000000..4f8a7e97 --- /dev/null +++ b/app/adapters/retriever_adapter.py @@ -0,0 +1,84 @@ +""" +Retriever service adapter with feature flag to switch between local and external service. +""" +import os +import logging +from typing import Dict, Any, List, Optional +from app.adapters.http_service_adapter import HTTPServiceAdapter, ServiceConfig + +logger = logging.getLogger(__name__) + +class RetrieverAdapter: + """ + Adapter that can use either local retriever implementation or external HTTP service. + Controlled by USE_EXTERNAL_SERVICES environment variable. + """ + + def __init__(self, config: Dict[str, Any], experimental_config: Dict[str, Any]): + self.use_external = os.getenv("USE_EXTERNAL_SERVICES", "false").lower() == "true" + + if self.use_external: + logger.info("Using external retriever service") + self._init_external_service() + else: + logger.info("Using local retriever implementation") + self._init_local() + + def _init_external_service(self): + """Initialize external retriever service.""" + retriever_url = os.getenv("RETRIEVER_SERVICE_URL", "http://localhost:8001") + service_config = ServiceConfig( + base_url=retriever_url, + timeout=int(os.getenv("RETRIEVER_TIMEOUT", "60")), + max_retries=int(os.getenv("RETRIEVER_MAX_RETRIES", "3")) + ) + self.client = HTTPServiceAdapter(service_config) + + def _init_local(self): + """Initialize local retriever implementation.""" + # Import local retriever function + from retriever.retriever import retrieve + self.retrieve_func = retrieve + + def retrieve(self, config: Any, experimental_config: Any) -> None: + """ + Execute retrieval process. + + Args: + config: Global configuration object + experimental_config: Experiment-specific configuration + """ + if self.use_external: + return self._retrieve_external(config, experimental_config) + else: + return self._retrieve_local(config, experimental_config) + + def _retrieve_external(self, config: Any, experimental_config: Any) -> None: + """Execute retrieval using external service.""" + try: + # Convert config objects to dictionaries for JSON serialization + config_dict = self._config_to_dict(config) + experimental_config_dict = self._config_to_dict(experimental_config) + + payload = { + "config": config_dict, + "experimental_config": experimental_config_dict + } + + response = self.client.post("/retrieve", json=payload) + logger.info(f"External retriever response: {response}") + + except Exception as e: + logger.error(f"External retriever failed: {e}") + raise + + def _retrieve_local(self, config: Any, experimental_config: Any) -> None: + """Execute retrieval using local implementation.""" + return self.retrieve_func(config, experimental_config) + + def _config_to_dict(self, config_obj: Any) -> Dict[str, Any]: + """Convert config object to dictionary for JSON serialization.""" + if hasattr(config_obj, '__dict__'): + return {k: v for k, v in config_obj.__dict__.items() + if not k.startswith('_')} + return config_obj diff --git a/app/utils/__init__.py b/app/utils/__init__.py new file mode 100644 index 00000000..71bc66f9 --- /dev/null +++ b/app/utils/__init__.py @@ -0,0 +1,9 @@ +""" +FloTorch utilities package. +""" +from .legacy_cleanup import LegacyCleanup + +__all__ = [ + 'LegacyCleanup' +] + diff --git a/app/validation.py b/app/validation.py new file mode 100644 index 00000000..c23167b7 --- /dev/null +++ b/app/validation.py @@ -0,0 +1,78 @@ +import logging +import os + +from typing import List + +from config.config import get_config + + +logger = logging.getLogger(__name__) + + +def log_core_version() -> None: + """Log the installed FloTorch-core version if available. Never raises.""" + try: + try: + # Python 3.8+ + from importlib.metadata import version, PackageNotFoundError # type: ignore + except Exception: # pragma: no cover + from importlib_metadata import version, PackageNotFoundError # type: ignore + + dist_names = ["FloTorch-core", "flotorch-core", "flotorch_core"] + for name in dist_names: + try: + v = version(name) + logger.info(f"FloTorch-core detected ({name}) version: {v}") + return + except PackageNotFoundError: + continue + logger.warning("FloTorch-core package not detected. Some features may rely on it.") + except Exception as exc: # pragma: no cover + logger.warning(f"Unable to determine FloTorch-core version: {exc}") + + +def validate_base_config() -> List[str]: + """ + Perform lightweight configuration validation. Returns a list of warnings. + This must never raise to avoid behavior changes in existing environments. + """ + warnings: List[str] = [] + try: + cfg = get_config() + + if not (cfg.aws_region and isinstance(cfg.aws_region, str)): + warnings.append("aws_region is not set (config.aws_region).") + + if not (cfg.s3_bucket and isinstance(cfg.s3_bucket, str)): + warnings.append("s3_bucket is not set (config.s3_bucket).") + + # OpenSearch basic presence check (endpoint env is used in util.open_search_config_utils) + if not os.getenv("OPENSEARCH_ENDPOINT") and not cfg.opensearch_host: + warnings.append("OpenSearch endpoint is not configured (OPENSEARCH_ENDPOINT/opensearch_host).") + + # Planned service adapters + for key in ("INDEXER_URL", "RETRIEVER_URL", "EVAL_URL"): + if not os.getenv(key): + warnings.append(f"Optional external service URL not set: {key} (ok to ignore if unused).") + + # Database selection + db_type = os.getenv("DB_TYPE") + if not db_type: + warnings.append("DB_TYPE is not set; default routing may not be explicit.") + + except Exception as exc: # pragma: no cover + warnings.append(f"Validation encountered an error: {exc}") + + return warnings + + +def log_validation_warnings(warnings: List[str]) -> None: + if not warnings: + logger.info("Configuration validation passed with no warnings.") + return + logger.warning("Configuration validation warnings:") + for msg in warnings: + logger.warning(f" - {msg}") + + + diff --git a/baseclasses/base_classes.py b/baseclasses/base_classes.py index 11337d2c..53b387dc 100644 --- a/baseclasses/base_classes.py +++ b/baseclasses/base_classes.py @@ -11,7 +11,7 @@ from dataclasses import dataclass from config.config import Config from config.experimental_config import ExperimentalConfig, NShotPromptGuide -from core.dynamodb import DynamoDBOperations +# Removed local core dependency; functionality now provided via adapters/flotorch_core import random from dataclasses import dataclass, asdict from decimal import Decimal