Allow database override for reinjecting failed API calls (#660)

The database is set to default to the one in the security configuration in parts of the feedback process.
To make it easier to use the API call reinjection these should take the database as input instead, and only the overall feedback_callback function should have the default.

A similar problem occurs with the machine configuration which will need to be set when using the API reinjection for now.

One naming issue is we import the database spec as db into feedback.py then have _db as a session object. This might be confusing.

Author: stephen-riggs

src/murfey/cli/repost_failed_calls.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 from queue import Empty, Queue
 
-from sqlmodel import Session
+from sqlmodel import Session, create_engine
 from workflows.transport.pika_transport import PikaTransport
 
 import murfey.server.api.auth
@@ -25,7 +25,7 @@
 import murfey.server.api.session_info
 import murfey.server.api.websocket
 import murfey.server.api.workflow
-from murfey.server.murfey_db import get_murfey_db_session
+from murfey.server.murfey_db import url
 from murfey.util.config import security_from_file
 
 
@@ -162,7 +162,10 @@ def run():
     - feedback messages that can be sent back to rabbitmq
     """
     parser = argparse.ArgumentParser(
-        description="Purge and reinject failed murfey messages"
+        description=(
+            "Purge and reinject failed murfey messages. "
+            "Provide security configuration and set machine configuration."
+        )
     )
     parser.add_argument(
         "-c",
@@ -177,7 +180,6 @@ def run():
 
     # Read the security config file
     security_config = security_from_file(args.config)
-    murfey_db = get_murfey_db_session(security_config)
 
     # Purge the queue and repost/reinject any messages found
     dlq_dump_path = Path(args.dir)
@@ -187,7 +189,14 @@ def run():
         security_config.feedback_queue,
         security_config.rabbitmq_credentials,
     )
-    handle_failed_posts(exported_messages, murfey_db)
+
+    # Set up database and retry api calls
+    _url = url(security_config)
+    engine = create_engine(_url)
+    with Session(engine) as murfey_db:
+        handle_failed_posts(exported_messages, murfey_db)
+
+    # Reinject all remaining messages to rabbitmq
     handle_dlq_messages(exported_messages, security_config.rabbitmq_credentials)
 
     # Clean up any created directories

src/murfey/server/api/workflow.py

@@ -768,7 +768,7 @@ def register_completed_tilt_series(
     db.commit()
     for ts in tilt_series_db:
         if (
-            check_tilt_series_mc(ts.id)
+            check_tilt_series_mc(ts.id, db)
             and not ts.processing_requested
             and ts.tilt_series_length > 2
         ):
@@ -795,9 +795,9 @@ def register_completed_tilt_series(
             machine_config = get_machine_config(instrument_name=instrument_name)[
                 instrument_name
             ]
-            tilts = get_all_tilts(ts.id)
-            ids = get_job_ids(ts.id, collected_ids[3].id)
-            preproc_params = get_tomo_proc_params(ids.dcgid)
+            tilts = get_all_tilts(ts.id, db)
+            ids = get_job_ids(ts.id, collected_ids[3].id, db)
+            preproc_params = get_tomo_proc_params(ids.dcgid, db)
 
             first_tilt = db.exec(
                 select(Tilt).where(Tilt.tilt_series_id == ts.id)

src/murfey/server/demo_api.py

@@ -748,7 +748,7 @@ def register_dc_group(
 
     if dcg_params.atlas:
         _flush_grid_square_records(
-            {"session_id": session_id, "tag": dcg_params.tag}, demo=True
+            {"session_id": session_id, "tag": dcg_params.tag}, _db=db, demo=True
         )
     return dcg_params