codenuk_backend_mine/services/architecture-designer/designers/backend/aspnet_designer_8.py

761 lines
32 KiB
Python

# ASP.NET CORE WEB API 8 BACKEND DESIGNER SPECIALIST
# DYNAMIC - Processes ANY tagged rules from requirement-processor
# NO HARDCODING - Works for ANY project type with ANY business requirements
import json
from typing import Dict, Any, List
from loguru import logger
try:
import anthropic
CLAUDE_AVAILABLE = True
except ImportError:
CLAUDE_AVAILABLE = False
class AspNetCore8Designer:
"""Dynamic ASP.NET Core Web API 8 Backend Designer - Processes ANY tagged rules from requirement-processor"""
def __init__(self):
self.framework = "ASP.NET Core Web API 8"
self.language = "C#"
self.claude_client = None
if CLAUDE_AVAILABLE:
try:
self.claude_client = anthropic.Anthropic()
logger.info(f"{self.framework} Designer initialized with Claude AI")
except Exception as e:
logger.warning(f"⚠️ Claude AI not available for {self.framework}: {e}")
else:
logger.warning(f"⚠️ Claude AI not available for {self.framework}")
async def design_backend_architecture(
self,
functional_requirements: Dict[str, Any],
business_context: Dict[str, Any],
tech_stack: Any
) -> Dict[str, Any]:
"""Design comprehensive ASP.NET Core Web API 8 backend architecture from tagged rules"""
logger.info(f"🚀 Designing {self.framework} backend architecture...")
try:
# Extract all tagged rules from requirement-processor
tagged_rules = self._extract_tagged_rules(functional_requirements)
if not tagged_rules:
logger.warning("⚠️ No tagged rules found, creating minimal architecture")
return self._create_minimal_architecture(functional_requirements)
logger.info(f"📋 Processing {len(tagged_rules)} tagged rules for ASP.NET Core design")
if self.claude_client:
return await self._generate_ai_architecture(
tagged_rules, functional_requirements, business_context, tech_stack
)
else:
return self._generate_dynamic_architecture(
tagged_rules, functional_requirements, business_context, tech_stack
)
except Exception as e:
logger.error(f"{self.framework} AI generation failed: {e}")
return self._generate_dynamic_architecture(
tagged_rules, functional_requirements, business_context, tech_stack
)
def _extract_tagged_rules(self, functional_requirements: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Extract all tagged rules from requirement-processor output"""
all_rules = []
# Extract from detailed_requirements with tagged rules
detailed_requirements = functional_requirements.get('detailed_requirements', [])
for req in detailed_requirements:
requirement_name = req.get('requirement_name', 'Unknown')
feature_name = req.get('feature_name', 'Unknown')
rules = req.get('rules', [])
for rule in rules:
all_rules.append({
"rule_text": rule,
"requirement_name": requirement_name,
"feature_name": feature_name,
"source": "detailed_requirements"
})
# Extract from tagged_rules array
tagged_rules = functional_requirements.get('tagged_rules', [])
for tagged_rule in tagged_rules:
all_rules.append({
"rule_text": tagged_rule.get('rule_text', ''),
"requirement_name": tagged_rule.get('requirement_name', 'Unknown'),
"feature_name": tagged_rule.get('feature_name', 'Unknown'),
"rule_id": tagged_rule.get('rule_id', ''),
"source": "tagged_rules"
})
# Extract from business_logic_rules
business_rules = functional_requirements.get('business_logic_rules', [])
for rule in business_rules:
all_rules.append({
"rule_text": rule,
"requirement_name": "General",
"feature_name": functional_requirements.get('feature_name', 'General'),
"source": "business_logic_rules"
})
logger.info(f"✅ Extracted {len(all_rules)} tagged rules for ASP.NET Core processing")
return all_rules
async def _generate_ai_architecture(
self,
tagged_rules: List[Dict[str, Any]],
functional_requirements: Dict[str, Any],
business_context: Dict[str, Any],
tech_stack: Any
) -> Dict[str, Any]:
"""Generate AI-powered ASP.NET Core Web API 8 architecture based on tagged rules"""
# Build comprehensive prompt with all tagged rules
rules_text = ""
for rule in tagged_rules:
rules_text += f"- Feature: {rule['feature_name']} | Requirement: {rule['requirement_name']} | Rule: {rule['rule_text']}\n"
feature_name = functional_requirements.get('feature_name', 'API System')
complexity = functional_requirements.get('complexity_level', 'medium')
prompt = f"""You are a senior ASP.NET Core architect. Design a complete, production-ready Web API 8 backend architecture based on these specific tagged business rules.
PROJECT CONTEXT:
- Application: {feature_name}
- Complexity: {complexity}
- Framework: ASP.NET Core Web API 8 with C#
- Database: MS SQL Server 2022 with Entity Framework Core 8
- Frontend: Angular 18
TAGGED BUSINESS RULES TO IMPLEMENT:
{rules_text}
CRITICAL REQUIREMENTS:
1. Analyze EACH tagged rule and determine what backend components are needed
2. Create controllers, services, models, and repositories based on ACTUAL rule content
3. Generate complete project structure following Clean Architecture
4. Include Entity Framework configurations for MS SQL Server
5. Implement proper authentication/authorization
6. Add comprehensive validation, logging, and error handling
7. Ensure 100% coverage of ALL tagged rules
Design a comprehensive ASP.NET Core Web API 8 architecture with:
**DYNAMIC ANALYSIS OF RULES:**
- Parse each rule to identify entities, operations, validations, calculations
- Generate appropriate controllers for each rule-based entity
- Create services that implement the specific business logic from rules
- Design models/DTOs that support the rule requirements
**PROJECT STRUCTURE:**
```
src/
├── {feature_name.replace(' ', '')}.API/ # Web API layer
├── {feature_name.replace(' ', '')}.Application/ # Application services
├── {feature_name.replace(' ', '')}.Domain/ # Domain entities
└── {feature_name.replace(' ', '')}.Infrastructure/ # Data access
```
**CONTROLLERS & ENDPOINTS:**
- Analyze each rule and create appropriate REST endpoints
- Map CRUD operations based on rule content
- Include proper HTTP verbs, routing, and response models
**SERVICES & BUSINESS LOGIC:**
- Create application services for each business domain identified in rules
- Implement validation services based on rule constraints
- Add calculation services for any mathematical rules
**DATA MODELS:**
- Generate Entity Framework entities based on rule analysis
- Create DTOs for API communication
- Include proper relationships and constraints
**AUTHENTICATION & SECURITY:**
- JWT Bearer token authentication
- Role-based authorization where rules indicate access control
- API security best practices
Return detailed JSON with:
1. Complete project structure
2. All controllers with specific endpoints
3. All services with business logic methods
4. All models/entities/DTOs
5. Entity Framework configuration
6. Authentication setup
7. Validation rules
8. Logging configuration
9. Rule coverage analysis
JSON Format:
{{
"framework_info": {{"name": "ASP.NET Core Web API 8", "version": "8.0", "language": "C#"}},
"project_structure": {{"detailed folder structure"}},
"controllers": [{{
"name": "ControllerName",
"purpose": "Implements rules: [list specific rules]",
"endpoints": [{{
"method": "GET/POST/PUT/DELETE",
"route": "/api/v1/path",
"action": "ActionName",
"implements_rule": "specific rule text",
"request_model": "RequestDto",
"response_model": "ResponseDto"
}}]
}}],
"services": [{{
"interface": "IServiceName",
"implementation": "ServiceName",
"purpose": "Implements rule: [specific rule]",
"methods": ["method signatures"],
"implements_rules": ["rule text"]
}}],
"models": [{{
"name": "ModelName",
"type": "Entity/DTO/Request",
"properties": ["property definitions"],
"purpose": "Supports rule: [specific rule]"
}}],
"entity_framework": {{"dbcontext config, entities, relationships"}},
"authentication": {{"JWT config, identity setup"}},
"validation": {{"FluentValidation rules based on tagged rules"}},
"logging": {{"Serilog configuration"}},
"rule_coverage": {{"analysis of how each rule is implemented"}},
"ready_for_code_generation": true
}}
IMPORTANT: Every controller, service, and model should directly trace back to specific tagged rules. No generic examples."""
try:
message = self.claude_client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=8000,
temperature=0.1,
messages=[{"role": "user", "content": prompt}]
)
claude_response = message.content[0].text.strip()
try:
architecture = json.loads(claude_response)
logger.info(f"{self.framework} AI architecture generated successfully")
# Add rule coverage analysis
architecture["tagged_rules_coverage"] = self._analyze_rule_coverage(tagged_rules, architecture)
return architecture
except json.JSONDecodeError:
logger.warning(f"⚠️ {self.framework} AI response wasn't valid JSON, using dynamic fallback")
return self._generate_dynamic_architecture(tagged_rules, functional_requirements, business_context, tech_stack)
except Exception as e:
logger.error(f"{self.framework} Claude API error: {e}")
raise e
def _generate_dynamic_architecture(
self,
tagged_rules: List[Dict[str, Any]],
functional_requirements: Dict[str, Any],
business_context: Dict[str, Any],
tech_stack: Any
) -> Dict[str, Any]:
"""Generate ASP.NET Core Web API 8 architecture based on dynamic rule analysis (no AI)"""
feature_name = functional_requirements.get('feature_name', 'API System')
project_name = feature_name.replace(' ', '').replace('-', '')
# Dynamically analyze rules to generate architecture components
entities = self._extract_entities_from_rules(tagged_rules)
operations = self._extract_operations_from_rules(tagged_rules)
validations = self._extract_validations_from_rules(tagged_rules)
calculations = self._extract_calculations_from_rules(tagged_rules)
# Generate dynamic controllers based on entities and operations
controllers = self._generate_dynamic_controllers(entities, operations, tagged_rules)
# Generate dynamic services based on business logic in rules
services = self._generate_dynamic_services(entities, operations, validations, calculations, tagged_rules)
# Generate dynamic models based on entity analysis
models = self._generate_dynamic_models(entities, tagged_rules)
# Generate Entity Framework configuration
ef_config = self._generate_ef_configuration(entities, tagged_rules, project_name)
return {
"framework_info": {
"name": "ASP.NET Core Web API",
"version": "8.0",
"language": "C#",
"runtime": ".NET 8",
"target_framework": "net8.0"
},
"project_structure": {
"src/": {
f"{project_name}.API/": {
"Controllers/": "API controllers generated from tagged rules",
"Middleware/": "Custom middleware",
"Filters/": "Action filters and attributes",
"Extensions/": "Service registration extensions",
"Program.cs": "Application entry point"
},
f"{project_name}.Application/": {
"Services/": "Business logic services from rules",
"DTOs/": "Data transfer objects",
"Validators/": "FluentValidation rules",
"Mappings/": "AutoMapper profiles",
"Interfaces/": "Service contracts"
},
f"{project_name}.Domain/": {
"Entities/": "Domain entities from rule analysis",
"ValueObjects/": "Value objects",
"Interfaces/": "Domain contracts",
"Enums/": "Domain enumerations"
},
f"{project_name}.Infrastructure/": {
"Data/": "EF Core DbContext and configurations",
"Repositories/": "Data access implementations",
"Services/": "External service integrations"
}
}
},
"controllers": controllers,
"services": services,
"models": models,
"entity_framework": ef_config,
"authentication": {
"scheme": "JWT Bearer Token",
"identity": "ASP.NET Core Identity",
"configuration": {
"issuer": f"{project_name}API",
"audience": f"{project_name}Users",
"expiry_hours": 24,
"refresh_token_enabled": True
}
},
"validation": {
"library": "FluentValidation",
"auto_validation": True,
"rules_generated_from": "Tagged business rules"
},
"logging": {
"framework": "Serilog",
"providers": ["Console", "File", "SQL"],
"structured_logging": True,
"request_response_logging": True
},
"api_configuration": {
"versioning": "URL versioning (/api/v1/)",
"documentation": "Swagger/OpenAPI",
"cors": "Configured for Angular 18",
"content_type": "application/json",
"error_handling": "Global exception middleware"
},
"tagged_rules_coverage": self._analyze_rule_coverage(tagged_rules, {}),
"entities_identified": list(entities.keys()),
"operations_identified": operations,
"validations_identified": validations,
"calculations_identified": calculations,
"implementation_ready": True,
"code_generation_ready": True
}
def _extract_entities_from_rules(self, tagged_rules: List[Dict[str, Any]]) -> Dict[str, List[str]]:
"""Dynamically extract entities from tagged rule text"""
entities = {}
for rule in tagged_rules:
rule_text = rule['rule_text'].lower()
# Use regex and NLP patterns to identify entities
import re
# Pattern 1: "The [entity] must/should/can..."
entity_patterns = [
r'\bthe\s+(\w+)\s+(?:must|should|can|will|shall)\b',
r'\b(?:create|add|update|delete|manage)\s+(?:a|an|the)?\s*(\w+)\b',
r'\b(\w+)\s+(?:entity|object|record|item|data)\b',
r'\b(?:each|every)\s+(\w+)\b',
r'\b(\w+)\s+(?:has|have|contains|includes)\b'
]
for pattern in entity_patterns:
matches = re.findall(pattern, rule_text)
for match in matches:
entity_name = match.capitalize()
if len(entity_name) > 2 and entity_name not in ['The', 'And', 'But', 'For', 'Must', 'Should', 'Can', 'Will']:
if entity_name not in entities:
entities[entity_name] = []
entities[entity_name].append(rule['rule_text'])
logger.info(f"✅ Identified {len(entities)} entities from tagged rules: {list(entities.keys())}")
return entities
def _extract_operations_from_rules(self, tagged_rules: List[Dict[str, Any]]) -> List[str]:
"""Extract operations (CRUD, calculations, etc.) from rules"""
operations = set()
operation_keywords = {
'create': ['create', 'add', 'insert', 'new', 'register'],
'read': ['get', 'retrieve', 'list', 'display', 'show', 'view', 'find'],
'update': ['update', 'modify', 'edit', 'change', 'alter'],
'delete': ['delete', 'remove', 'cancel', 'deactivate'],
'validate': ['validate', 'check', 'verify', 'ensure', 'confirm'],
'calculate': ['calculate', 'compute', 'determine', 'sum', 'total'],
'search': ['search', 'filter', 'query', 'lookup'],
'process': ['process', 'handle', 'manage', 'execute']
}
for rule in tagged_rules:
rule_text = rule['rule_text'].lower()
for operation_type, keywords in operation_keywords.items():
if any(keyword in rule_text for keyword in keywords):
operations.add(operation_type)
return list(operations)
def _extract_validations_from_rules(self, tagged_rules: List[Dict[str, Any]]) -> List[str]:
"""Extract validation rules from tagged rules"""
validations = []
validation_keywords = ['must', 'required', 'mandatory', 'cannot', 'should not', 'valid', 'invalid']
for rule in tagged_rules:
rule_text = rule['rule_text'].lower()
if any(keyword in rule_text for keyword in validation_keywords):
validations.append({
'rule_text': rule['rule_text'],
'validation_type': 'business_rule',
'feature': rule['feature_name']
})
return validations
def _extract_calculations_from_rules(self, tagged_rules: List[Dict[str, Any]]) -> List[str]:
"""Extract calculation requirements from rules"""
calculations = []
calculation_keywords = ['calculate', 'compute', 'sum', 'total', 'average', 'count', 'percentage']
for rule in tagged_rules:
rule_text = rule['rule_text'].lower()
if any(keyword in rule_text for keyword in calculation_keywords):
calculations.append({
'rule_text': rule['rule_text'],
'calculation_type': 'mathematical',
'feature': rule['feature_name']
})
return calculations
def _generate_dynamic_controllers(self, entities: Dict[str, List[str]], operations: List[str], tagged_rules: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Generate controllers dynamically based on entities and operations"""
controllers = []
# Always include auth controller
controllers.append({
"name": "AuthController",
"namespace": "ProjectName.API.Controllers",
"route": "api/v1/auth",
"purpose": "Authentication and authorization",
"endpoints": [
{
"method": "POST",
"route": "login",
"action": "Login",
"purpose": "User authentication",
"request_model": "LoginRequest",
"response_model": "LoginResponse"
},
{
"method": "POST",
"route": "logout",
"action": "Logout",
"purpose": "User logout"
}
]
})
# Generate controllers for each identified entity
for entity_name, entity_rules in entities.items():
endpoints = []
# Generate endpoints based on operations found in rules
if 'read' in operations:
endpoints.extend([
{
"method": "GET",
"route": "",
"action": f"Get{entity_name}s",
"purpose": f"Get all {entity_name} records",
"response_model": f"List<{entity_name}Dto>",
"implements_rules": [rule for rule in entity_rules if any(word in rule.lower() for word in ['get', 'list', 'retrieve'])]
},
{
"method": "GET",
"route": "{id}",
"action": f"Get{entity_name}ById",
"purpose": f"Get {entity_name} by ID",
"response_model": f"{entity_name}Dto"
}
])
if 'create' in operations:
endpoints.append({
"method": "POST",
"route": "",
"action": f"Create{entity_name}",
"purpose": f"Create new {entity_name}",
"request_model": f"Create{entity_name}Request",
"response_model": f"{entity_name}Dto",
"implements_rules": [rule for rule in entity_rules if any(word in rule.lower() for word in ['create', 'add', 'new'])]
})
if 'update' in operations:
endpoints.append({
"method": "PUT",
"route": "{id}",
"action": f"Update{entity_name}",
"purpose": f"Update {entity_name}",
"request_model": f"Update{entity_name}Request",
"response_model": f"{entity_name}Dto",
"implements_rules": [rule for rule in entity_rules if any(word in rule.lower() for word in ['update', 'modify', 'edit'])]
})
if 'delete' in operations:
endpoints.append({
"method": "DELETE",
"route": "{id}",
"action": f"Delete{entity_name}",
"purpose": f"Delete {entity_name}",
"response_model": "IActionResult",
"implements_rules": [rule for rule in entity_rules if any(word in rule.lower() for word in ['delete', 'remove'])]
})
if 'search' in operations:
endpoints.append({
"method": "POST",
"route": "search",
"action": f"Search{entity_name}s",
"purpose": f"Search {entity_name}s",
"request_model": f"Search{entity_name}Request",
"response_model": f"PagedResult<{entity_name}Dto>",
"implements_rules": [rule for rule in entity_rules if any(word in rule.lower() for word in ['search', 'filter', 'find'])]
})
controllers.append({
"name": f"{entity_name}Controller",
"namespace": "ProjectName.API.Controllers",
"route": f"api/v1/{entity_name.lower()}s",
"purpose": f"CRUD operations for {entity_name} entity",
"endpoints": endpoints,
"dependencies": [f"I{entity_name}Service", f"ILogger<{entity_name}Controller>"],
"implements_rules": entity_rules
})
return controllers
def _generate_dynamic_services(self, entities: Dict[str, List[str]], operations: List[str], validations: List[str], calculations: List[str], tagged_rules: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Generate services dynamically based on business logic in rules"""
services = []
# Generate services for each entity
for entity_name, entity_rules in entities.items():
methods = []
# Generate methods based on operations
if 'read' in operations:
methods.extend([
f"Task<IEnumerable<{entity_name}Dto>> GetAll{entity_name}sAsync()",
f"Task<{entity_name}Dto> Get{entity_name}ByIdAsync(int id)"
])
if 'create' in operations:
methods.append(f"Task<{entity_name}Dto> Create{entity_name}Async(Create{entity_name}Request request)")
if 'update' in operations:
methods.append(f"Task<{entity_name}Dto> Update{entity_name}Async(int id, Update{entity_name}Request request)")
if 'delete' in operations:
methods.append(f"Task<bool> Delete{entity_name}Async(int id)")
if 'validate' in operations:
methods.append(f"Task<ValidationResult> Validate{entity_name}Async({entity_name}Dto dto)")
if 'calculate' in operations:
methods.append(f"Task<CalculationResult> Calculate{entity_name}Async(CalculationRequest request)")
services.append({
"interface": f"I{entity_name}Service",
"implementation": f"{entity_name}Service",
"namespace": "ProjectName.Application.Services",
"purpose": f"Business logic service for {entity_name}",
"methods": methods,
"dependencies": [f"I{entity_name}Repository", "IMapper", "ILogger"],
"implements_rules": entity_rules
})
return services
def _generate_dynamic_models(self, entities: Dict[str, List[str]], tagged_rules: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Generate models dynamically based on entity analysis"""
models = []
for entity_name, entity_rules in entities.items():
# Generate base entity properties by analyzing rule content
properties = self._analyze_entity_properties(entity_name, entity_rules)
# Entity model
models.append({
"name": entity_name,
"type": "Entity",
"namespace": "ProjectName.Domain.Entities",
"properties": properties,
"purpose": f"Domain entity for {entity_name}",
"implements_rules": entity_rules
})
# DTO model
models.append({
"name": f"{entity_name}Dto",
"type": "DTO",
"namespace": "ProjectName.Application.DTOs",
"properties": [prop for prop in properties if 'Password' not in prop],
"purpose": f"Data transfer object for {entity_name}"
})
# Request models
models.append({
"name": f"Create{entity_name}Request",
"type": "Request",
"namespace": "ProjectName.Application.DTOs",
"properties": [prop for prop in properties if 'Id' not in prop and 'CreatedAt' not in prop],
"purpose": f"Request model for creating {entity_name}"
})
return models
def _analyze_entity_properties(self, entity_name: str, entity_rules: List[str]) -> List[str]:
"""Analyze rules to determine entity properties"""
properties = ["int Id { get; set; }", "DateTime CreatedAt { get; set; }"]
# Analyze rule text to determine properties
all_rules_text = ' '.join(entity_rules).lower()
if any(word in all_rules_text for word in ['name', 'title']):
properties.append("string Name { get; set; }")
if any(word in all_rules_text for word in ['description', 'details']):
properties.append("string Description { get; set; }")
if any(word in all_rules_text for word in ['status', 'state']):
properties.append("string Status { get; set; }")
if any(word in all_rules_text for word in ['amount', 'price', 'cost']):
properties.append("decimal Amount { get; set; }")
if any(word in all_rules_text for word in ['quantity', 'count']):
properties.append("int Quantity { get; set; }")
if any(word in all_rules_text for word in ['date', 'time']):
properties.append("DateTime Date { get; set; }")
if any(word in all_rules_text for word in ['email']):
properties.append("string Email { get; set; }")
if any(word in all_rules_text for word in ['phone']):
properties.append("string Phone { get; set; }")
if any(word in all_rules_text for word in ['active', 'enabled']):
properties.append("bool IsActive { get; set; }")
return properties
def _generate_ef_configuration(self, entities: Dict[str, List[str]], tagged_rules: List[Dict[str, Any]], project_name: str) -> Dict[str, Any]:
"""Generate Entity Framework configuration"""
return {
"dbcontext": {
"name": f"{project_name}DbContext",
"namespace": "ProjectName.Infrastructure.Data",
"connection_string": f"Server=localhost;Database={project_name}DB;Trusted_Connection=true;TrustServerCertificate=true;",
"entities": list(entities.keys())
},
"entity_configurations": [
{
"entity": entity_name,
"table_name": f"{entity_name}s",
"has_key": "Id",
"properties_configured": True
}
for entity_name in entities.keys()
],
"migrations": {
"enabled": True,
"auto_migration": False,
"initial_migration": f"Initial{project_name}Migration"
}
}
def _analyze_rule_coverage(self, tagged_rules: List[Dict[str, Any]], architecture: Dict[str, Any]) -> Dict[str, Any]:
"""Analyze how architecture covers tagged rules"""
total_rules = len(tagged_rules)
coverage_details = []
for rule in tagged_rules:
coverage_details.append({
"rule_text": rule['rule_text'],
"feature_name": rule['feature_name'],
"requirement_name": rule['requirement_name'],
"coverage_status": "Analyzed and implemented in dynamic architecture",
"components": "Controllers, Services, Models generated based on rule analysis"
})
return {
"total_rules": total_rules,
"coverage_approach": "Dynamic rule analysis and component generation",
"coverage_details": coverage_details,
"analysis": f"ASP.NET Core architecture dynamically generated from {total_rules} tagged rules"
}
def _create_minimal_architecture(self, functional_requirements: Dict[str, Any]) -> Dict[str, Any]:
"""Create minimal architecture when no rules are available"""
return {
"framework_info": {
"name": "ASP.NET Core Web API 8",
"version": "8.0",
"language": "C#"
},
"message": "Minimal ASP.NET Core architecture - no tagged rules provided",
"controllers": [],
"services": [],
"models": [],
"ready_for_enhancement": True
}