Centralize privacy/deletion handler registration and verification.

src/codehelp/__init__.py

@@ -9,7 +9,7 @@
 
 from gened import base
 
-from . import admin, context_config, helper, privacy, tutor
+from . import admin, context_config, deletion_handler, helper, tutor
 
 
 def create_app(test_config: dict[str, Any] | None = None, instance_path: Path | None = None) -> Flask:
@@ -42,7 +42,7 @@ def create_app(test_config: dict[str, Any] | None = None, instance_path: Path |
         app_config = app_config | test_config
 
     # register app-specific functionality with gened
-    privacy.register_with_gened()
+    deletion_handler.register_with_gened()
     admin.register_with_gened()
 
     # create the base application

src/codehelp/deletion_handler.py

@@ -0,0 +1,117 @@
+# SPDX-FileCopyrightText: 2024 Mark Liffiton <[email protected]>
+#
+# SPDX-License-Identifier: AGPL-3.0-only
+
+"""Implementation of personal data deletion for CodeHelp."""
+
+from gened.data_deletion import register_handler
+from gened.db import get_db
+
+
+class CodeHelpDeletionHandler:
+    """CodeHelp implementation of personal data deletion."""
+
+    def delete_user_data(self, user_id: int) -> None:
+        """Delete/Anonymize personal data for a user while preserving non-personal data for analysis."""
+        db = get_db()
+
+        # Anonymize personal data in queries
+        db.execute("""
+            UPDATE queries
+            SET code = CASE
+                    WHEN code IS NOT NULL THEN '[deleted]'
+                    ELSE NULL
+                END,
+                error = CASE
+                    WHEN error IS NOT NULL THEN '[deleted]'
+                    ELSE NULL
+                END,
+                issue = '[deleted]',
+                context_name = '[deleted]',
+                context_string_id = NULL,
+                user_id = -1
+            WHERE user_id = ?
+        """, [user_id])
+
+        # Anonymize personal data in chats
+        db.execute("""
+            UPDATE chats
+            SET topic = '[deleted]',
+                chat_json = '[]',
+                context_name = '[deleted]',
+                context_string_id = NULL,
+                user_id = -1
+            WHERE user_id = ?
+        """, [user_id])
+
+        db.commit()
+
+    def delete_class_data(self, class_id: int) -> None:
+        """Delete/Anonymize personal data for a class while preserving non-personal data for analysis."""
+        db = get_db()
+
+        # Remove context names and configs as they may contain personal information
+        db.execute("""
+            UPDATE contexts
+            SET name = '[deleted]' || id,
+                config = '{}'
+            WHERE class_id = ?
+        """, [class_id])
+
+        # Remove context strings as they may contain personal information
+        db.execute("""
+            DELETE FROM context_strings
+            WHERE id IN (
+                SELECT context_string_id
+                FROM queries
+                WHERE role_id IN (
+                    SELECT id FROM roles WHERE class_id = ?
+                )
+                UNION
+                SELECT context_string_id
+                FROM chats
+                WHERE role_id IN (
+                    SELECT id FROM roles WHERE class_id = ?
+                )
+            )
+        """, [class_id, class_id])
+
+        # Anonymize personal data in queries
+        db.execute("""
+            UPDATE queries
+            SET code = CASE
+                    WHEN code IS NOT NULL THEN '[deleted]'
+                    ELSE NULL
+                END,
+                error = CASE
+                    WHEN error IS NOT NULL THEN '[deleted]'
+                    ELSE NULL
+                END,
+                issue = '[deleted]',
+                context_name = '[deleted]',
+                context_string_id = NULL,
+                user_id = -1
+            WHERE role_id IN (
+                SELECT id FROM roles WHERE class_id = ?
+            )
+        """, [class_id])
+
+        # Anonymize personal data in chats
+        db.execute("""
+            UPDATE chats
+            SET topic = '[deleted]',
+                chat_json = '[]',
+                context_name = '[deleted]',
+                context_string_id = NULL,
+                user_id = -1
+            WHERE role_id IN (
+                SELECT id FROM roles WHERE class_id = ?
+            )
+        """, [class_id])
+
+        db.commit()
+
+
+def register_with_gened() -> None:
+    """Register CodeHelp deletion handler with the gened framework."""
+    register_handler(CodeHelpDeletionHandler())

src/gened/base.py

@@ -20,6 +20,7 @@
     auth,
     class_config,
     classes,
+    data_deletion,
     db,
     demo,
     docs,
@@ -164,10 +165,14 @@ def create_app_base(import_name: str, app_config: dict[str, Any], instance_path:
     # configure the application
     app.config.from_mapping(total_config)
 
+    # verify deletion handler is registered
+    if data_deletion.get_handler() is None:
+        app.logger.error("No deletion handler registered. All Gen-Ed applications must provide one.")
+        sys.exit(1)
+
     admin.init_app(app)
     db.init_app(app)
     filters.init_app(app)
-    instructor.init_app(app)  # This will verify data deletion handler is registered
     migrate.init_app(app)
     oauth.init_app(app)
     tz.init_app(app)

src/gened/data_deletion.py

@@ -0,0 +1,51 @@
+# SPDX-FileCopyrightText: 2024 Mark Liffiton <[email protected]>
+#
+# SPDX-License-Identifier: AGPL-3.0-only
+
+"""Interface and registry for personal data deletion handlers.
+
+This module defines the interface that Gen-Ed applications must implement
+for handling personal data deletion and provides the registration mechanism
+for those handlers.
+"""
+
+from typing import Protocol
+
+
+class DeletionHandler(Protocol):
+    """Protocol defining the interface for personal data deletion handlers."""
+    def delete_user_data(self, user_id: int) -> None:
+        """Delete/anonymize all personal data for the given user."""
+        ...
+
+    def delete_class_data(self, class_id: int) -> None:
+        """Delete/anonymize all personal data for the given class."""
+        ...
+
+
+_handler: DeletionHandler | None = None
+
+
+def register_handler(handler: DeletionHandler) -> None:
+    """Register the application's deletion handler implementation."""
+    global _handler
+    _handler = handler
+
+
+def get_handler() -> DeletionHandler | None:
+    """Get the registered deletion handler."""
+    return _handler
+
+
+def delete_user_data(user_id: int) -> None:
+    """Delete/anonymize all personal data for the given user."""
+    if _handler is None:
+        raise RuntimeError("No deletion handler registered")
+    _handler.delete_user_data(user_id)
+
+
+def delete_class_data(class_id: int) -> None:
+    """Delete/anonymize all personal data for the given class."""
+    if _handler is None:
+        raise RuntimeError("No deletion handler registered")
+    _handler.delete_class_data(class_id)

src/gened/instructor.py

@@ -14,13 +14,10 @@
 """
 
 import datetime as dt
-import sys
-from collections.abc import Callable
 from sqlite3 import Row
 
 from flask import (
     Blueprint,
-    Flask,
     abort,
     flash,
     redirect,
@@ -30,30 +27,13 @@
 )
 from werkzeug.wrappers.response import Response
 
-from .auth import get_auth, get_auth_class, instructor_required
+from .auth import get_auth_class, instructor_required
 from .classes import switch_class
 from .csv import csv_response
+from .data_deletion import delete_class_data
 from .db import get_db
 from .redir import safe_redirect
 
-DataDeletionHandler = Callable[[int], None]
-
-# Register the handler for the current application
-_deletion_handlers: list[DataDeletionHandler] = []
-
-
-def register_class_deletion_handler(handler: DataDeletionHandler) -> None:
-    """Register the application's class deletion implementation"""
-    _deletion_handlers.append(handler)
-
-
-def init_app(app: Flask) -> None:
-    """Initialize the instructor module and verify deletion handler is registered"""
-    if not _deletion_handlers:
-        app.logger.error("No data deletion handler registered. All Gen-Ed applications must provide one.")
-        sys.exit(1)
-
-
 bp = Blueprint('instructor', __name__, url_prefix="/instructor", template_folder='templates')
 
 @bp.before_request
@@ -249,9 +229,7 @@ def delete_class() -> Response:
     assert str(class_id) == str(request.form.get('class_id'))
 
     # Call application-specific data deletion handler(s)
-    assert _deletion_handlers  # checked during init
-    for handler in _deletion_handlers:
-        handler(class_id)
+    delete_class_data(class_id)
 
     # Deactivate all roles and disable the class
     db.execute("UPDATE roles SET user_id=-1, active = 0 WHERE class_id = ?", [class_id])