Delete old (already deleted) consultations after 30 days. (SEA-1459).

setup.cfg

@@ -87,6 +87,7 @@ fanstatic.libraries =
     privatim:css = privatim.static:css_library
 
 console_scripts =
+    data_retention = privatim.cli.apply_data_retention_policy:hard_delete
     add_user = privatim.cli.user:add_user
     # delete_user = privatim.cli.user:delete_user
     add_content = privatim.cli.add_content:main

src/privatim/cli/apply_data_retention_policy.py

@@ -0,0 +1,65 @@
+from datetime import timedelta
+
+import click
+from pyramid.paster import bootstrap
+from pyramid.paster import get_appsettings
+from sedate import utcnow
+from sqlalchemy import delete
+
+from privatim.models import Consultation
+from privatim.models.soft_delete import SoftDeleteMixin
+from privatim.orm import get_engine, Base
+
+
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from sqlalchemy.orm.session import Session
+
+
+KEEP_DELETED_FILES_TIMESPAN = 30
+
+
+def delete_old_records(
+        session: 'Session',
+        model: type[Base],
+        days_threshold: int = 30
+) -> list[str]:
+    if not issubclass(model, SoftDeleteMixin):
+        raise ValueError(f'{model.__name__} does not support soft delete')
+
+    cutoff_date = utcnow() - timedelta(days=days_threshold)
+
+    # Combine the select and delete operations
+    stmt = (
+        delete(model)
+        .where(
+            model.updated <= cutoff_date,  # type:ignore[attr-defined]
+            model.deleted.is_(True)
+        )
+        .returning(model.id)  # type:ignore[attr-defined]
+    )
+    result = session.execute(stmt)
+    deleted_ids = result.scalars().all()
+    session.flush()
+    return deleted_ids
+
+
+@click.command()
+@click.argument('config_uri')
+def hard_delete(
+        config_uri: str,
+) -> None:
+    env = bootstrap(config_uri)
+    settings = get_appsettings(config_uri)
+    engine = get_engine(settings)
+    Base.metadata.create_all(engine)
+
+    with env['request'].tm:
+        session = env['request'].dbsession
+        deleted_consultation_ids = delete_old_records(session, Consultation)
+        for id in deleted_consultation_ids:
+            print(f"Deleted Consultation with ID: {id}")
+
+
+if __name__ == '__main__':
+    hard_delete()

tests/cli/test_data_retention_policy.py

@@ -0,0 +1,102 @@
+import pytest
+from datetime import timedelta
+from sedate import utcnow
+from sqlalchemy import select, Column, Integer
+
+from privatim.cli.apply_data_retention_policy import delete_old_records
+from privatim.models import Consultation, SearchableFile
+from privatim.orm import Base
+
+
+def create_consultation(
+    session, user, title, days_old, is_deleted=False, with_file=False
+):
+    """Helper function to create a consultation"""
+
+    consultation = Consultation(
+        title=title,
+        creator=user,
+        description='Test description',
+        status='Created',
+    )
+
+    if with_file:
+        file = SearchableFile(
+            filename='test.txt',
+            content=b'Test content',
+        )
+        session.add(file)
+        consultation.files.append(file)
+
+    consultation.updated = utcnow() - timedelta(days=days_old)
+    consultation.deleted = is_deleted
+    session.add(consultation)
+    session.flush()
+    return consultation
+
+
+def test_delete_old_records_consultations(session, user):
+    # Create various consultations
+    create_consultation(session, user, 'Recent Active', 10)
+    create_consultation(session, user, 'Old Active', 40)
+    old_deleted = create_consultation(
+        session, user, 'Old Deleted', 40, is_deleted=True
+    )
+    old_deleted_with_file = create_consultation(
+        session,
+        user,
+        'Old Deleted with File',
+        40,
+        is_deleted=True,
+        with_file=True,
+    )
+    create_consultation(session, user, 'Recent Deleted', 10, is_deleted=True)
+
+    session.flush()
+
+    # Run the delete_old_records function
+    deleted_ids = delete_old_records(session, Consultation)
+
+    # Check that the correct consultations were deleted
+    assert len(deleted_ids) == 2
+    assert old_deleted.id in deleted_ids
+    assert old_deleted_with_file.id in deleted_ids
+
+    with session.no_soft_delete_filter():
+        remaining_consultations = session.scalars(select(Consultation)).all()
+        remaining_ids = [c.id for c in remaining_consultations]
+
+        assert len(remaining_consultations) == 3
+        assert old_deleted.id not in remaining_ids
+        assert old_deleted_with_file.id not in remaining_ids
+
+        # Check that associated files were also deleted
+        remaining_files = session.scalars(select(SearchableFile)).all()
+        assert len(remaining_files) == 0
+
+
+def test_delete_old_records_no_deletions(session, user):
+    # Create only recent or active consultations
+    create_consultation(session, user, 'Recent Active', 10)
+    create_consultation(session, user, 'Recent Deleted', 10, is_deleted=True)
+
+    session.flush()
+
+    # Run the delete_old_records function
+    deleted_ids = delete_old_records(session, Consultation)
+
+    # Check that no consultations were deleted
+    assert len(deleted_ids) == 0
+
+    with session.no_soft_delete_filter():
+        remaining_consultations = session.scalars(select(Consultation)).all()
+        assert len(remaining_consultations) == 2
+
+
+def test_delete_old_records_invalid_model(session):
+    class InvalidModel(Base):
+        __tablename__ = 'invalid_model'
+        id = Column(Integer, primary_key=True)
+
+    with pytest.raises(ValueError, match="does not support soft delete"):
+        delete_old_records(session, InvalidModel)