191 lines
8.8 KiB
Python
191 lines
8.8 KiB
Python
# DYNAMIC POSTGRESQL DESIGNER - AI-powered PostgreSQL schema based on actual features
|
|
# Uses Claude AI to generate PostgreSQL database schema based on functional requirements
|
|
|
|
from typing import Dict, Any, List
|
|
from loguru import logger
|
|
from designers.base_designer import BaseDatabaseDesigner
|
|
from prompts.database.postgresql_prompts import PostgreSQLPrompts
|
|
|
|
class PostgreSQLDesigner(BaseDatabaseDesigner):
|
|
"""Dynamic PostgreSQL specialist - Generates database schema based on actual project features"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.prompts = PostgreSQLPrompts()
|
|
logger.info("🗄️ Dynamic PostgreSQL Designer initialized - AI-powered feature-based schema design")
|
|
|
|
def get_technology_name(self) -> str:
|
|
return "PostgreSQL"
|
|
|
|
async def design_architecture(self, context: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Design PostgreSQL database schema dynamically based on actual features and tech stack"""
|
|
try:
|
|
logger.info("🗄️ PostgreSQL Designer analyzing project features...")
|
|
|
|
# Extract real project data
|
|
functional_reqs = context['functional_requirements']
|
|
tech_stack = context['technology_stack']
|
|
business_context = context['business_context']
|
|
|
|
logger.info(f" Feature: {functional_reqs['feature_name']}")
|
|
logger.info(f" Technical Requirements: {len(functional_reqs['technical_requirements'])} items")
|
|
logger.info(f" Business Rules: {len(functional_reqs['business_logic_rules'])} rules")
|
|
|
|
# Generate AI prompt based on actual project requirements
|
|
prompt = self.prompts.create_dynamic_postgresql_prompt(
|
|
feature_name=functional_reqs['feature_name'],
|
|
feature_description=functional_reqs['description'],
|
|
technical_requirements=functional_reqs['technical_requirements'],
|
|
business_logic_rules=functional_reqs['business_logic_rules'],
|
|
complexity_level=functional_reqs['complexity_level'],
|
|
tech_stack=tech_stack,
|
|
all_features=functional_reqs['all_features']
|
|
)
|
|
|
|
# Get AI-generated PostgreSQL architecture
|
|
logger.info("🤖 Generating PostgreSQL schema with AI...")
|
|
response = await self.claude_client.generate_architecture(prompt)
|
|
|
|
if response.get('success'):
|
|
postgresql_architecture = response['data']
|
|
|
|
# Enhance with PostgreSQL-specific features based on requirements
|
|
enhanced_architecture = self._enhance_with_requirements(
|
|
postgresql_architecture, tech_stack, functional_reqs
|
|
)
|
|
|
|
logger.info("✅ Dynamic PostgreSQL schema generated successfully")
|
|
return {
|
|
"success": True,
|
|
"architecture": enhanced_architecture,
|
|
"specialist": "PostgreSQL",
|
|
"version": "PostgreSQL 14+",
|
|
"generated_for_feature": functional_reqs['feature_name'],
|
|
"business_rules_implemented": len(functional_reqs['business_logic_rules']),
|
|
"security_level": self._determine_security_level(functional_reqs),
|
|
"features_used": self._extract_postgresql_features(functional_reqs, tech_stack),
|
|
"ai_generated": True,
|
|
"feature_specific": True
|
|
}
|
|
else:
|
|
logger.warning("AI generation failed, creating feature-based fallback")
|
|
return self._create_feature_based_fallback(functional_reqs, tech_stack)
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ PostgreSQL architecture design failed: {e}")
|
|
return self._create_feature_based_fallback(functional_reqs, tech_stack)
|
|
|
|
async def design_schema(self, context: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Design PostgreSQL schema based on actual features"""
|
|
# Will implement specific schema design if needed
|
|
pass
|
|
|
|
async def design_indexes(self, context: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Design PostgreSQL indexes based on expected query patterns"""
|
|
# Will implement specific index design if needed
|
|
pass
|
|
|
|
async def design_relationships(self, context: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Design table relationships based on business logic"""
|
|
# Will implement specific relationship design if needed
|
|
pass
|
|
|
|
def _enhance_with_requirements(self, architecture: Dict, tech_stack: Dict, functional_reqs: Dict) -> Dict:
|
|
"""Enhance AI-generated architecture with dynamic PostgreSQL features"""
|
|
|
|
# This method enhances the Claude-generated architecture
|
|
# with additional PostgreSQL-specific features based on requirements
|
|
|
|
if 'database_schema' not in architecture:
|
|
architecture['database_schema'] = {}
|
|
|
|
# Add dynamic enhancements based on actual requirements
|
|
architecture['feature_analysis'] = {
|
|
'feature_name': functional_reqs.get('feature_name', 'Unknown'),
|
|
'complexity_level': functional_reqs.get('complexity_level', 'medium'),
|
|
'business_rules_count': len(functional_reqs.get('business_logic_rules', [])),
|
|
'technical_requirements_count': len(functional_reqs.get('technical_requirements', []))
|
|
}
|
|
|
|
return architecture
|
|
|
|
def _determine_security_level(self, functional_reqs: Dict) -> str:
|
|
"""Determine security level based on requirements"""
|
|
technical_reqs = functional_reqs.get('technical_requirements', [])
|
|
business_rules = functional_reqs.get('business_logic_rules', [])
|
|
|
|
security_keywords = ['hipaa', 'gdpr', 'encryption', 'compliance', 'audit', 'security', 'private']
|
|
|
|
security_mentions = 0
|
|
for req in technical_reqs + business_rules:
|
|
if any(keyword in req.lower() for keyword in security_keywords):
|
|
security_mentions += 1
|
|
|
|
if security_mentions >= 3:
|
|
return 'high'
|
|
elif security_mentions >= 1:
|
|
return 'medium'
|
|
else:
|
|
return 'standard'
|
|
|
|
def _extract_postgresql_features(self, functional_reqs: Dict, tech_stack: Dict) -> List[str]:
|
|
"""Extract PostgreSQL features to use based on requirements"""
|
|
features = [
|
|
"UUID Primary Keys with gen_random_uuid()",
|
|
"TIMESTAMP WITH TIME ZONE for all dates",
|
|
"Foreign Key Constraints"
|
|
]
|
|
|
|
# Add features based on technical requirements
|
|
technical_reqs = functional_reqs.get('technical_requirements', [])
|
|
|
|
for req in technical_reqs:
|
|
req_lower = req.lower()
|
|
if 'search' in req_lower or 'text' in req_lower:
|
|
features.append("Full Text Search with GIN indexes")
|
|
if 'audit' in req_lower or 'log' in req_lower:
|
|
features.append("Audit Triggers and Logging")
|
|
if 'encryption' in req_lower:
|
|
features.append("pgcrypto for data encryption")
|
|
|
|
# Add features based on business rules
|
|
business_rules = functional_reqs.get('business_logic_rules', [])
|
|
if business_rules:
|
|
features.extend([
|
|
"Row Level Security (RLS)",
|
|
"Check Constraints for business rules"
|
|
])
|
|
|
|
return features
|
|
|
|
def _create_feature_based_fallback(self, functional_reqs: Dict, tech_stack: Dict) -> Dict:
|
|
"""Create fallback PostgreSQL architecture based on actual features"""
|
|
logger.warning("Creating feature-based PostgreSQL fallback architecture")
|
|
|
|
feature_name = functional_reqs.get('feature_name', 'Application')
|
|
|
|
return {
|
|
"success": True,
|
|
"architecture": {
|
|
"database_schema": {
|
|
"note": f"Fallback schema for {feature_name}",
|
|
"tables": {
|
|
"users": "Basic user authentication table",
|
|
f"{feature_name.lower().replace(' ', '_')}": f"Main table for {feature_name} feature"
|
|
}
|
|
},
|
|
"security_implementation": {
|
|
"authentication": "SCRAM-SHA-256",
|
|
"ssl": "required"
|
|
},
|
|
"backup_strategy": {
|
|
"method": "pg_dump daily backups",
|
|
"retention": "30 days"
|
|
}
|
|
},
|
|
"specialist": "PostgreSQL",
|
|
"fallback": True,
|
|
"feature_based": True,
|
|
"generated_for": feature_name
|
|
}
|