feat: enhance gbfs validation pipeline for version tracking (#1064)

Author: cka-y

.github/workflows/db-update.yml

@@ -199,12 +199,12 @@ jobs:
       id: getsyspath
       run: echo "PATH=$(realpath systems.csv)" >> $GITHUB_OUTPUT
 
-    - name: GTFS - Update Database Content
-      run: scripts/populate-db.sh ${{ steps.getpath.outputs.PATH }} > populate.log
-
     - name: GBFS - Update Database Content
       run: scripts/populate-db.sh ${{ steps.getsyspath.outputs.PATH }} gbfs >> populate-gbfs.log
 
+    - name: GTFS - Update Database Content
+      run: scripts/populate-db.sh ${{ steps.getpath.outputs.PATH }} > populate.log
+
     - name: GTFS - Upload log file for verification
       if: ${{ always() }}
       uses: actions/upload-artifact@v4

api/src/scripts/gbfs_utils/comparison.py

@@ -38,6 +38,9 @@ def compare_db_to_csv(df_from_db, df_from_csv, logger):
     df_from_db = df_from_db.fillna("")
     df_from_csv = df_from_csv.fillna("")
 
+    df_from_db = df_from_db.drop(columns=["Supported Versions"])
+    df_from_csv = df_from_csv.drop(columns=["Supported Versions"])
+
     if df_from_db.empty:
         logger.info("No data found in the database.")
         return None, None
@@ -63,7 +66,7 @@ def compare_db_to_csv(df_from_db, df_from_csv, logger):
     df_db_common = df_from_db.loc[common_ids]
     df_csv_common = df_from_csv.loc[common_ids]
     differences = df_db_common != df_csv_common
-    differing_rows = df_db_common[differences.any(axis=1)]
+    differing_rows = df_csv_common[differences.any(axis=1)]
 
     if not differing_rows.empty:
         logger.info("Rows with differences:")

api/src/scripts/gbfs_utils/fetching.py

@@ -52,30 +52,3 @@ def get_field_url(fields, field_name):
         if field.get("name") == field_name:
             return field.get("url")
     return None
-
-
-def get_gbfs_versions(gbfs_versions_url, auto_discovery_url, auto_discovery_version, logger):
-    """Get the GBFS versions from the gbfs_versions_url."""
-    # Default version info extracted from auto-discovery url
-    version_info = {
-        "version": auto_discovery_version if auto_discovery_version else "1.0",
-        "url": auto_discovery_url,
-    }
-    try:
-        if not gbfs_versions_url:
-            return [version_info]
-        logger.info(f"Fetching GBFS versions from: {gbfs_versions_url}")
-        data = get_data_content(gbfs_versions_url, logger)
-        if not data:
-            logger.warning(f"No data found in the GBFS versions URL -> {gbfs_versions_url}.")
-            return [version_info]
-        gbfs_versions = data.get("versions", [])
-
-        # Append the version info from auto-discovery if it doesn't exist
-        if not any(gv.get("version") == auto_discovery_version for gv in gbfs_versions):
-            gbfs_versions.append(version_info)
-
-        return gbfs_versions
-    except Exception as e:
-        logger.error(f"Error fetching version data: {e}")
-        return [version_info]

api/src/scripts/gbfs_utils/gbfs_versions.py

@@ -1,16 +1,15 @@
 from datetime import datetime
 
 import pandas as pd
-import pytz
 import pycountry
+import pytz
 
-from shared.database.database import generate_unique_id, configure_polymorphic_mappers
-from shared.database_gen.sqlacodegen_models import Gbfsfeed, Location, Gbfsversion, Externalid
 from scripts.gbfs_utils.comparison import generate_system_csv_from_db, compare_db_to_csv
-from scripts.gbfs_utils.fetching import fetch_data, get_data_content, get_gbfs_versions
+from scripts.gbfs_utils.fetching import fetch_data, get_data_content
 from scripts.gbfs_utils.license import get_license_url
 from scripts.populate_db import DatabasePopulateHelper, set_up_configs
-from scripts.gbfs_utils.gbfs_versions import OFFICIAL_VERSIONS
+from shared.database.database import generate_unique_id, configure_polymorphic_mappers
+from shared.database_gen.sqlacodegen_models import Gbfsfeed, Location, Externalid
 
 
 class GBFSDatabasePopulateHelper(DatabasePopulateHelper):
@@ -45,7 +44,6 @@ def deprecate_feeds(self, deprecated_feeds):
                 if gbfs_feed:
                     self.logger.info(f"Deprecating feed with stable_id={stable_id}")
                     gbfs_feed.status = "deprecated"
-                    # session.flush()
 
     def populate_db(self):
         """Populate the database with the GBFS feeds"""
@@ -65,12 +63,9 @@ def populate_db(self):
                     self.logger.info(f"Processing row {index + 1} of {len(added_or_updated_feeds)}")
                     stable_id = self.get_stable_id(row)
                     gbfs_feed = self.query_feed_by_stable_id(session, stable_id, "gbfs")
-                    fetched_data = fetch_data(
-                        row["Auto-Discovery URL"], self.logger, ["system_information", "gbfs_versions"], ["version"]
-                    )
+                    fetched_data = fetch_data(row["Auto-Discovery URL"], self.logger, ["system_information"])
                     # If the feed already exists, update it. Otherwise, create a new feed.
                     if gbfs_feed:
-                        feed_id = gbfs_feed.id
                         self.logger.info(f"Updating feed {stable_id} - {row['Name']}")
                     else:
                         feed_id = generate_unique_id()
@@ -108,26 +103,6 @@ def populate_db(self):
                     gbfs_feed.locations.clear()
                     gbfs_feed.locations = [location]
 
-                    # Add the GBFS versions
-                    versions = get_gbfs_versions(
-                        fetched_data.get("gbfs_versions"),
-                        row["Auto-Discovery URL"],
-                        fetched_data.get("version"),
-                        self.logger,
-                    )
-                    existing_versions = [version.version for version in gbfs_feed.gbfsversions]
-                    for version in versions:
-                        version_value = version.get("version")
-                        if version_value.upper() in OFFICIAL_VERSIONS and version_value not in existing_versions:
-                            gbfs_feed.gbfsversions.append(
-                                Gbfsversion(
-                                    feed_id=feed_id,
-                                    url=version.get("url"),
-                                    version=version_value,
-                                )
-                            )
-
-                    # self.db.session.flush()
                     self.logger.info(80 * "-")
 
                     # self.db.session.commit()

api/src/scripts/populate_db_gbfs.py

@@ -7,7 +7,15 @@
 from dotenv import load_dotenv
 from sqlalchemy import create_engine, text, event
 from sqlalchemy.orm import load_only, Query, class_mapper, Session, mapper
-from shared.database_gen.sqlacodegen_models import Base, Feed, Gtfsfeed, Gtfsrealtimefeed, Gbfsfeed
+from shared.database_gen.sqlacodegen_models import (
+    Base,
+    Feed,
+    Gtfsfeed,
+    Gtfsrealtimefeed,
+    Gbfsversion,
+    Gbfsfeed,
+    Gbfsvalidationreport,
+)
 from sqlalchemy.orm import sessionmaker
 import logging
 
@@ -44,20 +52,26 @@ def configure_polymorphic_mappers():
     gbfsfeed_mapper.polymorphic_identity = Gbfsfeed.__tablename__.lower()
 
 
+cascade_entities = {
+    Gtfsfeed: [Gtfsfeed.redirectingids, Gtfsfeed.redirectingids_, Gtfsfeed.externalids],
+    Gbfsversion: [Gbfsversion.gbfsendpoints, Gbfsversion.gbfsvalidationreports],
+    Gbfsfeed: [Gbfsfeed.gbfsversions],
+    Gbfsvalidationreport: [Gbfsvalidationreport.gbfsnotices],
+}
+
+
 def set_cascade(mapper, class_):
     """
     Set cascade for relationships in Gtfsfeed.
     This allows to delete/add the relationships when their respective relation array changes.
     """
-    if class_.__name__ == "Gtfsfeed":
+    mapper.confirm_deleted_rows = False  # Disable confirm_deleted_rows to avoid warnings in logs with delete-orphan
+    if class_ in cascade_entities:
+        relationship_keys = {rel.prop.key for rel in cascade_entities[class_]}
         for rel in class_.__mapper__.relationships:
-            if rel.key in [
-                "redirectingids",
-                "redirectingids_",
-                "externalids",
-                "externalids_",
-            ]:
+            if rel.key in relationship_keys:
                 rel.cascade = "all, delete-orphan"
+                rel.passive_deletes = True
 
 
 def mapper_configure_listener(mapper, class_):

api/src/shared/database/database.py

@@ -32,9 +32,9 @@ The message published by the batch function to the Pub/Sub topic follows this fo
 
 - **`gbfs-validator-batch`**: Triggered per execution ID, this function iterates over all GBFS feeds, preparing and publishing individual messages to the Pub/Sub topic.
 - **`gbfs-validator-pubsub`**: Triggered per feed, this function performs the following steps:
-  1. **Download the feed snapshot to GCP**: It uploads all related files to the specified Cloud Storage bucket and updates the `gbfs.json` file to point to the newly uploaded files.
-  2. **Validate the feed**: Run the GBFS validator on the feed snapshot.
-  3. **Update the database**: The function updates the database with the snapshot information and validation report details.
+  1. **Access the autodiscovery URL and update versions**: The function accesses the autodiscovery URL to update the **GBFSVersions** table.
+  2. **Measure latency and validate the feed**: For each version, the function measures the response latency and validates the feed. The validation summary is stored in GCP, and the total error count is extracted and saved in the **GBFSValidationReport**.
+  3. **Store validation details**: The function stores detailed errors as **GBFSNotice** entities.
 
 ## Function Configuration
 

functions-python/gbfs_validator/README.md

@@ -23,4 +23,8 @@ google-cloud-datastore
 cloudevents~=1.10.1
 
 # Configuration
-python-dotenv==1.0.0
+python-dotenv==1.0.0
+
+# Additional packages for the function
+jsonpath-ng
+language-tags