diff --git a/app/dependencies/database.py b/app/dependencies/database.py index 2a844f1..6a1cee4 100644 --- a/app/dependencies/database.py +++ b/app/dependencies/database.py @@ -8,6 +8,10 @@ from flotorch_core.config.env_config_provider import EnvConfigProvider from flotorch_core.config.config_provider import ConfigProvider from flotorch_core.config.config import Config +from app.adapters.postgres_adapter import PostgresAdapter +from app.adapters.dual_write_adapter import DualWriteAdapter +import os +from app.common.logger import logger env_config_provider = EnvConfigProvider() core_config = Config(env_config_provider) @@ -31,21 +35,38 @@ def get_step_function_orchestrator() -> StepFunctionOrchestrator: class DBClientFactory(): def create_table_client(self, db_type: str, table_name: str) -> DBStorage: - cache_name = f"{db_type}_{table_name}" if table_name else db_name + + # Check if dual write is enabled + if os.getenv("ENABLE_DUAL_WRITES", "false").lower() == "true": + print(f"Using dual write adapter for {table_name}") + return DualWriteAdapter(table_name=table_name, aws_region=core_config.get_region()) + if db_type == "DYNAMODB": - return DynamoDB( - table_name=table_name, - region_name=core_config.get_region() - ) + try: + return DynamoDB( + table_name=table_name, + region_name=core_config.get_region() + ) + except Exception as e: + print(f"Failed to create DynamoDB client: {e}") + print("Falling back to PostgreSQL adapter") + return PostgresAdapter(table_name=table_name) elif db_type == "POSTGRESDB": - return PostgresDB( - dbname=core_config.get_postgres_db(), - user=core_config.get_postgres_user(), - password=core_config.get_postgres_password(), - table_name=table_name, - host=core_config.get_postgres_host(), - port=core_config.get_postgres_port() - ) + try: + return PostgresDB( + dbname=core_config.get_postgres_db(), + user=core_config.get_postgres_user(), + password=core_config.get_postgres_password(), + table_name=table_name, + host=core_config.get_postgres_host(), + port=core_config.get_postgres_port() + ) + except Exception as e: + print(f"Failed to create flotorch_core PostgresDB client: {e}") + print("Falling back to local PostgreSQL adapter") + return PostgresAdapter(table_name=table_name) + else: + raise ValueError(f"Unsupported database type: {db_type}") def get_db_dependency(table_name_func): """ diff --git a/app/main.py b/app/main.py index 696315e..748ced7 100644 --- a/app/main.py +++ b/app/main.py @@ -1,11 +1,14 @@ from fastapi import FastAPI +import os from fastapi.middleware.cors import CORSMiddleware from .seed_data import seed_models +from .validation import log_core_version, validate_base_config, log_validation_warnings -from app.routes import execution, experiment, health, uploads, bedrock_config, config, expert_eval +from app.routes import execution, experiment, health, uploads, bedrock_config, config, expert_eval, migration, cutover, cleanup, opensearch_admin from app.dependencies.database import ( get_execution_model_invocations_db ) +from app.database import init_database def create_app() -> FastAPI: @@ -15,21 +18,37 @@ def create_app() -> FastAPI: @app.on_event("startup") async def startup_event(): try: - db_client_generator = get_execution_model_invocations_db() - db_client = None - try: - db_client = next(db_client_generator) - seeded_count = seed_models(db_client) - except StopIteration: - print("Generator did not yield a DB client during startup.") - except Exception as e: - print(f"Error during seeding process: {e}") - finally: - if db_client_generator: - try: - db_client_generator.close() - except Exception as e: - print(f"Error closing DB client generator during startup: {e}") + # Observability-only: core version + config validation (warnings only) + log_core_version() + warnings = validate_base_config() + # In LOCAL_DEV, always just warn. In other envs, still warn only to avoid behavior changes. + log_validation_warnings(warnings) + + # Initialize PostgreSQL database if enabled + if os.getenv("DB_TYPE", "DYNAMODB").upper() == "POSTGRESDB": + try: + init_database() + print("PostgreSQL database initialized successfully") + except Exception as e: + print(f"PostgreSQL initialization failed: {e}") + # Continue startup even if PostgreSQL init fails + + if os.getenv("SKIP_SEEDING", "false").lower() != "true": + db_client_generator = get_execution_model_invocations_db() + db_client = None + try: + db_client = next(db_client_generator) + seeded_count = seed_models(db_client) + except StopIteration: + print("Generator did not yield a DB client during startup.") + except Exception as e: + print(f"Error during seeding process: {e}") + finally: + if db_client_generator: + try: + db_client_generator.close() + except Exception as e: + print(f"Error closing DB client generator during startup: {e}") print("Application startup tasks finished.") @@ -45,14 +64,18 @@ async def startup_event(): allow_headers=["*"], ) - # Register routers - app.include_router(uploads.router) - app.include_router(execution.router) - app.include_router(experiment.router) - app.include_router(health.router) - app.include_router(bedrock_config.router) - app.include_router(config.router) - app.include_router(expert_eval.router) + # Register routers with /api prefix + app.include_router(uploads.router, prefix="/api") + app.include_router(execution.router, prefix="/api") + app.include_router(experiment.router, prefix="/api") + app.include_router(health.router, prefix="/api") + app.include_router(bedrock_config.router, prefix="/api") + app.include_router(config.router, prefix="/api") + app.include_router(expert_eval.router, prefix="/api") + app.include_router(migration.router, prefix="/api") + app.include_router(cutover.router, prefix="/api") + app.include_router(cleanup.router, prefix="/api") + app.include_router(opensearch_admin.router, prefix="/api") return app diff --git a/app/orchestrator.py b/app/orchestrator.py index 5fbd587..d980182 100644 --- a/app/orchestrator.py +++ b/app/orchestrator.py @@ -6,6 +6,7 @@ from config.config import get_config import logging from typing import Dict, Any +from app.adapters.orchestrator_adapter import OrchestratorAdapter # Configure logging logging.basicConfig(level=logging.INFO) @@ -17,7 +18,7 @@ class StepFunctionOrchestrator: """ def __init__(self): self.config = get_config() - self.step_function_client = self._initialize_step_function_client() + self.orchestrator_adapter = OrchestratorAdapter() def _initialize_step_function_client(self) -> boto3.client: """ @@ -68,23 +69,7 @@ def run_experiment_orchestration(self, execution_id: str) -> Dict[str, Any]: Raises: HTTPException: If orchestration fails """ - try: - # Prepare the payload - payload = self._prepare_execution_payload(execution_id) - - # Start the Step Function execution - response = self.step_function_client.start_execution( - stateMachineArn=self.config.step_function_arn, - input=payload - ) - - logger.info(f"Started Step Function with Execution ARN: {response['executionArn']}") - return response - - except Exception as e: - error_message = f"Failed to execute orchestration: {str(e)}" - logger.error(error_message, exc_info=True) - raise HTTPException(status_code=500, detail=error_message) + return self.orchestrator_adapter.run_experiment_orchestration(execution_id) # Create a singleton instance orchestrator = StepFunctionOrchestrator()