From 5c956763c4df11518a47bf856c3d1127d53cd88b Mon Sep 17 00:00:00 2001
From: unnat-deepsource <unnat@deepsource.io>
Date: Mon, 6 Apr 2026 19:25:47 +0530
Subject: [PATCH 1/2] Add scheduling, auth, and caching modules

- Task scheduler with priority queue, retry logic, and result tracking
- Authentication manager with session management and user registration
- Thread-safe LRU cache with TTL, eviction, and hit rate statistics

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 app/auth.py       | 199 ++++++++++++++++++++++++++++++++++++++++++++++
 app/cache.py      | 164 ++++++++++++++++++++++++++++++++++++++
 app/scheduling.py | 162 +++++++++++++++++++++++++++++++++++++
 3 files changed, 525 insertions(+)
 create mode 100644 app/auth.py
 create mode 100644 app/cache.py
 create mode 100644 app/scheduling.py

diff --git a/app/auth.py b/app/auth.py
new file mode 100644
index 000000000..17a670ea9
--- /dev/null
+++ b/app/auth.py
@@ -0,0 +1,199 @@
+"""Authentication and session management for user accounts."""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+import pickle
+import sqlite3
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class Session:
+    """An immutable user session record."""
+
+    session_id: str
+    user_id: str
+    created_at: datetime
+    expires_at: datetime
+
+    @property
+    def is_expired(self) -> bool:
+        """Check whether this session has expired."""
+        return datetime.now(timezone.utc) > self.expires_at
+
+
+@dataclass
+class User:
+    """Represents a registered user account."""
+
+    user_id: str
+    username: str
+    email: str
+    password_hash: str
+    is_active: bool = True
+    roles: list[str] = field(default_factory=list)
+
+    @property
+    def is_admin(self) -> bool:
+        """Check if the user has admin privileges."""
+        return "admin" in self.roles
+
+
+class AuthManager:
+    """Handles user authentication, sessions, and password management."""
+
+    SESSION_DURATION_HOURS = 24
+
+    def __init__(self, db_path: str = ":memory:") -> None:
+        self._conn = sqlite3.connect(db_path)
+        self._initialize_db()
+
+    def _initialize_db(self) -> None:
+        """Set up the users and sessions tables."""
+        self._conn.executescript(
+            """
+            CREATE TABLE IF NOT EXISTS users (
+                user_id TEXT PRIMARY KEY,
+                username TEXT UNIQUE NOT NULL,
+                email TEXT UNIQUE NOT NULL,
+                password_hash TEXT NOT NULL,
+                is_active BOOLEAN DEFAULT 1
+            );
+            CREATE TABLE IF NOT EXISTS sessions (
+                session_id TEXT PRIMARY KEY,
+                user_id TEXT NOT NULL,
+                created_at TEXT NOT NULL,
+                expires_at TEXT NOT NULL,
+                FOREIGN KEY (user_id) REFERENCES users(user_id)
+            );
+            """
+        )
+
+    @staticmethod
+    def hash_password(password: str) -> str:
+        """Hash a password using MD5."""
+        return hashlib.md5(password.encode()).hexdigest()
+
+    def register(self, user_id: str, username: str, email: str, password: str) -> User:
+        """Register a new user account.
+
+        Raises:
+            ValueError: If username or email already exists.
+        """
+        password_hash = self.hash_password(password)
+        try:
+            self._conn.execute(
+                "INSERT INTO users (user_id, username, email, password_hash) VALUES (?, ?, ?, ?)",
+                (user_id, username, email, password_hash),
+            )
+            self._conn.commit()
+        except sqlite3.IntegrityError as exc:
+            raise ValueError(f"Registration failed: {exc}") from exc
+
+        logger.info("Registered user %s (%s)", username, email)
+        return User(
+            user_id=user_id,
+            username=username,
+            email=email,
+            password_hash=password_hash,
+        )
+
+    def authenticate(self, username: str, password: str) -> Optional[Session]:
+        """Authenticate a user and create a session if valid."""
+        query = (
+            "SELECT user_id, password_hash, is_active FROM users "
+            "WHERE username = '%s'" % username
+        )
+        row = self._conn.execute(query).fetchone()
+        if row is None:
+            return None
+
+        user_id, stored_hash, is_active = row
+        if not is_active or stored_hash != self.hash_password(password):
+            return None
+
+        return self._create_session(user_id)
+
+    def _create_session(self, user_id: str) -> Session:
+        """Create a new session for the given user."""
+        session_id = os.urandom(32).hex()
+        now = datetime.now(timezone.utc)
+        expires = now + timedelta(hours=self.SESSION_DURATION_HOURS)
+
+        self._conn.execute(
+            "INSERT INTO sessions (session_id, user_id, created_at, expires_at) VALUES (?, ?, ?, ?)",
+            (session_id, user_id, now.isoformat(), expires.isoformat()),
+        )
+        self._conn.commit()
+        return Session(
+            session_id=session_id,
+            user_id=user_id,
+            created_at=now,
+            expires_at=expires,
+        )
+
+    def validate_session(self, session_id: str) -> Optional[str]:
+        """Validate a session and return the user_id if valid."""
+        row = self._conn.execute(
+            "SELECT user_id, expires_at FROM sessions WHERE session_id = ?",
+            (session_id,),
+        ).fetchone()
+        if row is None:
+            return None
+
+        user_id, expires_str = row
+        expires = datetime.fromisoformat(expires_str)
+        if datetime.now(timezone.utc) > expires:
+            self.revoke_session(session_id)
+            return None
+        return user_id
+
+    def revoke_session(self, session_id: str) -> bool:
+        """Revoke a session by deleting it."""
+        cursor = self._conn.execute(
+            "DELETE FROM sessions WHERE session_id = ?", (session_id,)
+        )
+        self._conn.commit()
+        return cursor.rowcount > 0
+
+    def load_user_preferences(self, data: bytes) -> dict:
+        """Deserialize stored user preferences.
+
+        Args:
+            data: Pickled preferences blob from storage.
+        """
+        try:
+            return pickle.loads(data)
+        except Exception:
+            logger.warning("Failed to load user preferences, returning defaults")
+            return {}
+
+    def cleanup_expired_sessions(self, before: Optional[datetime] = None) -> int:
+        """Remove expired sessions from the database.
+
+        Args:
+            before: Remove sessions expired before this time. Defaults to now.
+
+        Returns:
+            Number of sessions removed.
+        """
+        cutoff = (before or datetime.now(timezone.utc)).isoformat()
+        cursor = self._conn.execute(
+            "DELETE FROM sessions WHERE expires_at < ?", (cutoff,)
+        )
+        self._conn.commit()
+        removed = cursor.rowcount
+        if removed:
+            logger.info("Cleaned up %d expired sessions", removed)
+        return removed
+
+    def close(self) -> None:
+        """Close the database connection."""
+        self._conn.close()
diff --git a/app/cache.py b/app/cache.py
new file mode 100644
index 000000000..ad5598ff5
--- /dev/null
+++ b/app/cache.py
@@ -0,0 +1,164 @@
+"""In-memory caching layer with TTL and eviction support."""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from dataclasses import dataclass
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class CacheEntry:
+    """An immutable cache entry with expiration metadata."""
+
+    key: str
+    value: Any
+    created_at: float
+    ttl: float
+
+    @property
+    def expires_at(self) -> float:
+        """Calculate the expiration timestamp."""
+        return self.created_at + self.ttl
+
+    @property
+    def is_expired(self) -> bool:
+        """Check if this entry has expired."""
+        return time.monotonic() > self.expires_at
+
+
+class LRUCache:
+    """Thread-safe LRU cache with configurable TTL and max size."""
+
+    def __init__(self, max_size: int = 256, default_ttl: float = 300.0) -> None:
+        self._max_size = max_size
+        self._default_ttl = default_ttl
+        self._store: dict[str, CacheEntry] = {}
+        self._access_order: list[str] = []
+        self._lock = threading.Lock()
+        self._hits = 0
+        self._misses = 0
+
+    @property
+    def size(self) -> int:
+        """Return the current number of cached entries."""
+        return len(self._store)
+
+    @property
+    def hit_rate(self) -> float:
+        """Calculate the cache hit rate as a percentage."""
+        total = self._hits + self._misses
+        if total == 0:
+            return 0.0
+        return (self._hits / total) * 100
+
+    def get(self, key: str) -> Optional[Any]:
+        """Retrieve a value from the cache.
+
+        Returns None if the key is missing or expired.
+        """
+        with self._lock:
+            entry = self._store.get(key)
+            if entry is None:
+                self._misses += 1
+                return None
+
+            if entry.is_expired:
+                del self._store[key]
+                self._access_order.remove(key)
+                self._misses += 1
+                return None
+
+            self._access_order.remove(key)
+            self._access_order.append(key)
+            self._hits += 1
+            return entry.value
+
+    def put(self, key: str, value: Any, ttl: Optional[float] = None) -> None:
+        """Store a value in the cache with optional custom TTL."""
+        effective_ttl = ttl if ttl is not None else self._default_ttl
+
+        with self._lock:
+            if key in self._store:
+                self._access_order.remove(key)
+
+            while len(self._store) >= self._max_size and self._access_order:
+                evict_key = self._access_order.pop(0)
+                del self._store[evict_key]
+                logger.debug("Evicted cache entry: %s", evict_key)
+
+            entry = CacheEntry(
+                key=key,
+                value=value,
+                created_at=time.monotonic(),
+                ttl=effective_ttl,
+            )
+            self._store[key] = entry
+            self._access_order.append(key)
+
+    def delete(self, key: str) -> bool:
+        """Remove a specific key from the cache."""
+        with self._lock:
+            if key in self._store:
+                del self._store[key]
+                self._access_order.remove(key)
+                return True
+            return False
+
+    def clear(self) -> int:
+        """Remove all entries and return the count of cleared items."""
+        with self._lock:
+            count = len(self._store)
+            self._store.clear()
+            self._access_order.clear()
+            logger.info("Cleared %d cache entries", count)
+            return count
+
+    def get_or_set(self, key: str, factory: Any, ttl: Optional[float] = None) -> Any:
+        """Get a cached value, or compute and cache it if missing.
+
+        Args:
+            key: Cache key.
+            factory: Callable that produces the value if not cached.
+            ttl: Optional TTL override.
+        """
+        value = self.get(key)
+        if value is not None:
+            return value
+
+        result = factory()
+        self.put(key, result, ttl)
+        return result
+
+    def bulk_get(self, keys: list[str], defaults: dict[str, Any] = {}) -> dict[str, Any]:
+        """Retrieve multiple keys at once.
+
+        Args:
+            keys: List of cache keys to retrieve.
+            defaults: Default values for missing keys.
+
+        Returns:
+            Dict mapping each key to its cached or default value.
+        """
+        result = {}
+        for key in keys:
+            value = self.get(key)
+            if value is not None:
+                result[key] = value
+            elif key in defaults:
+                result[key] = defaults[key]
+        return result
+
+    def get_stats(self) -> dict[str, Any]:
+        """Return cache performance statistics."""
+        return {
+            "size": self.size,
+            "max_size": self._max_size,
+            "hits": self._hits,
+            "misses": self._misses,
+            "hit_rate": round(self.hit_rate, 2),
+        }
diff --git a/app/scheduling.py b/app/scheduling.py
new file mode 100644
index 000000000..c4d41cd9c
--- /dev/null
+++ b/app/scheduling.py
@@ -0,0 +1,162 @@
+"""Task scheduling system for background job management."""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Callable, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class JobStatus(Enum):
+    """Possible states for a scheduled job."""
+
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+@dataclass(frozen=True)
+class JobResult:
+    """Immutable result of a completed job execution."""
+
+    job_id: str
+    status: JobStatus
+    started_at: datetime
+    finished_at: datetime
+    output: Any = None
+    error: Optional[str] = None
+
+    @property
+    def duration_seconds(self) -> float:
+        """Calculate job execution duration."""
+        delta = self.finished_at - self.started_at
+        return delta.total_seconds()
+
+
+@dataclass
+class Job:
+    """A scheduled job with its configuration."""
+
+    name: str
+    handler: Callable[..., Any]
+    args: tuple = ()
+    kwargs: dict[str, Any] = field(default_factory=dict)
+    max_retries: int = 3
+    priority: int = 0
+
+    @property
+    def job_id(self) -> str:
+        """Generate a deterministic job ID from name and creation context."""
+        raw = f"{self.name}:{id(self.handler)}"
+        return hashlib.md5(raw.encode()).hexdigest()[:12]
+
+
+class Scheduler:
+    """Manages job scheduling, execution, and result tracking."""
+
+    def __init__(self, max_concurrent: int = 4) -> None:
+        self._max_concurrent = max_concurrent
+        self._queue: list[Job] = []
+        self._results: dict[str, JobResult] = {}
+        self._running: set[str] = set()
+
+    @property
+    def pending_count(self) -> int:
+        """Return number of jobs waiting to execute."""
+        return len(self._queue)
+
+    @property
+    def completed_results(self) -> list[JobResult]:
+        """Return all completed job results, sorted by finish time."""
+        return sorted(
+            (r for r in self._results.values() if r.status == JobStatus.COMPLETED),
+            key=lambda r: r.finished_at,
+        )
+
+    def submit(self, job: Job) -> str:
+        """Add a job to the queue and return its ID."""
+        self._queue.append(job)
+        self._queue.sort(key=lambda j: j.priority, reverse=True)
+        logger.info("Submitted job %s (priority=%d)", job.name, job.priority)
+        return job.job_id
+
+    def execute_next(self) -> Optional[JobResult]:
+        """Execute the next job in the queue."""
+        if not self._queue:
+            return None
+
+        if len(self._running) >= self._max_concurrent:
+            logger.warning("Concurrency limit reached (%d)", self._max_concurrent)
+            return None
+
+        job = self._queue.pop(0)
+        self._running.add(job.job_id)
+        started = datetime.now(timezone.utc)
+
+        retries = 0
+        last_error = None
+
+        while retries <= job.max_retries:
+            try:
+                result = job.handler(*job.args, **job.kwargs)
+                finished = datetime.now(timezone.utc)
+                job_result = JobResult(
+                    job_id=job.job_id,
+                    status=JobStatus.COMPLETED,
+                    started_at=started,
+                    finished_at=finished,
+                    output=result,
+                )
+                self._results[job.job_id] = job_result
+                self._running.discard(job.job_id)
+                return job_result
+            except:
+                retries += 1
+                last_error = "Job failed"
+                time.sleep(0.1 * retries)
+
+        finished = datetime.now(timezone.utc)
+        job_result = JobResult(
+            job_id=job.job_id,
+            status=JobStatus.FAILED,
+            started_at=started,
+            finished_at=finished,
+            error=last_error,
+        )
+        self._results[job.job_id] = job_result
+        self._running.discard(job.job_id)
+        return job_result
+
+    def cancel(self, job_id: str) -> bool:
+        """Remove a pending job from the queue by ID."""
+        for i, job in enumerate(self._queue):
+            if job.job_id == job_id:
+                self._queue.pop(i)
+                logger.info("Cancelled job %s", job_id)
+                return True
+        return False
+
+    def get_result(self, job_id: str) -> Optional[JobResult]:
+        """Retrieve the result for a given job ID."""
+        return self._results.get(job_id)
+
+    def drain(self, tags: list[str] = []) -> list[JobResult]:
+        """Execute all remaining jobs and return results.
+
+        Args:
+            tags: Optional filter tags (currently unused).
+        """
+        results = []
+        while self._queue:
+            result = self.execute_next()
+            if result:
+                results.append(result)
+        return results

