feat: udpate feed status on validation report (#1122)

Alessandro100 · web-flow · commit cdaec70fb028 · 2025-04-17T08:44:42.000-04:00
* udpate feed status on validation report
diff --git a/functions-python/helpers/feed_status.py b/functions-python/helpers/feed_status.py
@@ -0,0 +1,86 @@
+import logging
+from datetime import datetime, timezone
+from sqlalchemy import text
+from shared.database_gen.sqlacodegen_models import Gtfsdataset, Feed, t_feedsearch
+from shared.database.database import refresh_materialized_view
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+
+#  query to update the status of the feeds based on the service date range of the latest dataset
+def update_feed_statuses_query(session: "Session", stable_feed_ids: list[str]):
+    today_utc = datetime.now(timezone.utc).date()
+
+    latest_dataset_subq = (
+        session.query(
+            Gtfsdataset.feed_id,
+            Gtfsdataset.service_date_range_start,
+            Gtfsdataset.service_date_range_end,
+        )
+        .filter(
+            Gtfsdataset.latest.is_(True),
+            Gtfsdataset.service_date_range_start.isnot(None),
+            Gtfsdataset.service_date_range_end.isnot(None),
+        )
+        .subquery()
+    )
+
+    status_conditions = [
+        (
+            latest_dataset_subq.c.service_date_range_end < today_utc,
+            "inactive",
+        ),
+        (
+            latest_dataset_subq.c.service_date_range_start > today_utc,
+            "future",
+        ),
+        (
+            (latest_dataset_subq.c.service_date_range_start <= today_utc)
+            & (latest_dataset_subq.c.service_date_range_end >= today_utc),
+            "active",
+        ),
+    ]
+
+    try:
+        diff_counts: dict[str, int] = {}
+
+        def get_filters(status: str):
+            filters = [
+                Feed.id == latest_dataset_subq.c.feed_id,
+                Feed.status != text("'deprecated'::status"),
+                Feed.status != text("'development'::status"),
+                # We filter out feeds that already have the status so that the
+                # update count reflects the number of feeds that actually
+                # changed status.
+                Feed.status != text("'%s'::status" % status),
+                service_date_conditions,
+            ]
+
+            if len(stable_feed_ids) > 0:
+                filters.insert(0, Feed.stable_id.in_(stable_feed_ids))
+
+            return filters
+
+        for service_date_conditions, status in status_conditions:
+            diff_counts[status] = (
+                session.query(Feed)
+                .filter(*get_filters(status))
+                .update({Feed.status: status}, synchronize_session=False)
+            )
+    except Exception as e:
+        logging.error(f"Error updating feed statuses: {e}")
+        raise Exception(f"Error updating feed statuses: {e}")
+
+    try:
+        session.commit()
+        refresh_materialized_view(session, t_feedsearch.name)
+        logging.info("Feed Database changes for status committed.")
+        session.close()
+        return diff_counts
+    except Exception as e:
+        logging.error("Error committing changes:", e)
+        session.rollback()
+        session.close()
+        raise Exception(f"Error creating dataset: {e}")
diff --git a/functions-python/process_validation_report/src/main.py b/functions-python/process_validation_report/src/main.py
@@ -34,6 +34,7 @@
 )
 from shared.helpers.logger import Logger
 from shared.helpers.transform import get_nested_value
+from shared.helpers.feed_status import update_feed_statuses_query
 
 logging.basicConfig(level=logging.INFO)
 
