test: added gbfs data and gbfs_feeds endpoint test

Author: cka-y

api/src/feeds/impl/models/search_feed_item_result_impl.py

@@ -52,6 +52,29 @@ def from_orm_gtfs(cls, feed_search_row):
             else None,
         )
 
+    @classmethod
+    def from_orm_gbfs(cls, feed_search_row):
+        """Create a model instance from a SQLAlchemy a GTFS row object."""
+        return cls(
+            id=feed_search_row.feed_stable_id,
+            data_type=feed_search_row.data_type,
+            status=feed_search_row.status,
+            external_ids=feed_search_row.external_ids,
+            provider=feed_search_row.provider,
+            feed_contact_email=feed_search_row.feed_contact_email,
+            source_info=SourceInfo(
+                producer_url=feed_search_row.producer_url,
+                authentication_type=int(feed_search_row.authentication_type)
+                if feed_search_row.authentication_type
+                else None,
+                authentication_info_url=feed_search_row.authentication_info_url,
+                api_key_parameter_name=feed_search_row.api_key_parameter_name,
+                license_url=feed_search_row.license_url,
+            ),
+            redirects=feed_search_row.redirect_ids,
+            locations=cls.resolve_locations(feed_search_row.locations),
+        )
+
     @classmethod
     def from_orm_gtfs_rt(cls, feed_search_row):
         """Create a model instance from a SQLAlchemy a GTFS-RT row object."""
@@ -115,5 +138,7 @@ def from_orm(cls, feed_search_row):
                 return cls.from_orm_gtfs(feed_search_row)
             case "gtfs_rt":
                 return cls.from_orm_gtfs_rt(feed_search_row)
+            case "gbfs":
+                return cls.from_orm_gbfs(feed_search_row)
             case _:
                 raise ValueError(f"Unknown data type: {feed_search_row.data_type}")

api/src/feeds/impl/search_api_impl.py

@@ -46,7 +46,9 @@ def add_search_query_filters(query, search_query, data_type, feed_id, status, is
         if feed_id:
             query = query.where(t_feedsearch.c.feed_stable_id == feed_id.strip().lower())
         if data_type:
-            query = query.where(t_feedsearch.c.data_type == data_type.strip().lower())
+            data_types = [dt.strip().lower() for dt in data_type.split(",")]
+            if data_types:
+                query = query.where(t_feedsearch.c.data_type.in_(data_types))
         if status:
             status_list = [s.strip().lower() for s in status[0].split(",") if s]
             if status_list:

api/src/scripts/gbfs_utils/comparison.py

@@ -29,6 +29,9 @@ def generate_system_csv_from_db(df, db_session):
                 "Supported Versions": " ; ".join(supported_versions),
             }
         )
+    if not data:
+        # Return an empty DataFrame with the same columns
+        return pd.DataFrame(columns=df.columns)
     return pd.DataFrame(data)
 
 

api/src/scripts/populate_db.py

@@ -1,15 +1,21 @@
 import argparse
 import logging
 import os
+import traceback
 from pathlib import Path
 from typing import Type, TYPE_CHECKING
 
 import pandas
 from dotenv import load_dotenv
+from sqlalchemy import text
 
+from shared.common.logging_utils import Logger
 from shared.database.database import Database
+from shared.database.database import configure_polymorphic_mappers
 from shared.database_gen.sqlacodegen_models import Feed, Gtfsrealtimefeed, Gtfsfeed, Gbfsfeed
