Add ability to delete sessions (#414)

Author: sadhana-r

src/openlifu/db/database.py

@@ -219,6 +219,31 @@ def write_session(self, subject:Subject, session:Session, on_conflict=OnConflict
 
         self.logger.info(f"Added session with ID {session_id} for subject {subject.id} to the database.")
 
+    def delete_session(self, subject_id: str, session_id: str, on_conflict: OnConflictOpts = OnConflictOpts.ERROR):
+
+        # Check if the session ID exists in the database for this subject
+        session_ids = self.get_session_ids(subject_id)
+
+        if session_id not in session_ids:
+            if on_conflict == OnConflictOpts.ERROR:
+                raise ValueError(f"Session ID {session_id} does not exist in the database.")
+            elif on_conflict == OnConflictOpts.SKIP:
+                self.logger.info(f"Cannot delete session ID {session_id} as it does not exist in the database.")
+                return
+            else:
+                raise ValueError("Invalid 'on_conflict' option.")
+
+        # Delete the session file
+        session_filename = self.get_session_filename(subject_id, session_id)
+        if Path.is_file(session_filename):
+            shutil.rmtree(session_filename.parent)
+
+        if session_id in session_ids:
+            session_ids.remove(session_id)
+            self.write_session_ids(subject_id, session_ids)
+
+        self.logger.info(f"Removed Session with ID {session_id} from the database.")
+
     def write_run(self, run:Run, session:Session = None, protocol:Protocol = None, on_conflict=OnConflictOpts.ERROR):
         """Write a run with a snapshot of session and a snapshot of protocol if provided
 

tests/test_database.py

@@ -327,6 +327,35 @@ def test_write_session_with_transducer_tracking_results(example_database: Databa
     session.transducer_tracking_results = [example_transducer_tracking_result]
     example_database.write_session(example_subject, session, on_conflict = OnConflictOpts.OVERWRITE)
 
+def test_delete_session(example_database: Database, example_subject: Subject):
+    # Write a session
+    session = Session(name="bleh", id='a_session',subject_id=example_subject.id)
+
+    session_id = session.id
+    subject_id = example_subject.id
+
+    # Can add a new session, and it loads back in correctly.
+    example_database.write_session(example_subject, session)
+
+    assert session.id in example_database.get_session_ids(example_subject.id)
+
+    # Session is deleted
+    example_database.delete_session(subject_id, session_id)
+    assert session.id not in example_database.get_session_ids(subject_id)
+    with pytest.raises(FileNotFoundError):
+        example_database.load_session(example_subject, session_id)
+
+    # Error option
+    with pytest.raises(ValueError, match="does not exist in the database"):
+        example_database.delete_session(subject_id, "non_existent_protocol", on_conflict=OnConflictOpts.ERROR)
+
+    # Skip option
+    example_database.delete_session(subject_id, "non_existent_protocol", on_conflict=OnConflictOpts.SKIP)
+
+    # Invalid option
+    with pytest.raises(ValueError, match="Invalid"):
+        example_database.delete_session(subject_id, "non_existent_protocol", on_conflict=OnConflictOpts.OVERWRITE)
+
 def test_write_run(example_database: Database, tmp_path:Path):
     subject_id = "example_subject"
     session_id = "example_session"