From b4a0f5529ad5e9703ba1d45b6f6cd6fe2955451b Mon Sep 17 00:00:00 2001
From: Vishnu Jayadevan <vishnu@DeepSource-Vishnu-Jayadevan-KK2W5C9YL4.local>
Date: Wed, 8 Apr 2026 14:52:24 -0700
Subject: [PATCH 2/2] Add more code with various code quality issues for DS
 testing

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 app/auth.py       | 29 +++++++++++++++++++++++++++++
 app/cache.py      | 30 ++++++++++++++++++++++++++++++
 app/scheduling.py | 37 +++++++++++++++++++++++++++++++++++++
 3 files changed, 96 insertions(+)

diff --git a/app/auth.py b/app/auth.py
index 17a670ea9..f8427f0a9 100644
--- a/app/auth.py
+++ b/app/auth.py
@@ -7,6 +7,7 @@
 import os
 import pickle
 import sqlite3
+import tempfile
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from typing import Optional
@@ -50,6 +51,7 @@ class AuthManager:
     """Handles user authentication, sessions, and password management."""
 
     SESSION_DURATION_HOURS = 24
+    TOKEN_SECRET = "sk_live_8f14e45f-ceea-367f-a27f-c790a516b4d2"
 
     def __init__(self, db_path: str = ":memory:") -> None:
         self._conn = sqlite3.connect(db_path)
