learningcircuit--local-deep-research/.pre-commit-hooks/check-deprecated-db.py

#!/usr/bin/env python3
"""
Pre-commit hook to check for usage of deprecated database connection methods.
Ensures code uses per-user database connections instead of the deprecated shared database.
"""

import sys
import re
import os

# Set environment variable for pre-commit hooks to allow unencrypted databases
os.environ["LDR_ALLOW_UNENCRYPTED"] = "true"


def check_file(filepath):
    """Check a single file for deprecated database usage."""
    issues = []

    with open(filepath, "r", encoding="utf-8") as f:
        content = f.read()
        lines = content.split("\n")

    # Pattern to detect get_db_connection usage
    db_connection_pattern = re.compile(r"\bget_db_connection\s*\(")

    # Pattern to detect direct imports of get_db_connection
    import_pattern = re.compile(
        r"from\s+[\w.]+\s+import\s+.*\bget_db_connection\b"
    )

    # Pattern to detect db_manager.get_session() — returns a raw QueuePool
    # session that must be closed by the caller.  Safe only inside
    # inject_current_user / ensure_db_session (stored in g.db_session and
    # cleaned up by teardown_appcontext).  All other call sites should use
    # get_user_db_session() context manager instead.
    raw_session_pattern = re.compile(r"\bdb_manager\.get_session\s*\(")

    # Check for usage
    for i, line in enumerate(lines, 1):
        # Skip comment and docstring lines — they can legitimately reference
        # deprecated APIs (TODOs, migration notes, inline examples).
        stripped = line.lstrip()
        if stripped.startswith(("#", '"""', "'''")):
            continue

        if db_connection_pattern.search(line):
            issues.append(
                f"{filepath}:{i}: Usage of deprecated get_db_connection()"
            )

        if import_pattern.search(line):
            issues.append(
                f"{filepath}:{i}: Import of deprecated get_db_connection"
            )

        if (
            raw_session_pattern.search(line)
            and "# noqa: raw-session" not in line
        ):
            issues.append(
                f"{filepath}:{i}: Direct db_manager.get_session() call — "
                "returns an unmanaged QueuePool session that leaks FDs if not closed. "
                "Use 'with get_user_db_session(username) as session:' instead, "
                "or add '# noqa: raw-session' if this is intentional (e.g. stored in g.db_session)"
            )

    # (The previous file-level "from ..web.models.database import get_db_connection"
    # substring check was removed: it was redundant with the per-line import_pattern
    # above, and its file-level scope also fired on comments mentioning the import.)

    # Check for SQLite connections to shared database.
    #
    # Previously the exemption was "get_user_db_session not in content" — a
    # *file-level* check. One correct get_user_db_session() call anywhere in the
    # file silently allowed raw sqlite3.connect("ldr.db") calls elsewhere in the
    # same file. Gate per-line on a "# noqa: shared-db" sentinel instead.
    shared_db_pattern = re.compile(r"sqlite3\.connect\s*\([^)]*ldr\.db")
    for i, line in enumerate(lines, 1):
        if line.lstrip().startswith(("#", '"""', "'''")):
            continue
        if shared_db_pattern.search(line) and "# noqa: shared-db" not in line:
            issues.append(
                f"{filepath}:{i}: Direct SQLite connection to shared database - use get_user_db_session() instead"
            )

    return issues


def main():
    """Main function to check all provided files."""
    if len(sys.argv) < 2:
        print("No files to check")
        return 0

    all_issues = []

    for filepath in sys.argv[1:]:
        # Skip the database.py file itself (it contains the deprecated function definition)
        if "web/models/database.py" in filepath:
            continue

        # Skip files that legitimately manage raw sessions (store in g.db_session
        # or define the deprecated helpers themselves)
        if any(
            skip in filepath
            for skip in [
                "session_context.py",  # ensure_db_session stores in g.db_session
                "web/auth/decorators.py",  # inject_current_user stores in g.db_session
                "encrypted_db.py",  # defines get_session()
                "db_utils.py",  # defines get_db_session() wrapper
            ]
        ):
            continue

        # Skip migration scripts and test files that might legitimately need shared DB access
        if any(
            skip in filepath
            for skip in ["migrations/", "tests/", "test_", ".pre-commit-hooks/"]
        ):
            continue

        issues = check_file(filepath)
        all_issues.extend(issues)

    if all_issues:
        print("❌ Deprecated or unsafe database access detected!\n")
        print("The shared database (get_db_connection) is deprecated.")
        print(
            "Direct db_manager.get_session() leaks QueuePool connections (FDs)."
        )
        print(
            "Please use get_user_db_session(username) for per-user database access.\n"
        )
        print("Issues found:")
        for issue in all_issues:
            print(f"  - {issue}")
        print("\nExample fix:")
        print("  # Old (deprecated):")
        print("  conn = get_db_connection()")
        print("  cursor = conn.cursor()")
        print("  # ... SQL query execution ...")
        print()
        print("  # New (correct):")
        print("  from flask import session")
        print("  username = session.get('username', 'anonymous')")
        print("  with get_user_db_session(username) as db_session:")
        print("      results = db_session.query(Model).filter(...).all()")
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())