refactor: shared session load path for session and export APIs

api/_session_handlers.py

@@ -0,0 +1,94 @@
+"""Shared resolve/load/exclude/error helpers for session API handlers."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from flask import current_app
+
+from api._flask_types import FlaskReturn
+from api.error_codes import ErrorCode, error_response
+from models.session import SessionDict
+from models.stats import SessionStatsDict
+from utils.exclusion_rules import is_session_excluded
+from utils.session_cache import get_cached_session
+from utils.session_errors import SESSION_LOAD_ERRORS
+from utils.session_path import get_claude_projects_dir, safe_join
+from utils.session_stats import compute_stats
+
+__all__ = [
+    "SESSION_LOAD_ERRORS",
+    "LoadedSession",
+    "resolve_loaded_session",
+    "compute_stats_or_error",
+]
+
+
+@dataclass(frozen=True)
+class LoadedSession:
+    session: SessionDict
+    filepath: str
+
+
+def resolve_loaded_session(
+    project_name: str,
+    session_id: str,
+    *,
+    missing_file_message: str | Callable[[str], str],
+    parse_log_action: str = "Failed to parse session %s",
+) -> LoadedSession | FlaskReturn:
+    """Resolve path, load session, and apply exclusion rules.
+
+    Returns ``LoadedSession`` on success or an ``error_response`` tuple/Response.
+    """
+    base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
+    try:
+        filepath = safe_join(base, project_name, f"{session_id}.jsonl")
+    except ValueError:
+        return error_response(ErrorCode.INVALID_PATH, "Invalid path", 400)
+
+    if not os.path.isfile(filepath):
+        msg = (
+            missing_file_message(session_id)
+            if callable(missing_file_message)
+            else missing_file_message
+        )
+        return error_response(ErrorCode.SESSION_NOT_FOUND, msg, 404)
+
+    try:
+        session = get_cached_session(filepath)
+        rules = current_app.config.get("EXCLUSION_RULES") or []
+        if is_session_excluded(rules, session, project_name):
+            return error_response(
+                ErrorCode.SESSION_NOT_FOUND,
+                "Session not found",
+                404,
+            )
+    except SESSION_LOAD_ERRORS:
+        current_app.logger.exception(parse_log_action, session_id)
+        return error_response(
+            ErrorCode.PARSE_ERROR,
+            "Failed to parse session",
+            500,
+        )
+
+    return LoadedSession(session=session, filepath=filepath)
+
+
+def compute_stats_or_error(
+    session: SessionDict,
+    session_id: str,
+    *,
+    log_action: str,
+) -> SessionStatsDict | FlaskReturn:
+    try:
+        return compute_stats(session)
+    except SESSION_LOAD_ERRORS:
+        current_app.logger.exception(log_action, session_id)
+        return error_response(
+            ErrorCode.INTERNAL_ERROR,
+            "Failed to compute session stats",
+            500,
+        )

api/export_api.py

@@ -2,23 +2,21 @@
 
 import io
 import json
-import os
 import zipfile
 from datetime import datetime
 from typing import Any
 
 from flask import Blueprint, current_app, request, send_file
 
 from api._flask_types import FlaskReturn, json_response
+from api._session_handlers import (
+    LoadedSession,
+    compute_stats_or_error,
+    resolve_loaded_session,
+)
 from api.error_codes import ErrorCode, error_response
 from models.export import ExportStateDict
-from utils.exclusion_rules import is_session_excluded
-from utils.export_engine import (
-    EXPORT_ERRORS as _EXPORT_ERRORS,
-    ExportFailure,
-    ZipSink,
-    run_bulk_export,
-)
+from utils.export_engine import ExportFailure, ZipSink, run_bulk_export
 from utils.export_state_store import (
     EXPORT_STATE_FILE,
     atomic_write_export_state,
@@ -27,9 +25,7 @@
 )
 from utils.json_exporter import session_to_json
 from utils.md_exporter import session_to_markdown
-from utils.session_cache import get_cached_session
-from utils.session_path import get_claude_projects_dir, list_projects, safe_join
-from utils.session_stats import compute_stats
+from utils.session_path import get_claude_projects_dir, list_projects
 from utils.slugify import slugify
 
 export_bp = Blueprint("export", __name__)
@@ -222,67 +218,41 @@ def _on_export_error(sid: str, exc: Exception) -> None:
 
 @export_bp.route("/api/export/session/<path:project_name>/<session_id>")
 def export_session(project_name: str, session_id: str) -> FlaskReturn:
-    base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
-    try:
-        filepath = safe_join(base, project_name, f"{session_id}.jsonl")
-    except ValueError:
-        return error_response(ErrorCode.INVALID_PATH, "Invalid path", 400)
-
-    if not os.path.isfile(filepath):
-        return error_response(
-            ErrorCode.SESSION_NOT_FOUND,
-            "Session not found",
-            404,
-        )
-
-    fmt = request.args.get("format", "md")
-    try:
-        session = get_cached_session(filepath)
-    except _EXPORT_ERRORS:
-        current_app.logger.exception("Failed to parse session %s for export", session_id)
-        return error_response(
-            ErrorCode.PARSE_ERROR,
-            "Failed to parse session",
-            500,
-        )
-
-    rules = current_app.config.get("EXCLUSION_RULES") or []
-    if is_session_excluded(rules, session, project_name):
-        return error_response(
-            ErrorCode.SESSION_NOT_FOUND,
-            "Session not found",
-            404,
-        )
-
-    try:
-        stats = compute_stats(session)
-    except _EXPORT_ERRORS:
-        current_app.logger.exception("Failed to compute stats for export %s", session_id)
-        return error_response(
-            ErrorCode.INTERNAL_ERROR,
-            "Failed to compute session stats",
-            500,
-        )
-
-    title_slug = slugify(session["title"], default="session")
-
-    if fmt == "json":
-        content = session_to_json(session, stats)
-        buf = io.BytesIO(content.encode("utf-8"))
-        buf.seek(0)
-        return send_file(
-            buf,
-            mimetype="application/json",
-            as_attachment=True,
-            download_name=f"{title_slug}.json",  # type: ignore[call-arg]
-        )
-
-    md = session_to_markdown(session, stats)
-    buf = io.BytesIO(md.encode("utf-8"))
-    buf.seek(0)
-    return send_file(
-        buf,
-        mimetype="text/markdown",
-        as_attachment=True,
-        download_name=f"{title_slug}.md",  # type: ignore[call-arg]
+    loaded = resolve_loaded_session(
+        project_name,
+        session_id,
+        missing_file_message="Session not found",
+        parse_log_action="Failed to parse session %s for export",
     )
+    if isinstance(loaded, LoadedSession):
+        fmt = request.args.get("format", "md")
+        stats = compute_stats_or_error(
+            loaded.session,
+            session_id,
+            log_action="Failed to compute stats for export %s",
+        )
+        if isinstance(stats, dict):
+            title_slug = slugify(loaded.session["title"], default="session")
+
+            if fmt == "json":
+                content = session_to_json(loaded.session, stats)
+                buf = io.BytesIO(content.encode("utf-8"))
+                buf.seek(0)
+                return send_file(
+                    buf,
+                    mimetype="application/json",
+                    as_attachment=True,
+                    download_name=f"{title_slug}.json",  # type: ignore[call-arg]
+                )
+
+            md = session_to_markdown(loaded.session, stats)
+            buf = io.BytesIO(md.encode("utf-8"))
+            buf.seek(0)
+            return send_file(
+                buf,
+                mimetype="text/markdown",
+                as_attachment=True,
+                download_name=f"{title_slug}.md",  # type: ignore[call-arg]
+            )
+        return stats
+    return loaded

api/sessions.py

@@ -1,101 +1,47 @@
 """Session detail and stats endpoints."""
 
-import json
-import os
-
-from flask import Blueprint, current_app
+from flask import Blueprint
 
 from api._flask_types import FlaskReturn, json_response
-from api.error_codes import ErrorCode, error_response
-from utils.exclusion_rules import is_session_excluded
-from utils.session_cache import get_cached_session
-from utils.session_path import get_claude_projects_dir, safe_join
-from utils.session_stats import compute_stats
+from api._session_handlers import (
+    LoadedSession,
+    compute_stats_or_error,
+    resolve_loaded_session,
+)
 
 sessions_bp = Blueprint("sessions", __name__)
 
-_PARSE_ERRORS = (
-    json.JSONDecodeError,
-    KeyError,
-    ValueError,
-    OSError,
-    FileNotFoundError,
-)
+
+def _missing_session_message(session_id: str) -> str:
+    return f"Session {session_id} not found"
 
 
 @sessions_bp.route("/api/sessions/<path:project_name>/<session_id>")
 def get_session(project_name: str, session_id: str) -> FlaskReturn:
-    base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
-    try:
-        filepath = safe_join(base, project_name, f"{session_id}.jsonl")
-    except ValueError:
-        return error_response(ErrorCode.INVALID_PATH, "Invalid path", 400)
-
-    if not os.path.isfile(filepath):
-        return error_response(
-            ErrorCode.SESSION_NOT_FOUND,
-            f"Session {session_id} not found",
-            404,
-        )
-
-    try:
-        session = get_cached_session(filepath)
-        rules = current_app.config.get("EXCLUSION_RULES") or []
-        if is_session_excluded(rules, session, project_name):
-            return error_response(
-                ErrorCode.SESSION_NOT_FOUND,
-                "Session not found",
-                404,
-            )
-        return json_response(session)
-    except _PARSE_ERRORS:
-        current_app.logger.exception("Failed to parse session %s", session_id)
-        return error_response(
-            ErrorCode.PARSE_ERROR,
-            "Failed to parse session",
-            500,
-        )
+    loaded = resolve_loaded_session(
+        project_name,
+        session_id,
+        missing_file_message=_missing_session_message,
+    )
+    if isinstance(loaded, LoadedSession):
+        return json_response(loaded.session)
+    return loaded
 
 
 @sessions_bp.route("/api/sessions/<path:project_name>/<session_id>/stats")
 def get_session_stats(project_name: str, session_id: str) -> FlaskReturn:
-    base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
-    try:
-        filepath = safe_join(base, project_name, f"{session_id}.jsonl")
-    except ValueError:
-        return error_response(ErrorCode.INVALID_PATH, "Invalid path", 400)
-
-    if not os.path.isfile(filepath):
-        return error_response(
-            ErrorCode.SESSION_NOT_FOUND,
-            f"Session {session_id} not found",
-            404,
-        )
-
-    try:
-        session = get_cached_session(filepath)
-        rules = current_app.config.get("EXCLUSION_RULES") or []
-        if is_session_excluded(rules, session, project_name):
-            return error_response(
-                ErrorCode.SESSION_NOT_FOUND,
-                "Session not found",
-                404,
-            )
-    except _PARSE_ERRORS:
-        current_app.logger.exception("Failed to parse session %s", session_id)
-        return error_response(
-            ErrorCode.PARSE_ERROR,
-            "Failed to parse session",
-            500,
-        )
-
-    try:
-        stats = compute_stats(session)
-        return json_response(stats)
-    except _PARSE_ERRORS:
-        current_app.logger.exception("Failed to compute stats for %s", session_id)
-        return error_response(
-            ErrorCode.INTERNAL_ERROR,
-            "Failed to compute session stats",
-            500,
+    loaded = resolve_loaded_session(
+        project_name,
+        session_id,
+        missing_file_message=_missing_session_message,
+    )
+    if isinstance(loaded, LoadedSession):
+        stats = compute_stats_or_error(
+            loaded.session,
+            session_id,
+            log_action="Failed to compute stats for %s",
         )
+        if isinstance(stats, dict):
+            return json_response(stats)
+        return stats
+    return loaded

tests/test_api_routes.py

@@ -66,7 +66,7 @@ def test_session_detail_parse_failure_returns_500_without_leak(client, monkeypat
     def _boom(*_args, **_kwargs):
         raise KeyError("internal_secret_field_id")
 
-    monkeypatch.setattr("api.sessions.get_cached_session", _boom)
+    monkeypatch.setattr("api._session_handlers.get_cached_session", _boom)
     resp = client.get("/api/sessions/test-project/session_abc123")
     assert resp.status_code == 500
     body_text = resp.get_data(as_text=True)