Samir Boulahtit
7c43d6f4a2
Eliminate all 103 errors and 96 warnings from the architecture validator: Phase 1 - Validator rules & YAML: - Add NAM-001/NAM-002 exceptions for module-scoped router/service files - Fix API-004 to detect # public comments on decorator lines - Add module-specific exception bases to EXC-004 valid_bases - Exclude storefront files from AUTH-004 store context check - Add SVC-006 exceptions for loyalty service atomic commits - Fix _get_rule() to search naming_rules and auth_rules categories - Use plain # CODE comments instead of # noqa: CODE for custom rules Phase 2 - Billing module (5 route files): - Move _resolve_store_to_merchant to subscription_service - Move tier/feature queries to feature_service, admin_subscription_service - Extract 22 inline Pydantic schemas to billing/schemas/billing.py - Replace all HTTPException with domain exceptions Phase 3 - Loyalty module (4 routes + points_service): - Add 7 domain exceptions (Apple auth, enrollment, device registration) - Add service methods to card_service, program_service, apple_wallet_service - Move all db.query() from routes to service layer - Fix SVC-001: replace HTTPException in points_service with domain exception Phase 4 - Remaining modules: - tenancy: move store stats queries to admin_service - cms: move platform resolution to content_page_service, add NoPlatformSubscriptionException - messaging: move user/customer lookups to messaging_service - Add ConfigDict(from_attributes=True) to ContentPageResponse Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
194 lines
6.5 KiB
Python
194 lines
6.5 KiB
Python
# app/modules/monitoring/services/background_tasks_service.py
|
|
"""
|
|
Background Tasks Service
|
|
Service for monitoring background tasks across the system
|
|
"""
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import case, desc, func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.dev_tools.models import ArchitectureScan, TestRun
|
|
from app.modules.marketplace.models import MarketplaceImportJob
|
|
|
|
|
|
class BackgroundTasksService:
|
|
"""Service for monitoring background tasks"""
|
|
|
|
def get_import_jobs(
|
|
self, db: Session, status: str | None = None, limit: int = 50
|
|
) -> list[MarketplaceImportJob]:
|
|
"""Get import jobs with optional status filter"""
|
|
query = db.query(MarketplaceImportJob)
|
|
if status:
|
|
query = query.filter(MarketplaceImportJob.status == status)
|
|
return query.order_by(desc(MarketplaceImportJob.created_at)).limit(limit).all()
|
|
|
|
def get_test_runs(
|
|
self, db: Session, status: str | None = None, limit: int = 50
|
|
) -> list[TestRun]:
|
|
"""Get test runs with optional status filter"""
|
|
query = db.query(TestRun)
|
|
if status:
|
|
query = query.filter(TestRun.status == status)
|
|
return query.order_by(desc(TestRun.timestamp)).limit(limit).all()
|
|
|
|
def get_running_imports(self, db: Session) -> list[MarketplaceImportJob]:
|
|
"""Get currently running import jobs"""
|
|
return (
|
|
db.query(MarketplaceImportJob)
|
|
.filter(MarketplaceImportJob.status == "processing")
|
|
.all()
|
|
)
|
|
|
|
def get_running_test_runs(self, db: Session) -> list[TestRun]:
|
|
"""Get currently running test runs"""
|
|
# SVC-005 - Platform-level, TestRuns not store-scoped
|
|
return db.query(TestRun).filter(TestRun.status == "running").all() # SVC-005
|
|
|
|
def get_import_stats(self, db: Session) -> dict:
|
|
"""Get import job statistics"""
|
|
today_start = datetime.now(UTC).replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
)
|
|
|
|
stats = db.query(
|
|
func.count(MarketplaceImportJob.id).label("total"),
|
|
func.sum(
|
|
case((MarketplaceImportJob.status == "processing", 1), else_=0)
|
|
).label("running"),
|
|
func.sum(
|
|
case(
|
|
(
|
|
MarketplaceImportJob.status.in_(
|
|
["completed", "completed_with_errors"]
|
|
),
|
|
1,
|
|
),
|
|
else_=0,
|
|
)
|
|
).label("completed"),
|
|
func.sum(
|
|
case((MarketplaceImportJob.status == "failed", 1), else_=0)
|
|
).label("failed"),
|
|
).first()
|
|
|
|
today_count = (
|
|
db.query(func.count(MarketplaceImportJob.id))
|
|
.filter(MarketplaceImportJob.created_at >= today_start)
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
return {
|
|
"total": stats.total or 0,
|
|
"running": stats.running or 0,
|
|
"completed": stats.completed or 0,
|
|
"failed": stats.failed or 0,
|
|
"today": today_count,
|
|
}
|
|
|
|
def get_test_run_stats(self, db: Session) -> dict:
|
|
"""Get test run statistics"""
|
|
today_start = datetime.now(UTC).replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
)
|
|
|
|
stats = db.query(
|
|
func.count(TestRun.id).label("total"),
|
|
func.sum(case((TestRun.status == "running", 1), else_=0)).label(
|
|
"running"
|
|
),
|
|
func.sum(case((TestRun.status == "passed", 1), else_=0)).label(
|
|
"completed"
|
|
),
|
|
func.sum(
|
|
case((TestRun.status.in_(["failed", "error"]), 1), else_=0)
|
|
).label("failed"),
|
|
func.avg(TestRun.duration_seconds).label("avg_duration"),
|
|
).first()
|
|
|
|
today_count = (
|
|
db.query(func.count(TestRun.id))
|
|
.filter(TestRun.timestamp >= today_start)
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
return {
|
|
"total": stats.total or 0,
|
|
"running": stats.running or 0,
|
|
"completed": stats.completed or 0,
|
|
"failed": stats.failed or 0,
|
|
"today": today_count,
|
|
"avg_duration": round(stats.avg_duration or 0, 1),
|
|
}
|
|
|
|
def get_code_quality_scans(
|
|
self, db: Session, status: str | None = None, limit: int = 50
|
|
) -> list[ArchitectureScan]:
|
|
"""Get code quality scans with optional status filter"""
|
|
query = db.query(ArchitectureScan)
|
|
if status:
|
|
query = query.filter(ArchitectureScan.status == status)
|
|
return query.order_by(desc(ArchitectureScan.timestamp)).limit(limit).all()
|
|
|
|
def get_running_scans(self, db: Session) -> list[ArchitectureScan]:
|
|
"""Get currently running code quality scans"""
|
|
return (
|
|
db.query(ArchitectureScan)
|
|
.filter(ArchitectureScan.status.in_(["pending", "running"]))
|
|
.all()
|
|
)
|
|
|
|
def get_scan_stats(self, db: Session) -> dict:
|
|
"""Get code quality scan statistics"""
|
|
today_start = datetime.now(UTC).replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
)
|
|
|
|
stats = db.query(
|
|
func.count(ArchitectureScan.id).label("total"),
|
|
func.sum(
|
|
case(
|
|
(ArchitectureScan.status.in_(["pending", "running"]), 1), else_=0
|
|
)
|
|
).label("running"),
|
|
func.sum(
|
|
case(
|
|
(
|
|
ArchitectureScan.status.in_(
|
|
["completed", "completed_with_warnings"]
|
|
),
|
|
1,
|
|
),
|
|
else_=0,
|
|
)
|
|
).label("completed"),
|
|
func.sum(
|
|
case((ArchitectureScan.status == "failed", 1), else_=0)
|
|
).label("failed"),
|
|
func.avg(ArchitectureScan.duration_seconds).label("avg_duration"),
|
|
).first()
|
|
|
|
today_count = (
|
|
db.query(func.count(ArchitectureScan.id))
|
|
.filter(ArchitectureScan.timestamp >= today_start)
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
return {
|
|
"total": stats.total or 0,
|
|
"running": stats.running or 0,
|
|
"completed": stats.completed or 0,
|
|
"failed": stats.failed or 0,
|
|
"today": today_count,
|
|
"avg_duration": round(stats.avg_duration or 0, 1),
|
|
}
|
|
|
|
|
|
# Singleton instance
|
|
background_tasks_service = BackgroundTasksService()
|