@@ -269,8 +270,10 @@ def create_validation_report_entities(
             db_session.add(entity)
         logging.info(f"Committing {len(entities)} entities to the database.")
         db_session.commit()
-
         logging.info("Entities committed successfully.")
+
+        update_feed_statuses_query(db_session, [feed_stable_id])
+
         return f"Created {len(entities)} entities.", 200
     except Exception as error:
         logging.error(f"Error creating validation report entities: {error}")
diff --git a/functions-python/update_feed_status/src/main.py b/functions-python/update_feed_status/src/main.py
@@ -1,95 +1,20 @@
 import logging
 import functions_framework
-from datetime import datetime, timezone
 from shared.helpers.logger import Logger
-from typing import TYPE_CHECKING
-from sqlalchemy import text
-from shared.database_gen.sqlacodegen_models import Gtfsdataset, Feed, t_feedsearch
-from shared.database.database import refresh_materialized_view, with_db_session
-
-if TYPE_CHECKING:
-    from sqlalchemy.orm import Session
+from shared.helpers.feed_status import update_feed_statuses_query
+from shared.database.database import with_db_session
 
 logging.basicConfig(level=logging.INFO)
 
 
-#  query to update the status of the feeds based on the service date range of the latest dataset
-def update_feed_statuses_query(session: "Session"):
-    today_utc = datetime.now(timezone.utc).date()
-
-    latest_dataset_subq = (
-        session.query(
-            Gtfsdataset.feed_id,
-            Gtfsdataset.service_date_range_start,
-            Gtfsdataset.service_date_range_end,
-        )
-        .filter(
-            Gtfsdataset.latest.is_(True),
-            Gtfsdataset.service_date_range_start.isnot(None),
-            Gtfsdataset.service_date_range_end.isnot(None),
-        )
-        .subquery()
-    )
-
-    status_conditions = [
-        (
-            latest_dataset_subq.c.service_date_range_end < today_utc,
-            "inactive",
-        ),
-        (
-            latest_dataset_subq.c.service_date_range_start > today_utc,
-            "future",
-        ),
-        (
-            (latest_dataset_subq.c.service_date_range_start <= today_utc)
-            & (latest_dataset_subq.c.service_date_range_end >= today_utc),
-            "active",
-        ),
-    ]
-
-    try:
-        diff_counts: dict[str, int] = {}
-
-        for service_date_conditions, status in status_conditions:
-            diff_counts[status] = (
-                session.query(Feed)
-                .filter(
-                    Feed.id == latest_dataset_subq.c.feed_id,
-                    Feed.status != text("'deprecated'::status"),
-                    Feed.status != text("'development'::status"),
-                    # We filter out feeds that already have the status so that the
-                    # update count reflects the number of feeds that actually
-                    # changed status.
-                    Feed.status != text("'%s'::status" % status),
-                    service_date_conditions,
-                )
-                .update({Feed.status: status}, synchronize_session=False)
-            )
-    except Exception as e:
-        logging.error(f"Error updating feed statuses: {e}")
-        raise Exception(f"Error updating feed statuses: {e}")
-
-    try:
-        session.commit()
-        refresh_materialized_view(session, t_feedsearch.name)
-        logging.info("Feed Database changes committed.")
-        session.close()
-        return diff_counts
-    except Exception as e:
-        logging.error("Error committing changes:", e)
-        session.rollback()
-        session.close()
-        raise Exception(f"Error creating dataset: {e}")
-
-
 @with_db_session
 @functions_framework.http
 def update_feed_status(_, db_session):
     """Updates the Feed status based on the latets dataset service date range."""
     Logger.init_logger()
     try:
         logging.info("Database session started.")
-        diff_counts = update_feed_statuses_query(db_session)
+        diff_counts = update_feed_statuses_query(db_session, [])
         return diff_counts, 200
 
     except Exception as error:
diff --git a/functions-python/update_feed_status/tests/conftest.py b/functions-python/update_feed_status/tests/conftest.py
@@ -54,7 +54,9 @@ def populate_database(db_session):
     }
     for status, (a, b) in id_range_by_status.items():
         for _id in map(str, range(a, b + 1)):
-            db_session.add(Feed(id=str(_id), status=status))
+            db_session.add(
+                Feed(id=str(_id), status=status, stable_id="mdb-" + str(_id))
+            )
 
     # -> inactive
     for _id in [
diff --git a/functions-python/update_feed_status/tests/test_update_feed_status_main.py b/functions-python/update_feed_status/tests/test_update_feed_status_main.py
@@ -1,5 +1,6 @@
 from unittest.mock import patch, MagicMock
 
+from conftest import clean_testing_db, populate_database
 from shared.database.database import with_db_session
 from test_shared.test_utils.database_utils import default_db_url
 from main import (
@@ -39,7 +40,7 @@ def fetch_feeds(session: Session) -> Iterator[PartialFeed]:
 @with_db_session(db_url=default_db_url)
 def test_update_feed_status(db_session: Session) -> None:
     feeds_before: dict[str, PartialFeed] = {f.id: f for f in fetch_feeds(db_session)}
-    result = dict(update_feed_statuses_query(db_session))
+    result = dict(update_feed_statuses_query(db_session, []))
     assert result == {
         "inactive": 3,
         "active": 2,
@@ -62,6 +63,29 @@ def test_update_feed_status(db_session: Session) -> None:
         )
 
 
+@with_db_session(db_url=default_db_url)
+def test_update_feed_status_with_ids(db_session: Session) -> None:
+    clean_testing_db()
+    populate_database()
+    feeds_before: dict[str, PartialFeed] = {f.id: f for f in fetch_feeds(db_session)}
+    result = dict(update_feed_statuses_query(db_session, ["mdb-8"]))
+    assert result == {
+        "inactive": 1,
+        "active": 0,
+        "future": 0,
+    }
+
+    feeds_after: dict[str, PartialFeed] = {f.id: f for f in fetch_feeds(db_session)}
+    expected_status_changes = {
+        "8": "inactive",
+    }
+    for feed_id, feed_before in feeds_before.items():
+        feed_after = feeds_after[feed_id]
+        assert feed_after.status == expected_status_changes.get(
+            feed_id, feed_before.status
+        )
+
+
 def test_update_feed_status_failed_query():
     mock_session = MagicMock()
 
@@ -79,7 +103,7 @@ def test_update_feed_status_failed_query():
     mock_update_query.update.side_effect = Exception("Mocked exception")
 
     try:
-        update_feed_statuses_query(mock_session)
+        update_feed_statuses_query(mock_session, [])
     except Exception as e:
         assert str(e) == "Error updating feed statuses: Mocked exception"
 
