Merge branch 'main' into 1034-change-how-often-were-pulling-the-data-to-daily

Author: qcdyx

api/src/feeds/impl/feeds_api_impl.py

@@ -6,7 +6,12 @@
 from sqlalchemy.orm import joinedload, Session
 from sqlalchemy.orm.query import Query
 
-from shared.common.db_utils import get_gtfs_feeds_query, get_gtfs_rt_feeds_query, get_joinedload_options
+from shared.common.db_utils import (
+    get_gtfs_feeds_query,
+    get_gtfs_rt_feeds_query,
+    get_joinedload_options,
+    add_official_filter,
+)
 from shared.database.database import Database, with_db_session
 from shared.database_gen.sqlacodegen_models import (
     Feed,
@@ -98,8 +103,7 @@ def get_feeds(
             status=status, provider__ilike=provider, producer_url__ilike=producer_url, stable_id=None
         )
         feed_query = feed_filter.filter(Database().get_query_model(db_session, Feed))
-        if is_official:
-            feed_query = feed_query.filter(Feed.official)
+        feed_query = add_official_filter(feed_query, is_official)
         feed_query = feed_query.filter(Feed.data_type != "gbfs")  # Filter out GBFS feeds
         feed_query = feed_query.filter(
             or_(
@@ -127,13 +131,14 @@ def get_gtfs_feed(self, id: str, db_session: Session) -> GtfsFeed:
         else:
             raise_http_error(404, gtfs_feed_not_found.format(id))
 
-    @staticmethod
     def _get_gtfs_feed(
-        stable_id: str, db_session: Session, include_options_for_joinedload: bool = True
+        self, stable_id: str, db_session: Session, include_options_for_joinedload: bool = True
     ) -> Optional[Gtfsfeed]:
-        results = get_gtfs_feeds_query(
+        query = get_gtfs_feeds_query(
             db_session=db_session, stable_id=stable_id, include_options_for_joinedload=include_options_for_joinedload
-        ).all()
+        )
+        self.logger.debug("Query: %s", str(query.statement.compile(compile_kwargs={"literal_binds": True})))
+        results = query.all()
         if len(results) == 0:
             return None
         return results[0]
@@ -194,7 +199,7 @@ def get_gtfs_feeds(
         db_session: Session,
     ) -> List[GtfsFeed]:
         try:
-            include_wip = not is_user_email_restricted()
+            published_only = is_user_email_restricted()
             feed_query = get_gtfs_feeds_query(
                 limit=limit,
                 offset=offset,
@@ -207,7 +212,7 @@ def get_gtfs_feeds(
                 dataset_longitudes=dataset_longitudes,
                 bounding_filter_method=bounding_filter_method,
                 is_official=is_official,
-                include_wip=include_wip,
+                published_only=published_only,
                 db_session=db_session,
             )
         except InternalHTTPException as e:
@@ -265,7 +270,7 @@ def get_gtfs_rt_feeds(
     ) -> List[GtfsRTFeed]:
         """Get some (or all) GTFS Realtime feeds from the Mobility Database."""
         try:
-            include_wip = not is_user_email_restricted()
+            published_only = is_user_email_restricted()
             feed_query = get_gtfs_rt_feeds_query(
                 limit=limit,
                 offset=offset,
@@ -276,7 +281,7 @@ def get_gtfs_rt_feeds(
                 subdivision_name=subdivision_name,
                 municipality=municipality,
                 is_official=is_official,
-                include_wip=include_wip,
+                published_only=published_only,
                 db_session=db_session,
             )
         except InternalHTTPException as e:
@@ -328,8 +333,8 @@ def get_gtfs_rt_feeds(
             )
             .order_by(Gtfsrealtimefeed.provider, Gtfsrealtimefeed.stable_id)
         )
-        if is_official:
-            feed_query = feed_query.filter(Feed.official)
+        feed_query = add_official_filter(feed_query, is_official)
+
         feed_query = feed_query.limit(limit).offset(offset)
         return self._get_response(feed_query, GtfsRTFeedImpl)
 

api/src/feeds/impl/search_api_impl.py

@@ -52,8 +52,11 @@ def add_search_query_filters(query, search_query, data_type, feed_id, status, is
             status_list = [s.strip().lower() for s in status[0].split(",") if s]
             if status_list:
                 query = query.where(t_feedsearch.c.status.in_([s.strip().lower() for s in status_list]))
-        if is_official is not None and is_official:
-            query = query.where(t_feedsearch.c.official == is_official)
+        if is_official is not None:
+            if is_official:
+                query = query.where(t_feedsearch.c.official.is_(True))
+            else:
+                query = query.where(or_(t_feedsearch.c.official.is_(False), t_feedsearch.c.official.is_(None)))
         if search_query and len(search_query.strip()) > 0:
             query = query.filter(
                 t_feedsearch.c.document.op("@@")(SearchApiImpl.get_parsed_search_tsquery(search_query))

api/src/shared/common/db_utils.py

@@ -37,66 +37,72 @@ def get_gtfs_feeds_query(
     dataset_latitudes: str | None = None,
     dataset_longitudes: str | None = None,
     bounding_filter_method: str | None = None,
-    is_official: bool = False,
-    include_wip: bool = False,
+    is_official: bool | None = None,
+    published_only: bool = True,
     include_options_for_joinedload: bool = True,
 ) -> Query[any]:
     """Get the DB query to use to retrieve the GTFS feeds.."""
     gtfs_feed_filter = GtfsFeedFilter(
         stable_id=stable_id,
         provider__ilike=provider,
         producer_url__ilike=producer_url,
-        location=LocationFilter(
-            country_code=country_code,
-            subdivision_name__ilike=subdivision_name,
-            municipality__ilike=municipality,
-        ),
+        location=None,
     )
 
-    subquery = gtfs_feed_filter.filter(select(Gtfsfeed.id).join(Location, Gtfsfeed.locations))
+    subquery = gtfs_feed_filter.filter(select(Gtfsfeed.id))
     subquery = apply_bounding_filtering(
         subquery, dataset_latitudes, dataset_longitudes, bounding_filter_method
     ).subquery()
-
     feed_query = (
         db_session.query(Gtfsfeed)
         .outerjoin(Gtfsfeed.gtfsdatasets)
         .filter(Gtfsfeed.id.in_(subquery))
-        .filter((Gtfsdataset.latest) | (Gtfsdataset.id == None))  # noqa: E711
+        .filter(or_(Gtfsdataset.latest, Gtfsdataset.id == None))  # noqa: E711
     )
-    if not include_wip:
+
+    if country_code or subdivision_name or municipality:
+        location_filter = LocationFilter(
+            country_code=country_code,
+            subdivision_name__ilike=subdivision_name,
+            municipality__ilike=municipality,
+        )
+        location_subquery = location_filter.filter(select(Location.id))
+        feed_query = feed_query.filter(Gtfsfeed.locations.any(Location.id.in_(location_subquery)))
+
+    if published_only:
         feed_query = feed_query.filter(Gtfsfeed.operational_status == "published")
 
+    feed_query = add_official_filter(feed_query, is_official)
+
     if include_options_for_joinedload:
         feed_query = feed_query.options(
             contains_eager(Gtfsfeed.gtfsdatasets)
             .joinedload(Gtfsdataset.validation_reports)
             .joinedload(Validationreport.notices),
             *get_joinedload_options(),
         ).order_by(Gtfsfeed.provider, Gtfsfeed.stable_id)
-    if is_official:
-        feed_query = feed_query.filter(Feed.official)
+
     feed_query = feed_query.limit(limit).offset(offset)
     return feed_query
 
 
 def get_all_gtfs_feeds(
     db_session: Session,
-    include_wip: bool = False,
+    published_only: bool = True,
     batch_size: int = 250,
 ) -> Iterator[Gtfsfeed]:
     """
     Fetch all GTFS feeds.
 
     @param db_session: The database session.
-    @param include_wip: Whether to include or exclude WIP feeds.
+    @param published_only: Include only the published feeds.
     @param batch_size: The number of feeds to fetch from the database at a time.
         A lower value means less memory but more queries.
 
     @return: The GTFS feeds in an iterator.
     """
     feed_query = db_session.query(Gtfsfeed).order_by(Gtfsfeed.stable_id).yield_per(batch_size)
-    if not include_wip:
+    if published_only:
         feed_query = feed_query.filter(Gtfsfeed.operational_status == "published")
 
     for batch in batched(feed_query, batch_size):
@@ -126,7 +132,7 @@ def get_gtfs_rt_feeds_query(
     subdivision_name: str | None,
     municipality: str | None,
     is_official: bool | None,
-    include_wip: bool = False,
+    published_only: bool = True,
     db_session: Session = None,
 ) -> Query:
     """Get some (or all) GTFS Realtime feeds from the Mobility Database."""
@@ -160,37 +166,49 @@ def get_gtfs_rt_feeds_query(
     ).subquery()
     feed_query = db_session.query(Gtfsrealtimefeed).filter(Gtfsrealtimefeed.id.in_(subquery))
 
-    if not include_wip:
+    if published_only:
         feed_query = feed_query.filter(Gtfsrealtimefeed.operational_status == "published")
 
     feed_query = feed_query.options(
         joinedload(Gtfsrealtimefeed.entitytypes),
         joinedload(Gtfsrealtimefeed.gtfs_feeds),
         *get_joinedload_options(),
     )
-    if is_official:
-        feed_query = feed_query.filter(Feed.official)
+    feed_query = add_official_filter(feed_query, is_official)
+
     feed_query = feed_query.limit(limit).offset(offset)
     return feed_query
 
 
+def add_official_filter(query: Query, is_official: bool | None) -> Query:
+    """
+    Add the is_official filter to the query if necessary
+    """
+    if is_official is not None:
+        if is_official:
+            query = query.filter(Feed.official.is_(True))
+        else:
+            query = query.filter(or_(Feed.official.is_(False), Feed.official.is_(None)))
+    return query
+
+
 def get_all_gtfs_rt_feeds(
     db_session: Session,
-    include_wip: bool = False,
+    published_only: bool = True,
     batch_size: int = 250,
 ) -> Iterator[Gtfsrealtimefeed]:
     """
     Fetch all GTFS realtime feeds.
 
     @param db_session: The database session.
-    @param include_wip: Whether to include or exclude WIP feeds.
+    @param published_only: Include only the published feeds.
     @param batch_size: The number of feeds to fetch from the database at a time.
         A lower value means less memory but more queries.
 
     @return: The GTFS realtime feeds in an iterator.
     """
     feed_query = db_session.query(Gtfsrealtimefeed.stable_id).order_by(Gtfsrealtimefeed.stable_id).yield_per(batch_size)
-    if not include_wip:
+    if published_only:
         feed_query = feed_query.filter(Gtfsrealtimefeed.operational_status == "published")
 
     for batch in batched(feed_query, batch_size):

api/tests/integration/test_data/sources_test.csv

@@ -1,6 +1,6 @@
-mdb_source_id,data_type,entity_type,location.country_code,location.subdivision_name,location.municipality,provider,is_offical,name,note,feed_contact_email,static_reference,urls.direct_download,urls.authentication_type,urls.authentication_info,urls.api_key_parameter_name,urls.latest,urls.license,location.bounding_box.minimum_latitude,location.bounding_box.maximum_latitude,location.bounding_box.minimum_longitude,location.bounding_box.maximum_longitude,location.bounding_box.extracted_on,status,features,redirect.id,redirect.comment
+mdb_source_id,data_type,entity_type,location.country_code,location.subdivision_name,location.municipality,provider,is_official,name,note,feed_contact_email,static_reference,urls.direct_download,urls.authentication_type,urls.authentication_info,urls.api_key_parameter_name,urls.latest,urls.license,location.bounding_box.minimum_latitude,location.bounding_box.maximum_latitude,location.bounding_box.minimum_longitude,location.bounding_box.maximum_longitude,location.bounding_box.extracted_on,status,features,redirect.id,redirect.comment
 40,gtfs,,CA,Ontario,London,London Transit Commission,TRUE,,,[email protected],,http://www.londontransit.ca/gtfsfeed/google_transit.zip,0,,,https://storage.googleapis.com/storage/v1/b/mdb-latest/o/ca-ontario-london-transit-commission-gtfs-2.zip?alt=media,https://www.londontransit.ca/open-data/ltcs-open-data-terms-of-use/,42.905244,43.051188,-81.36311,-81.137591,2022-02-22T19:51:34+00:00,inactive,,,
 50,gtfs,,CA,Ontario,Barrie,ZBarrie Transit,FALSE,,,,,http://www.myridebarrie.ca/gtfs/Google_transit.zip,,,,https://storage.googleapis.com/storage/v1/b/mdb-latest/o/ca-ontario-barrie-transit-gtfs-3.zip?alt=media,https://www.barrie.ca/services-payments/transportation-parking/barrie-transit/barrie-gtfs,44.3218044,44.42020676,-79.74063237,-79.61089569,2022-03-01T22:43:25+00:00,deprecated,,40|mdb-702,Some|Comment
 702,gtfs,,CA,[British Columbia,Whistler],BC Transit (Whistler Transit System),,,,,,http://whistler.mapstrat.com/current/google_transit.zip,,,,https://storage.googleapis.com/storage/v1/b/mdb-latest/o/ca-british-columbia-bc-transit-whistler-transit-system-gtfs-702.zip?alt=media,https://www.bctransit.com/open-data/terms-of-use,50.077122,50.159071,-123.043635,-122.926836,2022-03-16T22:05:05+00:00,development,,,
-1562,gtfs-rt,sa,CA,BC,Vancouver,Vancouver-Transit(éèàçíóúČ),,Realtime(ŘŤÜÎ),,,40,http://foo.org/google_transit.zip,0,,,,,,,,,,active,,10,
-1563,gtfs-rt,tu,US,SomeState,SomeCity,SomeCity Bus,,RT,,,mdb-50,http://bar.com,0,,,,,,,,,,inactive,,10,
+1562,gtfs-rt,sa,CA,BC,Vancouver,Vancouver-Transit(éèàçíóúČ),True,Realtime(ŘŤÜÎ),,,40,http://foo.org/google_transit.zip,0,,,,,,,,,,active,,10,
+1563,gtfs-rt,tu,US,SomeState,SomeCity,SomeCity Bus,False,RT,,,mdb-50,http://bar.com,0,,,,,,,,,,inactive,,10,

api/tests/integration/test_feeds_api.py

@@ -247,22 +247,52 @@ def test_feeds_gtfs_rt_id_get(client: TestClient):
 
 
 @pytest.mark.parametrize(
-    "endpoint",
+    "values",
     [
-        "/v1/gtfs_feeds",
-        "/v1/gtfs_rt_feeds",
-        "/v1/feeds",
+        {
+            "endpoint": "/v1/gtfs_feeds",
+            "all_feeds_count": 10,
+            "false_feeds_count": 9,
+            "true_feeds_count": 1,
+            "official_feeds": ["mdb-40"],
+        },
+        {
+            "endpoint": "/v1/gtfs_rt_feeds",
+            "all_feeds_count": 3,
+            "false_feeds_count": 2,
+            "true_feeds_count": 1,
+            "official_feeds": ["mdb-1562"],
+        },
+        {
+            "endpoint": "/v1/feeds",
+            "all_feeds_count": 13,
+            "false_feeds_count": 11,
+            "true_feeds_count": 2,
+            "official_feeds": ["mdb-1562", "mdb-40"],
+        },
     ],
+    ids=["gtfs_feeds", "gtfs_rt_feeds", "feeds"],
 )
-def test_feeds_filter_by_official(client: TestClient, endpoint):
-    # 1 - Test with official=false should return all feeds
+def test_feeds_filter_by_official(client: TestClient, values):
+    endpoint = values["endpoint"]
+    official_feeds = values["official_feeds"]
+    all_feeds_count = values["all_feeds_count"]
+    false_feeds_count = values["false_feeds_count"]
+    true_feeds_count = values["true_feeds_count"]
+
+    # 1 - Test with official not specified should return all feeds
     response_no_filter = client.request(
         "GET",
         endpoint,
         headers=authHeaders,
     )
     assert response_no_filter.status_code == 200
     response_no_filter_json = response_no_filter.json()
+    assert (
+        len(response_no_filter_json) == all_feeds_count
+    ), f"official not specified should return {all_feeds_count} feeds but got {len(response_no_filter_json)}"
+
+    # 2 - Test with official=false
     response_official_false = client.request(
         "GET",
         endpoint,
@@ -271,8 +301,14 @@ def test_feeds_filter_by_official(client: TestClient, endpoint):
     )
     assert response_official_false.status_code == 200
     response_official_false_json = response_official_false.json()
-    assert response_no_filter_json == response_official_false_json, "official=false parameter should return all feeds"
-    # 2 - Test with official=true should return at least one feed
+    assert (
+        len(response_official_false_json) == false_feeds_count
+    ), f"official=false should return {false_feeds_count} feeds but got {len(response_official_false_json)}"
+    assert not any(
+        response["id"] in official_feeds for response in response_official_false_json
+    ), f"official=false expected no feed with stable_id {official_feeds} since it is official"
+
+    # 3 - Test with official=true
     response = client.request(
         "GET",
         endpoint,
@@ -281,7 +317,8 @@ def test_feeds_filter_by_official(client: TestClient, endpoint):
     )
     assert response.status_code == 200
     json_response = response.json()
-    assert len(json_response) < len(response_no_filter_json), "Not all feeds are official"
+    assert len(json_response) == true_feeds_count, f"official=true should return {true_feeds_count} feeds"
+    assert json_response[0]["id"] in official_feeds, f"official=true should return {official_feeds}"
 
 
 def test_non_existent_gtfs_rt_feed_get(client: TestClient):