""" Model-Schema Analyzer - Analyzes SQLAlchemy models and Pydantic schemas to detect coverage gaps """ import re from pathlib import Path from typing import Dict, List, Set, Tuple from dataclasses import dataclass @dataclass class ModelField: """Represents a field in a SQLAlchemy model""" name: str type: str nullable: bool file: str line: int @dataclass class SchemaField: """Represents a field in a Pydantic schema""" name: str type: str required: bool file: str line: int class ModelSchemaAnalyzer: """Analyzes model-schema coverage to detect missing fields""" def __init__(self, project_dir: str): self.project_dir = Path(project_dir) self.models_dir = self.project_dir / "backend" / "app" / "models" self.schemas_dir = self.project_dir / "backend" / "app" / "schemas" # Model to schema mapping (manual mapping for common patterns) # Only include models that should have response schemas self.model_schema_mapping = { "User": ["UserResponse", "UserCreate", "UserUpdate", "UserListResponse", "UserLogin", "UserProfileUpdate", "UserRoleUpdate"], "Stream": ["StreamResponse", "StreamCreate", "StreamUpdate", "StreamManualCreate"], "Comment": ["CommentResponse", "CommentCreate", "CommentUpdate"], "Tag": ["TagResponse", "TagCreate"], "ChatMessage": ["ChatMessageResponse", "ChatMessageCreate"], "WatchHistory": ["WatchHistoryResponse", "WatchHistoryCreate", "WatchHistoryUpdate"], "TwitchOAuth": ["TwitchOAuthResponse", "TwitchOAuthRequest", "TwitchUserInfo"], "SiteSetting": ["SiteSettingResponse", "SiteSettingUpdate", "SiteSettingBase", "SiteSettingsResponse", "SiteSettingsUpdateBulk"], "Permission": ["PermissionResponse", "PermissionCreate", "PermissionUpdate"], "Role": ["RoleResponse", "RoleCreate", "RoleUpdate"] } # Models without response schemas (internal models, skip validation) self.models_without_schemas = { "ActivityLog", # Internal audit log "Chapter", # Only in stream metadata "ProcessingJob", # Internal background job "Session", # Internal session management "StreamTag", # Join table "RolePermission" # Join table } # Fields to ignore (expected to be missing from schemas) self.fields_to_ignore = { # Primary keys (often excluded from response schemas) "id", # Timestamps (often auto-managed) "created_at", "updated_at", # Sensitive fields (should not be in response schemas) "password_hash", "refresh_token", "access_token", # Foreign keys (often excluded from response schemas) "user_id", "stream_id", "tag_id", "parent_id", # Internal fields "ip_address", "user_agent", # JSON fields (often not in response schemas or have complex types) "stream_metadata", "details", "badges", "emotes" } def analyze_models(self) -> Dict[str, List[ModelField]]: """Analyze SQLAlchemy models to extract all fields""" models_info = {} if not self.models_dir.exists(): return models_info for model_file in self.models_dir.glob("*.py"): if model_file.name == "__init__.py": continue model_data = self._extract_model_fields(model_file) for model_name, fields in model_data: if model_name and fields: models_info[model_name] = fields return models_info def _extract_model_fields(self, model_file: Path) -> List[Tuple[str, List[ModelField]]]: """Extract model name and fields from a model file""" try: with open(model_file, 'r', encoding='utf-8') as f: content = f.read() models_data = [] # Find all class definitions that inherit from Base class_pattern = r'class\s+(\w+)\s*\(\s*Base\s*\):' for class_match in re.finditer(class_pattern, content): class_name = class_match.group(1) class_start = class_match.start() class_end = class_match.end() # Extract the class content (start after class definition) lines = content[class_end:].split('\n') class_content = [] indent_level = None has_tablename = False for line in lines: if line.strip(): line_indent = len(line) - len(line.lstrip()) if indent_level is None: # First non-empty line in class body indent_level = line_indent if line_indent == indent_level: # Same level as class body class_content.append(line) if '__tablename__' in line: has_tablename = True elif line_indent < indent_level: # Dedented - end of class break else: # Empty line - keep it class_content.append(line) # Only process if this has __tablename__ (it's a model, not a base class) if not has_tablename: continue class_content_str = '\n'.join(class_content) # Extract fields fields = [] column_pattern = r'(\w+)\s*=\s*Column\(([^)]+)\)' for col_match in re.finditer(column_pattern, class_content_str): field_name = col_match.group(1) column_args = col_match.group(2) line_num = content[:class_start + class_content_str.index(col_match.group(0))].count('\n') + 1 # Skip common non-column attributes if field_name not in ['__tablename__', '__table_args__', '__mapper_args__']: # Extract type and nullable type_match = re.search(r'(String|Integer|Text|Boolean|DateTime|UUID|ForeignKey)\(?\d*\)?', column_args) nullable_match = re.search(r'nullable\s*=\s*(True|False)', column_args) field_type = type_match.group(1) if type_match else "Unknown" nullable = nullable_match.group(1) == "True" if nullable_match else True fields.append(ModelField( name=field_name, type=field_type, nullable=nullable, file=str(model_file.relative_to(self.project_dir)), line=line_num )) if fields: models_data.append((class_name, fields)) return models_data except Exception as e: print(f"Error analyzing {model_file}: {e}") return [] def analyze_schemas(self) -> Dict[str, List[SchemaField]]: """Analyze Pydantic schemas to extract all fields""" schemas_info = {} if not self.schemas_dir.exists(): return schemas_info for schema_file in self.schemas_dir.glob("*.py"): if schema_file.name == "__init__.py": continue schema_data = self._extract_schema_fields(schema_file) for schema_name, fields in schema_data: if schema_name and fields: schemas_info[schema_name] = fields return schemas_info def _extract_schema_fields(self, schema_file: Path) -> List[Tuple[str, List[SchemaField]]]: """Extract schema name and fields from a schema file""" try: with open(schema_file, 'r', encoding='utf-8') as f: content = f.read() schemas_data = [] # Find all class definitions that inherit from BaseModel class_pattern = r'class\s+(\w+)\(.*?BaseModel.*?\):' for class_match in re.finditer(class_pattern, content): schema_name = class_match.group(1) class_start = class_match.start() class_end = class_match.end() # Extract fields from this class (start after class definition) lines = content[class_end:].split('\n') class_content = [] indent_level = None for line in lines: if line.strip(): line_indent = len(line) - len(line.lstrip()) if indent_level is None: # First non-empty line in class body indent_level = line_indent if line_indent == indent_level: # Same level as class body class_content.append(line) elif line_indent < indent_level: # Dedented - end of class break else: # Empty line - keep it class_content.append(line) class_content_str = '\n'.join(class_content) # Extract fields (Pydantic v2 syntax: field_name: type = Field(...)) fields = [] # Pattern for field_name: type = Field(...) or field_name: type field_pattern = r'(\w+)\s*:\s*([^\s=]+)(?:\s*=\s*(?:Field\([^)]*\)|.*?))?(?:\s*#.*)?$' for line in class_content_str.split('\n'): line = line.strip() if not line or line.startswith('#') or line.startswith('@'): continue field_match = re.match(field_pattern, line) if field_match: field_name = field_match.group(1) field_type = field_match.group(2) line_num = content[:class_start + class_content_str.index(line)].count('\n') + 1 # Check if field is required (no default value and not Optional) is_optional = 'Optional' in field_type or '|' in field_type or 'None' in field_type has_default = '=' in line and 'Field(' not in line is_required = not is_optional and not has_default # Skip common non-field attributes if field_name not in ['model_config', 'Config']: fields.append(SchemaField( name=field_name, type=field_type, required=is_required, file=str(schema_file.relative_to(self.project_dir)), line=line_num )) schemas_data.append((schema_name, fields)) return schemas_data except Exception as e: print(f"Error analyzing {schema_file}: {e}") return [] def compare_model_schema_coverage(self) -> Dict: """Compare model fields with schema fields to detect coverage gaps""" models_info = self.analyze_models() schemas_info = self.analyze_schemas() coverage_report = { "total_models": len(models_info), "total_model_fields": 0, "covered_fields": 0, "missing_from_schemas": {}, "extra_in_schemas": {}, "coverage_percentage": 0.0, "model_details": {} } # Count total model fields for model_name, fields in models_info.items(): coverage_report["total_model_fields"] += len(fields) # Analyze each model for model_name, model_fields in models_info.items(): # Skip models without response schemas (internal models) if model_name in self.models_without_schemas: continue # Get related schemas related_schemas = self.model_schema_mapping.get(model_name, []) # Collect all schema fields from related schemas schema_fields_set = set() schema_fields_dict = {} for schema_name in related_schemas: if schema_name in schemas_info: for field in schemas_info[schema_name]: schema_fields_set.add(field.name) schema_fields_dict[field.name] = field # Check which model fields are missing from schemas model_field_names = {field.name for field in model_fields} missing_fields = model_field_names - schema_fields_set # Filter out ignored fields missing_fields = missing_fields - self.fields_to_ignore # Check which schema fields are extra (not in model) extra_fields = schema_fields_set - model_field_names # Count covered fields covered_count = len(model_field_names) - len(missing_fields) coverage_report["covered_fields"] += covered_count if missing_fields: coverage_report["missing_from_schemas"][model_name] = [ { "name": field.name, "type": field.type, "nullable": field.nullable, "file": field.file, "line": field.line } for field in model_fields if field.name in missing_fields ] if extra_fields: coverage_report["extra_in_schemas"][model_name] = [ { "name": field.name, "type": field.type, "required": field.required, "file": field.file, "line": field.line } for field_name in extra_fields for field in schema_fields_dict.values() if field.name == field_name ] # Store model details coverage_report["model_details"][model_name] = { "total_fields": len(model_fields), "covered_fields": covered_count, "missing_fields": len(missing_fields), "coverage_percentage": round(covered_count / len(model_fields) * 100, 1) if model_fields else 100.0, "related_schemas": related_schemas } # Calculate overall coverage percentage if coverage_report["total_model_fields"] > 0: coverage_report["coverage_percentage"] = round( coverage_report["covered_fields"] / coverage_report["total_model_fields"] * 100, 1 ) return coverage_report def print_report(self, coverage_report: Dict): """Print a human-readable report of model-schema coverage""" print("=" * 60) print("Model-Schema Coverage Report") print("=" * 60) print(f"\nTotal Models: {coverage_report['total_models']}") print(f"Total Model Fields: {coverage_report['total_model_fields']}") print(f"Covered Fields: {coverage_report['covered_fields']}") print(f"Coverage: {coverage_report['coverage_percentage']}%") if coverage_report["missing_from_schemas"]: print("\nMissing from Schemas:") for model_name, fields in coverage_report["missing_from_schemas"].items(): print(f"\n Model: {model_name}") for field in fields: print(f" - {field['name']}: {field['type']} (nullable={field['nullable']})") print(f" Defined in: {field['file']}:{field['line']}") if coverage_report["extra_in_schemas"]: print("\nExtra in Schemas (not in model):") for model_name, fields in coverage_report["extra_in_schemas"].items(): print(f"\n Model: {model_name}") for field in fields: print(f" - {field['name']}: {field['type']} (required={field['required']})") print(f" Defined in: {field['file']}:{field['line']}") if not coverage_report["missing_from_schemas"] and not coverage_report["extra_in_schemas"]: print("\n✓ All model fields are covered in schemas") print("\nModel Details:") for model_name, details in coverage_report["model_details"].items(): print(f" {model_name}: {details['covered_fields']}/{details['total_fields']} ({details['coverage_percentage']}%)") print("=" * 60) if __name__ == "__main__": # Test the analyzer project_dir = Path(__file__).parent.parent.parent analyzer = ModelSchemaAnalyzer(str(project_dir)) coverage_report = analyzer.compare_model_schema_coverage() analyzer.print_report(coverage_report)