Merge pull request #537 from MervinPraison/claude/issue-335-20250529_033521

fix: Add FORCE_SQLITE environment variable to bypass production database detection

Author: MervinPraison

src/praisonai/praisonai/ui/database_config.py

@@ -0,0 +1,56 @@
+"""
+Database configuration utilities for PraisonAI UI components.
+
+This module provides centralized database configuration functionality,
+particularly for handling the FORCE_SQLITE environment variable.
+"""
+
+import os
+from typing import Optional, Tuple
+
+
+def should_force_sqlite() -> bool:
+    """
+    Check if FORCE_SQLITE environment variable is set to true.
+    
+    Returns:
+        bool: True if FORCE_SQLITE is set to "true" (case-insensitive), False otherwise.
+    """
+    return os.getenv("FORCE_SQLITE", "false").lower() == "true"
+
+
+def get_database_url_with_sqlite_override() -> Optional[str]:
+    """
+    Get database URL respecting FORCE_SQLITE flag.
+    
+    When FORCE_SQLITE=true, this function returns None to force SQLite usage.
+    Otherwise, it returns the appropriate database URL from environment variables.
+    
+    Returns:
+        Optional[str]: Database URL if external database should be used, None for SQLite.
+    """
+    if should_force_sqlite():
+        return None
+    
+    database_url = os.getenv("DATABASE_URL")
+    supabase_url = os.getenv("SUPABASE_DATABASE_URL")
+    return supabase_url if supabase_url else database_url
+
+
+def get_database_config_for_sqlalchemy() -> Tuple[Optional[str], Optional[str]]:
+    """
+    Get database configuration for SQLAlchemy module respecting FORCE_SQLITE flag.
+    
+    Returns:
+        Tuple[Optional[str], Optional[str]]: (DATABASE_URL, SUPABASE_DATABASE_URL)
+            Both will be None if FORCE_SQLITE=true, otherwise original values.
+    """
+    if should_force_sqlite():
+        return None, None
+    else:
+        database_url = os.getenv("DATABASE_URL")
+        supabase_url = os.getenv("SUPABASE_DATABASE_URL")
+        # Apply Supabase override logic
+        if supabase_url:
+            database_url = supabase_url
+        return database_url, supabase_url

src/praisonai/praisonai/ui/db.py

@@ -9,6 +9,7 @@
 from sql_alchemy import SQLAlchemyDataLayer
 import chainlit.data as cl_data
 from chainlit.types import ThreadDict
+from database_config import get_database_url_with_sqlite_override
 
 def ensure_directories():
     """Ensure required directories exist"""
@@ -57,10 +58,8 @@ def ensure_directories():
 
 class DatabaseManager(SQLAlchemyDataLayer):
     def __init__(self):
-        self.database_url = os.getenv("DATABASE_URL")
-        supabase_url = os.getenv("SUPABASE_DATABASE_URL")
-        if supabase_url:
-            self.database_url = supabase_url
+        # Check FORCE_SQLITE flag to bypass external database detection
+        self.database_url = get_database_url_with_sqlite_override()
 
         if self.database_url:
             self.conninfo = self.database_url

src/praisonai/praisonai/ui/sql_alchemy.py

@@ -29,16 +29,14 @@
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
 from sqlalchemy.orm import sessionmaker
+from database_config import get_database_config_for_sqlalchemy
 
 if TYPE_CHECKING:
     from chainlit.element import Element
     from chainlit.step import StepDict
 
-DATABASE_URL = os.getenv("DATABASE_URL")
-SUPABASE_DATABASE_URL = os.getenv("SUPABASE_DATABASE_URL")
-if SUPABASE_DATABASE_URL:
-    # If a Supabase database URL is provided, use it.
-    DATABASE_URL = SUPABASE_DATABASE_URL
+# Check FORCE_SQLITE flag to bypass external database detection
+DATABASE_URL, SUPABASE_DATABASE_URL = get_database_config_for_sqlalchemy()
 
 class SQLAlchemyDataLayer(BaseDataLayer):
     def __init__(

src/praisonai/tests/unit/test_database_config.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+"""
+Test script for database_config module functionality.
+"""
+
+import sys
+import os
+sys.path.insert(0, 'src/praisonai/praisonai/ui')
+
+# Test the database_config module
+from database_config import should_force_sqlite, get_database_url_with_sqlite_override, get_database_config_for_sqlalchemy
+
+def test_database_config():
+    print("Testing database_config utilities...")
+    
+    # Test 1: FORCE_SQLITE=false (default)
+    print('\nTest 1: Default behavior (FORCE_SQLITE not set)')
+    os.environ.pop('FORCE_SQLITE', None)
+    os.environ.pop('DATABASE_URL', None) 
+    os.environ.pop('SUPABASE_DATABASE_URL', None)
+    print(f'should_force_sqlite(): {should_force_sqlite()}')
+    print(f'get_database_url_with_sqlite_override(): {get_database_url_with_sqlite_override()}')
+    print(f'get_database_config_for_sqlalchemy(): {get_database_config_for_sqlalchemy()}')
+    
+    # Test 2: FORCE_SQLITE=true
+    print('\nTest 2: FORCE_SQLITE=true')
+    os.environ['FORCE_SQLITE'] = 'true'
+    os.environ['DATABASE_URL'] = 'postgres://test'
+    os.environ['SUPABASE_DATABASE_URL'] = 'postgres://supabase'
+    print(f'should_force_sqlite(): {should_force_sqlite()}')
+    print(f'get_database_url_with_sqlite_override(): {get_database_url_with_sqlite_override()}')
+    print(f'get_database_config_for_sqlalchemy(): {get_database_config_for_sqlalchemy()}')
+    
+    # Test 3: FORCE_SQLITE=false with DATABASE_URL
+    print('\nTest 3: FORCE_SQLITE=false with DATABASE_URL')
+    os.environ['FORCE_SQLITE'] = 'false'
+    print(f'should_force_sqlite(): {should_force_sqlite()}')
+    print(f'get_database_url_with_sqlite_override(): {get_database_url_with_sqlite_override()}')
+    print(f'get_database_config_for_sqlalchemy(): {get_database_config_for_sqlalchemy()}')
+    
+    print('\n✅ All tests completed - utility functions work correctly!')
+
+if __name__ == "__main__":
+    test_database_config()