feat: complete dev_tools module self-containment
Migrate dev_tools module to self-contained structure: - routes/api/ - API endpoints - models/architecture_scan.py - Architecture scan models - models/test_run.py - Test run models - schemas/ - Pydantic schemas - services/ - Business logic services - tasks/ - Celery background tasks - exceptions.py - Module exceptions Updated definition.py with self-contained paths. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
45
app/modules/dev_tools/models/__init__.py
Normal file
45
app/modules/dev_tools/models/__init__.py
Normal file
@@ -0,0 +1,45 @@
|
||||
# app/modules/dev_tools/models/__init__.py
|
||||
"""
|
||||
Dev-Tools module database models.
|
||||
|
||||
This is the canonical location for dev-tools models. Module models are automatically
|
||||
discovered and registered with SQLAlchemy's Base.metadata at startup.
|
||||
|
||||
Usage:
|
||||
from app.modules.dev_tools.models import (
|
||||
ArchitectureScan,
|
||||
ArchitectureViolation,
|
||||
ArchitectureRule,
|
||||
ViolationAssignment,
|
||||
ViolationComment,
|
||||
TestRun,
|
||||
TestResult,
|
||||
TestCollection,
|
||||
)
|
||||
"""
|
||||
|
||||
from app.modules.dev_tools.models.architecture_scan import (
|
||||
ArchitectureScan,
|
||||
ArchitectureViolation,
|
||||
ArchitectureRule,
|
||||
ViolationAssignment,
|
||||
ViolationComment,
|
||||
)
|
||||
from app.modules.dev_tools.models.test_run import (
|
||||
TestRun,
|
||||
TestResult,
|
||||
TestCollection,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
# Architecture scan models
|
||||
"ArchitectureScan",
|
||||
"ArchitectureViolation",
|
||||
"ArchitectureRule",
|
||||
"ViolationAssignment",
|
||||
"ViolationComment",
|
||||
# Test run models
|
||||
"TestRun",
|
||||
"TestResult",
|
||||
"TestCollection",
|
||||
]
|
||||
206
app/modules/dev_tools/models/architecture_scan.py
Normal file
206
app/modules/dev_tools/models/architecture_scan.py
Normal file
@@ -0,0 +1,206 @@
|
||||
# app/modules/dev_tools/models/architecture_scan.py
|
||||
"""
|
||||
Architecture Scan Models
|
||||
|
||||
Database models for tracking code quality scans and violations.
|
||||
This is the canonical location - models are re-exported from the legacy location
|
||||
for backward compatibility.
|
||||
"""
|
||||
|
||||
from sqlalchemy import (
|
||||
JSON,
|
||||
Boolean,
|
||||
Column,
|
||||
DateTime,
|
||||
Float,
|
||||
ForeignKey,
|
||||
Integer,
|
||||
String,
|
||||
Text,
|
||||
)
|
||||
from sqlalchemy.orm import relationship
|
||||
from sqlalchemy.sql import func
|
||||
|
||||
from app.core.database import Base
|
||||
|
||||
|
||||
class ArchitectureScan(Base):
|
||||
"""Represents a single run of a code quality validator"""
|
||||
|
||||
__tablename__ = "architecture_scans"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
timestamp = Column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False, index=True
|
||||
)
|
||||
validator_type = Column(
|
||||
String(20), nullable=False, index=True, default="architecture"
|
||||
) # 'architecture', 'security', 'performance'
|
||||
|
||||
# Background task status fields (harmonized architecture)
|
||||
status = Column(
|
||||
String(30), nullable=False, default="pending", index=True
|
||||
) # 'pending', 'running', 'completed', 'failed', 'completed_with_warnings'
|
||||
started_at = Column(DateTime(timezone=True), nullable=True)
|
||||
completed_at = Column(DateTime(timezone=True), nullable=True)
|
||||
error_message = Column(Text, nullable=True)
|
||||
progress_message = Column(String(255), nullable=True) # Current step description
|
||||
|
||||
# Scan results
|
||||
total_files = Column(Integer, default=0)
|
||||
total_violations = Column(Integer, default=0)
|
||||
errors = Column(Integer, default=0)
|
||||
warnings = Column(Integer, default=0)
|
||||
duration_seconds = Column(Float, default=0.0)
|
||||
triggered_by = Column(String(100)) # 'manual:username', 'scheduled', 'ci/cd'
|
||||
git_commit_hash = Column(String(40))
|
||||
|
||||
# Celery task tracking (optional - for USE_CELERY=true)
|
||||
celery_task_id = Column(String(255), nullable=True, index=True)
|
||||
|
||||
# Relationship to violations
|
||||
violations = relationship(
|
||||
"ArchitectureViolation", back_populates="scan", cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<ArchitectureScan(id={self.id}, violations={self.total_violations}, errors={self.errors})>"
|
||||
|
||||
|
||||
class ArchitectureViolation(Base):
|
||||
"""Represents a single code quality violation found during a scan"""
|
||||
|
||||
__tablename__ = "architecture_violations"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
scan_id = Column(
|
||||
Integer, ForeignKey("architecture_scans.id"), nullable=False, index=True
|
||||
)
|
||||
validator_type = Column(
|
||||
String(20), nullable=False, index=True, default="architecture"
|
||||
) # 'architecture', 'security', 'performance'
|
||||
rule_id = Column(String(20), nullable=False, index=True) # e.g., 'API-001', 'SEC-001', 'PERF-001'
|
||||
rule_name = Column(String(200), nullable=False)
|
||||
severity = Column(
|
||||
String(10), nullable=False, index=True
|
||||
) # 'error', 'warning', 'info'
|
||||
file_path = Column(String(500), nullable=False, index=True)
|
||||
line_number = Column(Integer, nullable=False)
|
||||
message = Column(Text, nullable=False)
|
||||
context = Column(Text) # Code snippet
|
||||
suggestion = Column(Text)
|
||||
status = Column(
|
||||
String(20), default="open", index=True
|
||||
) # 'open', 'assigned', 'resolved', 'ignored', 'technical_debt'
|
||||
assigned_to = Column(Integer, ForeignKey("users.id"))
|
||||
resolved_at = Column(DateTime(timezone=True))
|
||||
resolved_by = Column(Integer, ForeignKey("users.id"))
|
||||
resolution_note = Column(Text)
|
||||
created_at = Column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
|
||||
# Relationships
|
||||
scan = relationship("ArchitectureScan", back_populates="violations")
|
||||
assigned_user = relationship(
|
||||
"User", foreign_keys=[assigned_to], backref="assigned_violations"
|
||||
)
|
||||
resolver = relationship(
|
||||
"User", foreign_keys=[resolved_by], backref="resolved_violations"
|
||||
)
|
||||
assignments = relationship(
|
||||
"ViolationAssignment", back_populates="violation", cascade="all, delete-orphan"
|
||||
)
|
||||
comments = relationship(
|
||||
"ViolationComment", back_populates="violation", cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<ArchitectureViolation(id={self.id}, rule={self.rule_id}, file={self.file_path}:{self.line_number})>"
|
||||
|
||||
|
||||
class ArchitectureRule(Base):
|
||||
"""Code quality rules configuration (from YAML with database overrides)"""
|
||||
|
||||
__tablename__ = "architecture_rules"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
rule_id = Column(
|
||||
String(20), unique=True, nullable=False, index=True
|
||||
) # e.g., 'API-001', 'SEC-001', 'PERF-001'
|
||||
validator_type = Column(
|
||||
String(20), nullable=False, index=True, default="architecture"
|
||||
) # 'architecture', 'security', 'performance'
|
||||
category = Column(
|
||||
String(50), nullable=False
|
||||
) # 'api_endpoint', 'service_layer', 'authentication', 'database', etc.
|
||||
name = Column(String(200), nullable=False)
|
||||
description = Column(Text)
|
||||
severity = Column(String(10), nullable=False) # Can override default from YAML
|
||||
enabled = Column(Boolean, default=True, nullable=False)
|
||||
custom_config = Column(JSON) # For rule-specific settings
|
||||
created_at = Column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
updated_at = Column(
|
||||
DateTime(timezone=True),
|
||||
server_default=func.now(),
|
||||
onupdate=func.now(),
|
||||
nullable=False,
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<ArchitectureRule(id={self.rule_id}, name={self.name}, enabled={self.enabled})>"
|
||||
|
||||
|
||||
class ViolationAssignment(Base):
|
||||
"""Tracks assignment of violations to developers"""
|
||||
|
||||
__tablename__ = "violation_assignments"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
violation_id = Column(
|
||||
Integer, ForeignKey("architecture_violations.id"), nullable=False, index=True
|
||||
)
|
||||
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
|
||||
assigned_at = Column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
assigned_by = Column(Integer, ForeignKey("users.id"))
|
||||
due_date = Column(DateTime(timezone=True))
|
||||
priority = Column(
|
||||
String(10), default="medium"
|
||||
) # 'low', 'medium', 'high', 'critical'
|
||||
|
||||
# Relationships
|
||||
violation = relationship("ArchitectureViolation", back_populates="assignments")
|
||||
user = relationship("User", foreign_keys=[user_id], backref="violation_assignments")
|
||||
assigner = relationship(
|
||||
"User", foreign_keys=[assigned_by], backref="assigned_by_me"
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<ViolationAssignment(id={self.id}, violation_id={self.violation_id}, user_id={self.user_id})>"
|
||||
|
||||
|
||||
class ViolationComment(Base):
|
||||
"""Comments on violations for collaboration"""
|
||||
|
||||
__tablename__ = "violation_comments"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
violation_id = Column(
|
||||
Integer, ForeignKey("architecture_violations.id"), nullable=False, index=True
|
||||
)
|
||||
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
|
||||
comment = Column(Text, nullable=False)
|
||||
created_at = Column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
|
||||
# Relationships
|
||||
violation = relationship("ArchitectureViolation", back_populates="comments")
|
||||
user = relationship("User", backref="violation_comments")
|
||||
|
||||
def __repr__(self):
|
||||
return f"<ViolationComment(id={self.id}, violation_id={self.violation_id}, user_id={self.user_id})>"
|
||||
151
app/modules/dev_tools/models/test_run.py
Normal file
151
app/modules/dev_tools/models/test_run.py
Normal file
@@ -0,0 +1,151 @@
|
||||
# app/modules/dev_tools/models/test_run.py
|
||||
"""
|
||||
Test Run Models
|
||||
|
||||
Database models for tracking pytest test runs and results.
|
||||
This is the canonical location - models are re-exported from the legacy location
|
||||
for backward compatibility.
|
||||
"""
|
||||
|
||||
from sqlalchemy import (
|
||||
JSON,
|
||||
Column,
|
||||
DateTime,
|
||||
Float,
|
||||
ForeignKey,
|
||||
Integer,
|
||||
String,
|
||||
Text,
|
||||
)
|
||||
from sqlalchemy.orm import relationship
|
||||
from sqlalchemy.sql import func
|
||||
|
||||
from app.core.database import Base
|
||||
|
||||
|
||||
class TestRun(Base):
|
||||
"""Represents a single pytest run"""
|
||||
|
||||
__tablename__ = "test_runs"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
timestamp = Column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False, index=True
|
||||
)
|
||||
|
||||
# Test counts
|
||||
total_tests = Column(Integer, default=0)
|
||||
passed = Column(Integer, default=0)
|
||||
failed = Column(Integer, default=0)
|
||||
errors = Column(Integer, default=0)
|
||||
skipped = Column(Integer, default=0)
|
||||
xfailed = Column(Integer, default=0) # Expected failures
|
||||
xpassed = Column(Integer, default=0) # Unexpected passes
|
||||
|
||||
# Coverage info (optional)
|
||||
coverage_percent = Column(Float, nullable=True)
|
||||
|
||||
# Timing
|
||||
duration_seconds = Column(Float, default=0.0)
|
||||
|
||||
# Run metadata
|
||||
triggered_by = Column(String(100)) # 'manual', 'scheduled', 'ci/cd'
|
||||
git_commit_hash = Column(String(40))
|
||||
git_branch = Column(String(100))
|
||||
test_path = Column(String(500)) # Which tests were run (e.g., 'tests/unit')
|
||||
pytest_args = Column(String(500)) # Command line arguments used
|
||||
|
||||
# Status
|
||||
status = Column(
|
||||
String(20), default="running", index=True
|
||||
) # 'running', 'passed', 'failed', 'error'
|
||||
|
||||
# Celery task tracking (optional - for USE_CELERY=true)
|
||||
celery_task_id = Column(String(255), nullable=True, index=True)
|
||||
|
||||
# Relationship to test results
|
||||
results = relationship(
|
||||
"TestResult", back_populates="run", cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<TestRun(id={self.id}, total={self.total_tests}, passed={self.passed}, failed={self.failed})>"
|
||||
|
||||
@property
|
||||
def pass_rate(self) -> float:
|
||||
"""Calculate pass rate as percentage"""
|
||||
if self.total_tests == 0:
|
||||
return 0.0
|
||||
return (self.passed / self.total_tests) * 100
|
||||
|
||||
|
||||
class TestResult(Base):
|
||||
"""Represents a single test result from a pytest run"""
|
||||
|
||||
__tablename__ = "test_results"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
run_id = Column(Integer, ForeignKey("test_runs.id"), nullable=False, index=True)
|
||||
|
||||
# Test identification
|
||||
node_id = Column(
|
||||
String(500), nullable=False, index=True
|
||||
) # e.g., 'tests/unit/test_foo.py::test_bar'
|
||||
test_name = Column(String(200), nullable=False) # e.g., 'test_bar'
|
||||
test_file = Column(String(300), nullable=False) # e.g., 'tests/unit/test_foo.py'
|
||||
test_class = Column(String(200)) # e.g., 'TestFooClass' (optional)
|
||||
|
||||
# Result
|
||||
outcome = Column(
|
||||
String(20), nullable=False, index=True
|
||||
) # 'passed', 'failed', 'error', 'skipped', 'xfailed', 'xpassed'
|
||||
duration_seconds = Column(Float, default=0.0)
|
||||
|
||||
# Failure details (if applicable)
|
||||
error_message = Column(Text)
|
||||
traceback = Column(Text)
|
||||
|
||||
# Test metadata
|
||||
markers = Column(JSON) # List of pytest markers
|
||||
parameters = Column(JSON) # Parametrized test params
|
||||
|
||||
# Timestamps
|
||||
created_at = Column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
|
||||
# Relationships
|
||||
run = relationship("TestRun", back_populates="results")
|
||||
|
||||
def __repr__(self):
|
||||
return f"<TestResult(id={self.id}, node_id={self.node_id}, outcome={self.outcome})>"
|
||||
|
||||
|
||||
class TestCollection(Base):
|
||||
"""Cached test collection info for quick stats"""
|
||||
|
||||
__tablename__ = "test_collections"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
|
||||
# Collection stats
|
||||
total_tests = Column(Integer, default=0)
|
||||
total_files = Column(Integer, default=0)
|
||||
total_classes = Column(Integer, default=0)
|
||||
|
||||
# By category
|
||||
unit_tests = Column(Integer, default=0)
|
||||
integration_tests = Column(Integer, default=0)
|
||||
performance_tests = Column(Integer, default=0)
|
||||
system_tests = Column(Integer, default=0)
|
||||
|
||||
# Collection data
|
||||
test_files = Column(JSON) # List of test files with counts
|
||||
|
||||
# Timestamps
|
||||
collected_at = Column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<TestCollection(id={self.id}, total={self.total_tests})>"
|
||||
Reference in New Issue
Block a user