fix: gbfs validator scheduler failure (#984)

Author: cka-y

functions-python/gbfs_validator/src/gbfs_utils.py

@@ -12,7 +12,9 @@
     Gbfsvalidationreport,
     Gbfsnotice,
 )
+from sqlalchemy.orm import Session
 from shared.dataset_service.main import Status
+from shared.helpers.database import with_db_session
 
 
 class GBFSValidator:
@@ -107,8 +109,9 @@ def validate_gbfs_feed(self, bucket: storage.Bucket) -> Dict[str, Any]:
         }
 
 
+@with_db_session(echo=False)
 def save_snapshot_and_report(
-    session, snapshot: Gbfssnapshot, validation_result: Dict[str, Any]
+    snapshot: Gbfssnapshot, validation_result: Dict[str, Any], db_session: Session
 ):
     """Save the snapshot and validation report to the database."""
     validation_report = Gbfsvalidationreport(
@@ -132,8 +135,8 @@ def save_snapshot_and_report(
         for error in file_summary["groupedErrors"]
     ]
     snapshot.gbfsvalidationreports = [validation_report]
-    session.add(snapshot)
-    session.commit()
+    db_session.add(snapshot)
+    db_session.commit()
 
 
 def fetch_gbfs_files(url: str) -> Dict[str, Any]:

functions-python/gbfs_validator/src/main.py

@@ -18,7 +18,7 @@
     PipelineStage,
     MaxExecutionsReachedError,
 )
-from shared.helpers.database import Database, with_db_session
+from shared.helpers.database import with_db_session
 from shared.helpers.logger import Logger, StableIdFilter
 from shared.helpers.parser import jsonify_pubsub
 from gbfs_utils import (
@@ -33,12 +33,13 @@
 BUCKET_NAME = os.getenv("BUCKET_NAME", "mobilitydata-gbfs-snapshots-dev")
 
 
-@with_db_session
+@with_db_session(echo=False)
 def fetch_all_gbfs_feeds(db_session: "Session") -> List[Gbfsfeed]:
     try:
         gbfs_feeds = (
             db_session.query(Gbfsfeed).options(joinedload(Gbfsfeed.gbfsversions)).all()
         )
+        db_session.expunge_all()
         return gbfs_feeds
     except Exception as e:
         logging.error(f"Error fetching all GBFS feeds: {e}")
@@ -103,9 +104,7 @@ def gbfs_validator_pubsub(cloud_event: CloudEvent):
         try:
             snapshot = validator.create_snapshot(feed_id)
             validation_results = validator.validate_gbfs_feed(bucket)
-            db = Database(database_url=os.getenv("FEEDS_DATABASE_URL"))
-            with db.start_db_session() as session:
-                save_snapshot_and_report(session, snapshot, validation_results)
+            save_snapshot_and_report(snapshot, validation_results)
         except Exception as e:
             error_message = f"Error validating GBFS feed: {e}"
             logging.error(f"{error_message}\nTraceback:\n{traceback.format_exc()}")

functions-python/gbfs_validator/tests/test_gbfs_utils.py

@@ -175,7 +175,9 @@ def test_save_snapshot_and_report(self, mock_gbfsnotice, mock_gbfsvalidationrepo
             },
         }
 
-        save_snapshot_and_report(mock_session, mock_snapshot, validation_result)
+        save_snapshot_and_report(
+            mock_snapshot, validation_result, db_session=mock_session
+        )
 
         mock_session.add.assert_called_once_with(mock_snapshot)
         mock_session.commit.assert_called_once()

functions-python/gbfs_validator/tests/test_gbfs_validator.py

@@ -33,7 +33,6 @@ def tearDown(self) -> None:
             "VALIDATOR_URL": "https://mock-validator-url.com",
         },
     )
-    @patch("main.Database")
     @patch("main.DatasetTraceService")
     @patch("main.fetch_gbfs_files")
     @patch("main.GBFSValidator.create_gbfs_json_with_bucket_paths")
@@ -52,12 +51,8 @@ def test_gbfs_validator_pubsub(
         mock_create_gbfs_json,
         mock_fetch_gbfs_files,
         mock_dataset_trace_service,
-        mock_database,
     ):
         # Prepare mocks
-        mock_session = MagicMock()
-        mock_database.return_value.start_db_session.return_value = mock_session
-
         mock_trace_service = MagicMock()
         mock_dataset_trace_service.return_value = mock_trace_service