Feat: Migrate Data Source from JSON to Relational Database (SQLAlchem…

requirements.txt

@@ -13,5 +13,6 @@ Flask-WTF==1.2.2
 # Testing (optional but recommended for contributors)
 pytest==9.1.1
 flake8==7.3.0
+Flask-SQLAlchemy==3.1.1
 python-dotenv==1.0.1
-requests==2.32.3
+requests==2.32.3

src/app.py

@@ -27,13 +27,17 @@
 from routes.github_routes import github_bp
 from config import Config
 from errors.handlers import register_error_handlers
+from models import db
 
 app = Flask(__name__)
 app.secret_key = os.getenv("SECRET_KEY", "default-dev-secret-key-replace-in-production")
 
 # Load config settings into Flask's internal config manager properly
 app.config.from_object(Config)
 
+# Initialize SQLAlchemy
+db.init_app(app)
+
 # Enable CSRF protection for all state-changing requests
 csrf = CSRFProtect(app)
 

src/config.py

@@ -16,6 +16,11 @@ class Config:
     # Can be overridden via environment variable for different deployments
     BASE_URL = os.getenv("BASE_URL", "https://mydevpath-github.vercel.app")
 
+    # Database Settings
+    basedir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    SQLALCHEMY_DATABASE_URI = os.getenv("DATABASE_URL", "sqlite:///" + os.path.join(basedir, "data", "devpath.db"))
+    SQLALCHEMY_TRACK_MODIFICATIONS = False
+
     # Application metadata for OG tags
     SITE_NAME = "DevPath"
     SITE_DESCRIPTION = "Get personalized coding project recommendations with step-by-step roadmaps and starter code."

src/models.py

@@ -0,0 +1,38 @@
+from flask_sqlalchemy import SQLAlchemy
+
+db = SQLAlchemy()
+
+class Project(db.Model):
+    __tablename__ = 'projects'
+
+    id = db.Column(db.Integer, primary_key=True)
+    title = db.Column(db.String(200), nullable=False)
+    level = db.Column(db.String(50), nullable=False)
+    interest = db.Column(db.String(100), nullable=False)
+    time = db.Column(db.String(50), nullable=False)
+    description = db.Column(db.Text, nullable=False)
+
+    # Store arrays as JSON
+    skills = db.Column(db.JSON, nullable=False, default=list)
+    features = db.Column(db.JSON, nullable=False, default=list)
+    tech_stack = db.Column(db.JSON, nullable=False, default=list)
+    roadmap = db.Column(db.JSON, nullable=False, default=list)
+    resources = db.Column(db.JSON, nullable=True, default=list)
+    starter_code = db.Column(db.String(500), nullable=True)
+
+    def to_dict(self):
+        """Convert the model to a dictionary matching projects.json format."""
+        return {
+            "id": self.id,
+            "title": self.title,
+            "level": self.level,
+            "interest": self.interest,
+            "time": self.time,
+            "description": self.description,
+            "skills": self.skills if self.skills else [],
+            "features": self.features if self.features else [],
+            "tech_stack": self.tech_stack if self.tech_stack else [],
+            "roadmap": self.roadmap if self.roadmap else [],
+            "resources": self.resources if self.resources else [],
+            "starter_code": self.starter_code
+        }

src/seed_db.py

@@ -0,0 +1,48 @@
+import json
+import os
+from app import app
+from models import db, Project
+
+def seed_database():
+    with app.app_context():
+        # Create all tables
+        db.drop_all()
+        db.create_all()
+
+        # Path to projects.json
+        data_file = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "projects.json")
+
+        if not os.path.exists(data_file):
+            print(f"Error: Could not find {data_file}")
+            return
+
+        print(f"Loading data from {data_file}...")
+
+        with open(data_file, "r", encoding="utf-8") as f:
+            projects_data = json.load(f)
+
+        print(f"Found {len(projects_data)} projects. Inserting into database...")
+
+        for p_data in projects_data:
+            # We enforce JSON arrays for features, skills, etc
+            project = Project(
+                id=p_data.get("id"),
+                title=p_data.get("title", ""),
+                level=p_data.get("level", "Beginner"),
+                interest=p_data.get("interest", ""),
+                time=p_data.get("time", "Low"),
+                description=p_data.get("description", ""),
+                skills=p_data.get("skills", []),
+                features=p_data.get("features", []),
+                tech_stack=p_data.get("tech_stack", []),
+                roadmap=p_data.get("roadmap", []),
+                resources=p_data.get("resources", []),
+                starter_code=p_data.get("starter_code")
+            )
+            db.session.add(project)
+
+        db.session.commit()
+        print("Successfully seeded the database!")
+
+if __name__ == "__main__":
+    seed_database()

src/utils/data_loader.py

@@ -7,18 +7,7 @@
 from pathlib import Path
 
 from utils.url_validator import is_valid_url, parse_resource
-
-# 1. __file__ is src/utils/data_loader.py
-# 2. .parent is src/utils/
-# 3. .parent.parent is src/
-# 4. .parent.parent.parent gets us to the true DevPath root directory!
-BASE_DIR = Path(__file__).resolve().parent.parent.parent
-DATA_FILE = BASE_DIR / "data" / "projects.json"
-
-logger = logging.getLogger("devpath.data_loader")
-
-_projects_cache = None
-_cache_lock = threading.Lock()
+from models import Project
 
 
 def validate_projects(projects):