-from shared.common.logging_utils import Logger
+from shared.database_gen.sqlacodegen_models import (
+    t_feedsearch,
+)
 
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
@@ -117,3 +123,35 @@ def filter_data(self):
         Filter the data to only include the necessary columns
         """
         pass  # Should be implemented in the child class
+
+    def populate_db(self, session: "Session", fetch_url: bool = True):
+        """
+        Populate the database with the data
+        """
+        pass  # Should be implemented in the child class
+
+    def trigger_downstream_tasks(self):
+        """
+        Trigger downstream tasks
+        """
+        pass  # Should be implemented in the child class
+
+    # Extracted the following code from main, so it can be executed as a library function
+    def initialize(self, trigger_downstream_tasks: bool = True, fetch_url: bool = True):
+        try:
+            configure_polymorphic_mappers()
+            with self.db.start_db_session() as session:
+                self.populate_db(session, fetch_url=fetch_url)
+                session.commit()
+
+                self.logger.info("Refreshing MATERIALIZED FEED SEARCH VIEW - Started")
+                session.execute(text(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {t_feedsearch.name}"))
+                self.logger.info("Refreshing MATERIALIZED FEED SEARCH VIEW - Completed")
+                session.commit()
+                self.logger.info("\n----- Database populated with sources.csv data. -----")
+                if trigger_downstream_tasks:
+                    self.trigger_downstream_tasks()
+        except Exception as e:
+            self.logger.error(f"\n------ Failed to populate the database with sources.csv: {e} -----\n")
+            traceback.print_exc()
+            exit(1)

api/src/scripts/populate_db_gbfs.py

@@ -13,12 +13,12 @@
 
 
 class GBFSDatabasePopulateHelper(DatabasePopulateHelper):
-    def __init__(self, file_path):
-        super().__init__(file_path)
+    def __init__(self, filepaths):
+        super().__init__(filepaths)
 
     def filter_data(self):
         """Filter out rows with Authentication Info and duplicate System IDs"""
-        self.df = self.df[pd.isna(self.df["Authentication Info"])]
+        self.df = self.df[pd.isna(self.df["Authentication Info URL"])]
         self.df = self.df[~self.df.duplicated(subset="System ID", keep=False)]
         self.logger.info(f"Data = {self.df}")
 
@@ -45,77 +45,80 @@ def deprecate_feeds(self, deprecated_feeds):
                     self.logger.info(f"Deprecating feed with stable_id={stable_id}")
                     gbfs_feed.status = "deprecated"
 
-    def populate_db(self):
+    def populate_db(self, session, fetch_url=True):
         """Populate the database with the GBFS feeds"""
         start_time = datetime.now()
         configure_polymorphic_mappers()
 
         try:
-            with self.db.start_db_session() as session:
-                # Compare the database to the CSV file
-                df_from_db = generate_system_csv_from_db(self.df, session)
-                added_or_updated_feeds, deprecated_feeds = compare_db_to_csv(df_from_db, self.df, self.logger)
-
-                self.deprecate_feeds(deprecated_feeds)
-                if added_or_updated_feeds is None:
-                    added_or_updated_feeds = self.df
-                for index, row in added_or_updated_feeds.iterrows():
-                    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")
+            # Compare the database to the CSV file
+            df_from_db = generate_system_csv_from_db(self.df, session)
+            added_or_updated_feeds, deprecated_feeds = compare_db_to_csv(df_from_db, self.df, self.logger)
+
+            self.deprecate_feeds(deprecated_feeds)
+            if added_or_updated_feeds is None:
+                added_or_updated_feeds = self.df
+            for index, row in added_or_updated_feeds.iterrows():
+                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")
+                if fetch_url:
                     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:
-                        self.logger.info(f"Updating feed {stable_id} - {row['Name']}")
-                    else:
-                        feed_id = generate_unique_id()
-                        self.logger.info(f"Creating new feed for {stable_id} - {row['Name']}")
-                        gbfs_feed = Gbfsfeed(
-                            id=feed_id,
-                            data_type="gbfs",
-                            stable_id=stable_id,
-                            created_at=datetime.now(pytz.utc),
-                            operational_status="published",
-                        )
-                        gbfs_feed.externalids = [self.get_external_id(feed_id, row["System ID"])]
-                        session.add(gbfs_feed)
-
-                    system_information_content = get_data_content(fetched_data.get("system_information"), self.logger)
-                    gbfs_feed.license_url = get_license_url(system_information_content, self.logger)
-                    gbfs_feed.feed_contact_email = (
-                        system_information_content.get("feed_contact_email") if system_information_content else None
+                else:
+                    fetched_data = dict()
+                # If the feed already exists, update it. Otherwise, create a new feed.
+                if gbfs_feed:
+                    self.logger.info(f"Updating feed {stable_id} - {row['Name']}")
+                else:
+                    feed_id = generate_unique_id()
+                    self.logger.info(f"Creating new feed for {stable_id} - {row['Name']}")
+                    gbfs_feed = Gbfsfeed(
+                        id=feed_id,
+                        data_type="gbfs",
+                        stable_id=stable_id,
+                        created_at=datetime.now(pytz.utc),
+                        operational_status="published",
                     )
-                    gbfs_feed.system_id = str(row["System ID"]).strip()
-                    gbfs_feed.operator = row["Name"]
-                    gbfs_feed.provider = row["Name"]
-                    gbfs_feed.operator_url = row["URL"]
-                    gbfs_feed.producer_url = row["URL"]
-                    gbfs_feed.auto_discovery_url = row["Auto-Discovery URL"]
-                    gbfs_feed.updated_at = datetime.now(pytz.utc)
-
-                    if not gbfs_feed.locations:  # If locations are empty, create a new location (no overwrite)
-                        country_code = self.get_safe_value(row, "Country Code", "")
-                        municipality = self.get_safe_value(row, "Location", "")
-                        location_id = self.get_location_id(country_code, None, municipality)
-                        country = pycountry.countries.get(alpha_2=country_code) if country_code else None
-                        location = session.get(Location, location_id) or Location(
-                            id=location_id,
-                            country_code=country_code,
-                            country=country.name if country else None,
-                            municipality=municipality,
-                        )
-                        gbfs_feed.locations.clear()
-                        gbfs_feed.locations = [location]
-
-                    self.logger.info(80 * "-")
-
-                    # self.db.session.commit()
-                    end_time = datetime.now()
-                    self.logger.info(f"Time taken: {end_time - start_time} seconds")
+                    gbfs_feed.externalids = [self.get_external_id(feed_id, row["System ID"])]
+                    session.add(gbfs_feed)
+
+                system_information_content = get_data_content(fetched_data.get("system_information"), self.logger)
+                gbfs_feed.license_url = get_license_url(system_information_content, self.logger)
+                gbfs_feed.feed_contact_email = (
+                    system_information_content.get("feed_contact_email") if system_information_content else None
+                )
+                gbfs_feed.system_id = str(row["System ID"]).strip()
+                gbfs_feed.operator = row["Name"]
+                gbfs_feed.provider = row["Name"]
+                gbfs_feed.operator_url = row["URL"]
+                gbfs_feed.producer_url = row["URL"]
+                gbfs_feed.auto_discovery_url = row["Auto-Discovery URL"]
+                gbfs_feed.updated_at = datetime.now(pytz.utc)
+
+                if not gbfs_feed.locations:  # If locations are empty, create a new location (no overwrite)
+                    country_code = self.get_safe_value(row, "Country Code", "")
+                    municipality = self.get_safe_value(row, "Location", "")
+                    location_id = self.get_location_id(country_code, None, municipality)
+                    country = pycountry.countries.get(alpha_2=country_code) if country_code else None
+                    location = session.get(Location, location_id) or Location(
+                        id=location_id,
+                        country_code=country_code,
+                        country=country.name if country else None,
+                        municipality=municipality,
+                    )
+                    gbfs_feed.locations.clear()
+                    gbfs_feed.locations = [location]
+
+                session.flush()
+                self.logger.info(80 * "-")
+
+                # self.db.session.commit()
+                end_time = datetime.now()
+                self.logger.info(f"Time taken: {end_time - start_time} seconds")
         except Exception as e:
             self.logger.error(f"Error populating the database: {e}")
             raise e
 
 
 if __name__ == "__main__":
-    GBFSDatabasePopulateHelper(set_up_configs()).populate_db()
+    GBFSDatabasePopulateHelper(set_up_configs()).initialize(trigger_downstream_tasks=False)

api/src/scripts/populate_db_gtfs.py

@@ -1,22 +1,19 @@
 import os
-import traceback
 from datetime import datetime
 from typing import TYPE_CHECKING
 
 import pycountry
 import pytz
-from sqlalchemy import text
 
 from scripts.load_dataset_on_create import publish_all
 from scripts.populate_db import DatabasePopulateHelper, set_up_configs
-from shared.database.database import generate_unique_id, configure_polymorphic_mappers
+from shared.database.database import generate_unique_id
 from shared.database_gen.sqlacodegen_models import (
     Entitytype,
     Externalid,
     Gtfsrealtimefeed,
     Location,
     Redirectingid,
-    t_feedsearch,
 )
 from utils.data_utils import set_up_defaults
 
@@ -191,7 +188,7 @@ def process_redirects(self, session: "Session"):
                         # Flush to avoid FK violation
                         session.flush()
 
-    def populate_db(self, session: "Session"):
+    def populate_db(self, session: "Session", fetch_url: bool = True):
         """
         Populate the database with the sources.csv data
         """
@@ -286,26 +283,6 @@ def post_process_locations(self, session: "Session"):
         session.commit()
         self.logger.info(f"Had to set the country for {set_country_count} locations")
 
-    # Extracted the following code from main, so it can be executed as a library function
-    def initialize(self, trigger_downstream_tasks: bool = True):
-        try:
-            configure_polymorphic_mappers()
-            with self.db.start_db_session() as session:
-                self.populate_db(session)
-                session.commit()
-
-                self.logger.info("Refreshing MATERIALIZED FEED SEARCH VIEW - Started")
-                session.execute(text(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {t_feedsearch.name}"))
-                self.logger.info("Refreshing MATERIALIZED FEED SEARCH VIEW - Completed")
-                session.commit()
-                self.logger.info("\n----- Database populated with sources.csv data. -----")
-                if trigger_downstream_tasks:
-                    self.trigger_downstream_tasks()
-        except Exception as e:
-            self.logger.error(f"\n------ Failed to populate the database with sources.csv: {e} -----\n")
-            traceback.print_exc()
-            exit(1)
-
 
 if __name__ == "__main__":
     db_helper = GTFSDatabasePopulateHelper(set_up_configs())

api/src/scripts/populate_db_test_data.py

@@ -47,6 +47,7 @@ def populate_test_datasets(self, filepath, db_session: "Session"):
         """
         Populate the database with the test datasets
         """
+        # TODO: parse GBFS versions
         # Load the JSON file
         with open(filepath) as f:
             data = json.load(f)