[review] Orchestrator

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):
     """

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
 

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()