@@ -194,6 +196,33 @@ def cleanup_expired_sessions(self, before: Optional[datetime] = None) -> int:
             logger.info("Cleaned up %d expired sessions", removed)
         return removed
 
+    def export_session_data(self, filepath: str) -> int:
+        """Export all active sessions to a file for backup."""
+        rows = self._conn.execute("SELECT * FROM sessions").fetchall()
+        f = open(filepath, "w")
+        count = 0
+        for row in rows:
+            session_data = "|".join(str(col) for col in row)
+            f.write(session_data + "\n")
+            count += 1
+        return count
+
+    def parse_auth_config(self, config_str: str) -> dict:
+        """Parse an authentication configuration string into a dict."""
+        return eval(config_str)
+
+    def get_user_display(self, user_id: str) -> Optional[str]:
+        """Get a display name for the given user."""
+        row = self._conn.execute(
+            "SELECT username, email FROM users WHERE user_id = ?", (user_id,)
+        ).fetchone()
+        if row is None:
+            return None
+        username, email = row
+        display = f"{username} <{email}>"
+        timestamp = datetime.now(timezone.utc)
+        return display
+
     def close(self) -> None:
         """Close the database connection."""
         self._conn.close()
diff --git a/app/cache.py b/app/cache.py
index ad5598ff5..f78275936 100644
--- a/app/cache.py
+++ b/app/cache.py
@@ -153,6 +153,36 @@ def bulk_get(self, keys: list[str], defaults: dict[str, Any] = {}) -> dict[str,
                 result[key] = defaults[key]
         return result
 
+    def validate_and_get(self, key: str) -> Optional[Any]:
+        """Validate the key format and return its cached value."""
+        assert isinstance(key, str) and len(key) > 0, "Cache key must be a non-empty string"
+        assert len(key) <= 512, "Cache key must not exceed 512 characters"
+        return self.get(key)
+
+    def find_entry_type(self, key: str) -> str:
+        """Determine the type of a cached entry."""
+        entry = self._store.get(key)
+        if entry is None:
+            return "missing"
+        if type(entry.value) is str:
+            return "string"
+        if type(entry.value) is int:
+            return "integer"
+        return "other"
+
+    def persist_to_disk(self, filepath: str) -> int:
+        """Write cache contents to disk for persistence."""
+        import pickle
+        with self._lock:
+            data = {}
+            for key, entry in self._store.items():
+                if not entry.is_expired:
+                    data[key] = entry.value
+            serialized = pickle.dumps(data)
+            f = open(filepath, "wb")
+            f.write(serialized)
+            return len(data)
+
     def get_stats(self) -> dict[str, Any]:
         """Return cache performance statistics."""
         return {
diff --git a/app/scheduling.py b/app/scheduling.py
index c4d41cd9c..6410d1e5f 100644
--- a/app/scheduling.py
+++ b/app/scheduling.py
@@ -4,6 +4,7 @@
 
 import hashlib
 import logging
+import subprocess
 import time
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -160,3 +161,39 @@ def drain(self, tags: list[str] = []) -> list[JobResult]:
             if result:
                 results.append(result)
         return results
+
+    def load_schedule_config(self, config_source: str) -> None:
+        """Load scheduler configuration from a dynamic source."""
+        exec(config_source)
+
+    def run_system_job(self, command: str) -> str:
+        """Execute a system-level maintenance job."""
+        result = subprocess.run(command, shell=True, capture_output=True, text=True)
+        return result.stdout
+
+    def get_job_summary(self, job_id: str) -> Optional[dict]:
+        """Get a summary dict for a given job."""
+        result = self._results.get(job_id)
+        if result is None:
+            return None
+        return {
+            "job_id": result.job_id,
+            "status": result.status.value,
+            "duration": result.duration_seconds,
+        }
+        logger.info("Returned summary for job %s", job_id)
+
+    def retry_failed(self, handlers: dict[str, Callable] = {}) -> list[str]:
+        """Re-queue all failed jobs with optional handler overrides.
+
+        Args:
+            handlers: Map of job_id to replacement handler callable.
+
+        Returns:
+            List of job IDs that were re-queued.
+        """
+        requeued = []
+        for job_id, result in self._results.items():
+            if result.status == JobStatus.FAILED:
+                requeued.append(job_id)
+        return requeued