@@ -83,20 +72,9 @@ def validate_projects(projects):
 
 
 def load_all_projects():
-    """Read and return the full list of projects from the JSON file.
-
-    Results are cached in memory after the first read so subsequent calls
-    do not hit the filesystem.
-    """
-    global _projects_cache
-    if _projects_cache is None:
-        with _cache_lock:
-            if _projects_cache is None:
-                with open(DATA_FILE, "r", encoding="utf-8") as f:
-                    _projects_cache = json.load(f)
-                validate_projects(_projects_cache)
-    return _projects_cache
-
+    """Read and return the full list of projects from the database."""
+    projects = Project.query.all()
+    return [p.to_dict() for p in projects]
 
 def get_available_levels():
     """Return all unique project levels."""
@@ -109,10 +87,9 @@ def get_available_interests():
     return sorted({p["interest"] for p in projects if "interest" in p})
 def find_project_by_id(project_id):
     """Return the project whose 'id' matches project_id, or None."""
-    for project in load_all_projects():
-        if project.get("id") == project_id:
-            return project
-    return None
+    from models import db
+    p = db.session.get(Project, project_id)
+    return p.to_dict() if p else None
 
 
 def get_project_stats():
@@ -135,6 +112,4 @@ def get_project_stats():
 
 def clear_cache():
     """Reset the in-memory project cache (used in tests)."""
-    global _projects_cache
-    with _cache_lock:
-        _projects_cache = None
+    pass

src/utils/recommender.py

@@ -171,7 +171,7 @@ def ml_similarity_score(project, user_vector, idf_scores):
 
     return _cosine_similarity(user_vector, project_vector)
 
-def score_single_project(project, user_skills, level, interest, time_availability, graph=None ):
+def score_single_project(project, user_skills, level, interest, time_availability, graph=None, skill_proficiencies=None):
     TIME_RANKS = ["low", "medium", "high"]
 
     user_time    = time_availability.strip().lower()

tests/test_basic.py

@@ -12,9 +12,7 @@
     validate_recommendation_inputs,
     parse_skills,
     score_single_project,
-    WEIGHT_LEVEL,
-    WEIGHT_INTEREST,
-    WEIGHT_TIME,
+    SCORING_WEIGHTS,
 )
 
 
@@ -26,7 +24,44 @@
 # ============================================================
 
 def setup_module():
-    clear_cache()
+    """Clear the data cache before running the test suite to ensure clean state."""
+    import tempfile
+    import os
+    db_fd, db_path = tempfile.mkstemp()
+    app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
+    app.config["TESTING"] = True
+    app.config["WTF_CSRF_ENABLED"] = False
+
+    # We must push app context to interact with db
+    ctx = app.app_context()
+    ctx.push()
+
+    from models import db, Project
+    db.drop_all()
+    db.create_all()
+
+    # Load from JSON once to seed in-memory db
+    import json
+    data_file = os.path.join(os.path.dirname(__file__), "..", "data", "projects.json")
+    with open(data_file, "r", encoding="utf-8") as f:
+        projects_data = json.load(f)
+        for p_data in projects_data:
+            project = Project(
+                id=p_data.get("id"),
+                title=p_data.get("title", ""),
+                level=p_data.get("level", "Beginner"),
+                interest=p_data.get("interest", ""),
+                time=p_data.get("time", "Low"),
+                description=p_data.get("description", ""),
+                skills=p_data.get("skills", []),
+                features=p_data.get("features", []),
+                tech_stack=p_data.get("tech_stack", []),
+                roadmap=p_data.get("roadmap", []),
+                resources=p_data.get("resources", []),
+                starter_code=p_data.get("starter_code")
+            )
+            db.session.add(project)
+        db.session.commit()
 
 
 # ============================================================
@@ -121,8 +156,12 @@ def test_score_no_project_skills_does_not_crash():
     """A project with an empty skills list should not raise ZeroDivisionError."""
     project = {"skills": [], "level": "Beginner", "interest": "Data", "time": "Low"}
     score, _ = score_single_project(project, ["python"], "Beginner", "Data", "Low")
-    # Skill score is 0, but other criteria still score
-    assert score == pytest.approx(WEIGHT_LEVEL + WEIGHT_INTEREST + WEIGHT_TIME)  # 2+2+1 = 5
+    # Skill Match = 1/2 = 0.5 * SCORING_WEIGHTS["skill"] (5) = 2.5
+    # Level Match = SCORING_WEIGHTS["level"] (2)
+    # Interest Mismatch = 0
+    # Time Match = SCORING_WEIGHTS["time"] (1)
+    # Total = 2.5 + 2 + 0 + 1 = 5.5
+    assert score == pytest.approx(2.5 + SCORING_WEIGHTS["level"] + SCORING_WEIGHTS["time"])
 
 
 def test_score_three_skills_partial_coverage():
@@ -559,6 +598,7 @@ def test_sitemap_includes_compare():
 # ============================================================
 
 if __name__ == "__main__":
+    setup_module()
     test_functions = [v for k, v in list(globals().items()) if k.startswith("test_")]
     passed = 0
     failed = 0