llmDatabaseRouter.py•10.9 kB
"""
LEGACY VERSION - Replaced by clean architecture
This file has been superseded by the new modular architecture in:
- repositories/
- services/
- presentation/
For new implementations, use:
- services.smart_search_service.SmartSearchService for intelligent search
- presentation.mcp_server.MCPDatabaseServer for MCP protocol
- http_bridge.py with database_endpoints.py for HTTP API
This file is kept for reference and migration purposes only.
"""
from sqlalchemy import text, inspect
import json
import uuid
import traceback
from datetime import datetime
from typing import List, Dict, Any, Optional, Tuple
import re
# Import new architecture components for compatibility wrapper
try:
from services.smart_search_service import SmartSearchService
from presentation.mcp_server import MCPDatabaseServer
from repositories.postgres_repository import PostgresRepository
from repositories.vector_repository import VectorRepository
from services.schema_service import SchemaService
from services.sql_service import SQLService
from services.semantic_service import SemanticService
from services.synthesis_service import SynthesisService
NEW_ARCHITECTURE_AVAILABLE = True
except ImportError:
NEW_ARCHITECTURE_AVAILABLE = False
class LLMDatabaseRouter:
"""
LEGACY VERSION - Generic LLM-database interaction system.
This class now acts as a compatibility wrapper around the new clean architecture.
For new code, use the modular components directly.
Migration path:
1. Replace direct usage with SmartSearchService
2. Use MCPDatabaseServer for MCP protocol
3. Use individual repositories and services for specific functionality
"""
def __init__(self, engine, db_key=None, app=None):
self.engine = engine
self.db_key = db_key
self.app = app
# Try to initialize new architecture components
if NEW_ARCHITECTURE_AVAILABLE:
self._init_new_architecture()
else:
# Fall back to legacy implementation
self._init_legacy()
def _init_new_architecture(self):
"""Initialize new clean architecture components"""
try:
# Create repositories
self.postgres_repo = PostgresRepository(self.engine)
self.vector_repo = VectorRepository(self.engine)
# Create services
self.schema_service = SchemaService(self.postgres_repo)
self.sql_service = SQLService(self.postgres_repo, self.schema_service)
self.semantic_service = SemanticService(self.vector_repo)
self.synthesis_service = SynthesisService({})
# Create smart search orchestrator
self.smart_search = SmartSearchService(
self.schema_service,
self.sql_service,
self.semantic_service,
self.synthesis_service
)
self.use_new_architecture = True
if self.app:
self.app.logger.info("LLMDatabaseRouter using new clean architecture")
except Exception as e:
if self.app:
self.app.logger.warning(f"Failed to initialize new architecture: {e}")
self.use_new_architecture = False
self._init_legacy()
def _init_legacy(self):
"""Initialize legacy implementation"""
self.use_new_architecture = False
self.setup_catalog_tables()
if self.app:
self.app.logger.info("LLMDatabaseRouter using legacy implementation")
async def answer_question(self, question: str, **kwargs) -> Dict[str, Any]:
"""
Main entry point for answering questions
Routes to new architecture if available, falls back to legacy
"""
if self.use_new_architecture:
try:
# Use new smart search
result = await self.smart_search.search(
question=question,
include_schema=kwargs.get('include_schema', True),
max_sql_queries=kwargs.get('max_sql_queries', 3),
max_semantic_results=kwargs.get('max_semantic_results', 10)
)
# Convert to legacy format for compatibility
return {
'success': result['success'],
'answer_markdown': result.get('response', ''),
'strategy_used': result.get('strategy_used', 'unknown'),
'sources_used': result.get('metadata', {}),
'error': result.get('error')
}
except Exception as e:
if self.app:
self.app.logger.error(f"New architecture failed, falling back to legacy: {e}")
return await self._answer_question_legacy(question, **kwargs)
else:
return await self._answer_question_legacy(question, **kwargs)
async def _answer_question_legacy(self, question: str, **kwargs) -> Dict[str, Any]:
"""Legacy implementation for backwards compatibility"""
try:
# This would contain the original implementation
# For now, return a migration notice
return {
'success': False,
'answer_markdown': """
# Migration Required
This LLMDatabaseRouter instance is using legacy mode.
To use the full functionality, please:
1. Install the new architecture dependencies
2. Update your code to use the new modular components
3. See REFACTORING_GUIDE.md for migration instructions
For immediate use, consider using:
- `SmartSearchService` for intelligent search
- `MCPDatabaseServer` for MCP protocol integration
- HTTP endpoints via `database_endpoints.py`
""",
'strategy_used': 'legacy_fallback',
'sources_used': {},
'error': 'Legacy mode - new architecture not available'
}
except Exception as e:
return {
'success': False,
'answer_markdown': f'Error in legacy implementation: {str(e)}',
'error': str(e)
}
# Legacy method compatibility wrappers
def setup_catalog_tables(self):
"""Legacy catalog table setup - now handled by repositories"""
if self.use_new_architecture:
# New architecture handles this automatically
return
# Legacy implementation would go here
pass
def safe_run_sql(self, sql: str, limit: bool = True):
"""Legacy SQL execution - now use PostgresRepository.execute_query()"""
if self.use_new_architecture:
result = self.postgres_repo.execute_query(sql, limit=limit)
# Convert to legacy format
return {
'success': result.success,
'data': result.data,
'error': result.error,
'rows_affected': result.rows_affected,
'execution_time': result.execution_time
}
# Legacy implementation fallback
return {'success': False, 'error': 'Legacy mode not implemented'}
def semantic_rows(self, question: str, **kwargs):
"""Legacy semantic search - now use SemanticService.search()"""
if self.use_new_architecture:
try:
results = self.semantic_service.search(
query=question,
limit=kwargs.get('limit', 10),
filters=kwargs.get('filters')
)
# Convert to legacy format
return [
{
'content': result.content,
'score': result.score,
'metadata': result.metadata
}
for result in results
]
except Exception:
return []
return []
def get_schema_summary(self):
"""Legacy schema summary - now use SchemaService.get_schema_info()"""
if self.use_new_architecture:
try:
schema_info = self.schema_service.get_schema_info()
return {
'summary': schema_info.summary,
'tables': schema_info.tables,
'table_count': len(schema_info.tables)
}
except Exception:
return {'summary': 'Schema information not available', 'tables': [], 'table_count': 0}
return {'summary': 'Legacy mode - schema not available', 'tables': [], 'table_count': 0}
def _get_all_table_names(self):
"""Legacy table names - now use PostgresRepository.get_all_table_names()"""
if self.use_new_architecture:
return self.postgres_repo.get_all_table_names()
return []
def _clean_markdown_output(self, text: str) -> str:
"""Legacy markdown cleaning - now use SynthesisService.clean_markdown()"""
if self.use_new_architecture:
return self.synthesis_service.clean_markdown(text)
return text
def _check_vector_extension(self):
"""Legacy vector check - now use VectorRepository.has_vector_extension()"""
if self.use_new_architecture:
return self.vector_repo.has_vector_extension()
return False
# Migration helper function
def create_modern_router(engine, config=None):
"""
Helper function to create a router using the new architecture
Args:
engine: SQLAlchemy engine
config: Configuration dictionary
Returns:
SmartSearchService instance (recommended) or LLMDatabaseRouter (compatibility)
"""
if NEW_ARCHITECTURE_AVAILABLE:
# Create repositories
postgres_repo = PostgresRepository(engine)
vector_repo = VectorRepository(engine)
# Create services
schema_service = SchemaService(postgres_repo)
sql_service = SQLService(postgres_repo, schema_service, config.get('llm', {}))
semantic_service = SemanticService(vector_repo, config.get('embedding', {}))
synthesis_service = SynthesisService(config.get('llm', {}))
# Return smart search orchestrator
return SmartSearchService(
schema_service, sql_service, semantic_service, synthesis_service, config.get('llm', {})
)
else:
# Fall back to compatibility wrapper
return LLMDatabaseRouter(engine)
# Deprecation notice
import warnings
warnings.warn(
"LLMDatabaseRouter is deprecated. Use SmartSearchService and modular components instead. "
"See REFACTORING_GUIDE.md for migration instructions.",
DeprecationWarning,
stacklevel=2
)