refactore the structure of the backend services - dto - routes - models

Author: MonsefRH

backend/.gitignore

@@ -1 +1,4 @@
 .env
+*/__pycache__/
+*.pyc
+*.log

backend/requirements.txt

@@ -5,6 +5,7 @@ flask_wtf
 flask_cors
 pytz
 requests
+pydantic
 python-dotenv
 psycopg2-binary
 GitPython

backend/routes/History.py

@@ -1,56 +1,22 @@
 from flask import Blueprint, request, jsonify
-from routes.checkov import logger
-from utils.db import db
-from models.scan_history import ScanHistory
-from models.file_content import FileContent
+from services.history_service import get_scan_history
+from schemas.history_dto import ScanHistoryResponse, HistoryErrorResponse
+import logging
 
-history_bp = Blueprint('history', __name__)
+history_bp = Blueprint("history", __name__)
+logger = logging.getLogger(__name__)
 
 @history_bp.route("/history", methods=["GET"])
-def get_scan_history():
+def get_scan_history_route():
     user_id = request.headers.get("X-User-ID")
-    if not user_id:
-        return jsonify({"error": "user_id is required"}), 400
-
     scan_type = request.args.get("scan_type")
 
     try:
-        # Base query
-        query = ScanHistory.query.filter_by(user_id=user_id)
-        
-        # Add scan_type filter if provided
-        if scan_type:
-            query = query.filter_by(scan_type=scan_type)
-        
-        # Order by created_at DESC
-        query = query.order_by(ScanHistory.created_at.desc())
-        
-        history = []
-        for scan in query.all():
-            item_name = None
-            if scan.input_type:
-                file_content = FileContent.query.filter_by(scan_id=scan.id).first()
-                if file_content:
-                    item_name = f"{file_content.file_path} {file_content.id}"
-                    logger.debug(f"Item name for scan_id {scan.id}: {item_name}")
-
-            element = {
-                "id": scan.id,
-                "repo_id": scan.repo_id,
-                "scan_result": scan.scan_result,
-                "repo_url": scan.repo_url,
-                "item_name": item_name,
-                "status": scan.status,
-                "score": scan.score,
-                "compliant": scan.compliant,
-                "created_at": scan.created_at.isoformat(),
-                "input_type": scan.input_type,
-                "scan_type": scan.scan_type
-            }
-            history.append(element)
-
-        return jsonify(history)
-    except Exception as e:
-        logger.error(f"Failed to fetch scan history for user_id {user_id}: {str(e)}")
-        db.session.rollback()
-        return jsonify({"error": "Failed to fetch scan history"}), 500
+        history = get_scan_history(user_id, scan_type)
+        return jsonify([ScanHistoryResponse(**item).dict() for item in history]), 200
+    except ValueError as e:
+        logger.error(f"Invalid input for scan history: {str(e)}")
+        return jsonify(HistoryErrorResponse(error=str(e)).dict()), 400
+    except RuntimeError as e:
+        logger.error(f"Error fetching scan history: {str(e)}")
+        return jsonify(HistoryErrorResponse(error=str(e)).dict()), 500

backend/routes/admin.py

@@ -1,66 +1,23 @@
 from flask import Blueprint, jsonify
 from flask_jwt_extended import jwt_required, get_jwt_identity
-from utils.db import db
-from models.user import User
-from models.scan_history import ScanHistory
-from sqlalchemy import func
+from services.admin_service import get_admin_stats
+from schemas.admin_dto import AdminStatsResponse
+from pydantic import ValidationError
 
 admin_bp = Blueprint("admin", __name__)
 
 @admin_bp.route("/api/admin-stats", methods=["GET"])
 @jwt_required()
-def get_admin_stats():
-    user_id = get_jwt_identity()
-
+def get_admin_stats_route():
     try:
-        # Verify user is admin
-        user = User.query.filter_by(id=user_id).first()
-        if not user or user.role != "admin":
-            return jsonify({"error": "Access denied: Admin role required"}), 403
-
-        # Get total users (with role 'user')
-        total_users = User.query.filter_by(role="user").count()
-
-        # Get total scans
-        total_scans = ScanHistory.query.count()
-
-        # Get user data with scan counts
-        users = (
-            db.session.query(
-                User.id,
-                User.name,
-                User.email,
-                User.role,
-                User.created_at,
-                func.count(ScanHistory.id).label("scan_count")
-            )
-            .outerjoin(ScanHistory, User.id == ScanHistory.user_id)
-            .filter(User.role == "user")
-            .group_by(User.id, User.name, User.email, User.role, User.created_at)
-            .order_by(User.created_at.desc())
-            .all()
-        )
-
-        # Format users data for response
-        users_data = [
-            {
-                "id": user.id,
-                "name": user.name,
-                "email": user.email,
-                "role": user.role,
-                "created_at": user.created_at.isoformat(),
-                "scan_count": user.scan_count
-            }
-            for user in users
-        ]
-
-        return jsonify({
-            "totalUsers": total_users,
-            "totalScans": total_scans,
-            "users": users_data
-        })
-
+        user_id = get_jwt_identity()
+        stats = get_admin_stats(user_id)
+        return jsonify(AdminStatsResponse(**stats).dict()), 200
+    except ValidationError as e:
+        return jsonify({"error": e.errors()}), 400
+    except ValueError as e:
+        return jsonify({"error": str(e)}), 403
+    except RuntimeError as e:
+        return jsonify({"error": str(e)}), 500
     except Exception as e:
-        print(f"❌ Error fetching admin stats: {e}")
-        db.session.rollback()
-        return jsonify({"error": "Internal server error"}), 500
+        return jsonify({"error": f"Internal server error: {str(e)}"}), 500