# 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 }