feat(odpt): import Toei bus and train feeds absent from the portal API

functions-python/tasks_executor/src/tasks/data_import/odpt/import_odpt_feeds.py

@@ -67,6 +67,48 @@
     "CC0": "https://creativecommons.org/publicdomain/zero/1.0/",
 }
 
+# Feeds that ODPT serves from api-public.odpt.org but never advertises through
+# ODPT_METADATA_API: they are published via ODPT's token-gated developer catalog
+# instead, under a different path namespace (/api/v4/files/<org>/data/<file>.zip)
+# that PUBLIC_GTFS_ENDPOINT cannot express -- note the absent "odpt" segment and
+# "date" parameter. They carry an explicit "gtfs_endpoint" for that reason.
+#
+# These are declared as ordinary feed items, rather than seeded straight into the DB,
+# so they flow through the same per-item processing as portal feeds -- in particular
+# so their stable_ids reach processed_stable_ids and survive the stale sweep.
+STATIC_FEEDS: Final[List[dict]] = [
+    {
+        "org_label": "Toei",
+        "dataset_label": "ToeiBus",
+        "org_name_ja": "東京都交通局",
+        "org_name_en": "Tokyo Metropolitan Bureau of Transportation",
+        "dataset_name_ja": "都営バス",
+        "dataset_name_en": "Toei Bus",
+        "license_type": "CC BY 4.0",
+        "gtfs_endpoint": (
+            "https://api-public.odpt.org/api/v4/files/Toei/data/ToeiBus-GTFS.zip"
+        ),
+        "vehicle_endpoint": "https://api-public.odpt.org/api/v4/gtfs/realtime/ToeiBus",
+        "trip_update": None,
+        "alert": None,
+    },
+    {
+        "org_label": "Toei",
+        "dataset_label": "ToeiTrain",
+        "org_name_ja": "東京都交通局",
+        "org_name_en": "Tokyo Metropolitan Bureau of Transportation",
+        "dataset_name_ja": "都営地下鉄・日暮里舎人ライナー・都電荒川線",
+        "dataset_name_en": "Toei Subway, Nippori-Toneri Liner and Toden Arakawa Line",
+        "license_type": "CC BY 4.0",
+        "gtfs_endpoint": (
+            "https://api-public.odpt.org/api/v4/files/Toei/data/Toei-Train-GTFS.zip"
+        ),
+        "vehicle_endpoint": "https://api-public.odpt.org/api/v4/gtfs/realtime/toei_odpt_train_vehicle",
+        "trip_update": "https://api-public.odpt.org/api/v4/gtfs/realtime/toei_odpt_train_trip_update",
+        "alert": "https://api-public.odpt.org/api/v4/gtfs/realtime/toei_odpt_train_alert",
+    },
+]
+
 
 def import_odpt_handler(payload: dict | None = None) -> dict:
     """
@@ -494,7 +536,7 @@ def _import_odpt(db_session: Session, dry_run: bool = True) -> dict:
 
     # Fetch list
     try:
-        feeds_list = _fetch_feeds(session_http)
+        portal_feeds = _fetch_feeds(session_http)
     except Exception as e:
         logger.exception("Exception during ODPT_METADATA_API request")
         return {
@@ -510,6 +552,20 @@ def _import_odpt(db_session: Session, dry_run: bool = True) -> dict:
             "total_processed_items": 0,
         }
 
+    # Appended here rather than inside _fetch_feeds so that `portal_feeds` remains a
+    # faithful record of what the source actually returned. The stale sweep below keys
+    # off that, and must not read a list made up solely of our own static entries as
+    # evidence that the portal answered.
+    # Copied so a feed item mutated downstream can't corrupt the module-level constant
+    # for the next invocation -- Cloud Function instances are reused between runs.
+    feeds_list = portal_feeds + [dict(feed) for feed in STATIC_FEEDS]
+    logger.info(
+        "Feed list assembled: %d from portal + %d static = %d total",
+        len(portal_feeds),
+        len(STATIC_FEEDS),
+        len(feeds_list),
+    )
+
     logger.info(
         "Commit batch size (env COMMIT_BATCH_SIZE)=%s",
         os.getenv("COMMIT_BATCH_SIZE", "5"),
@@ -576,7 +632,7 @@ def _import_odpt(db_session: Session, dry_run: bool = True) -> dict:
     # Deprecate feeds the source no longer advertises. Run unconditionally so a
     # dry run can report what *would* be deprecated; the dry-run rollback below
     # still guarantees nothing persists.
-    if feeds_list:
+    if portal_feeds:
         newly_deprecated = deprecate_stale_feeds(
             db_session, "odpt-", processed_stable_ids
         )
@@ -587,7 +643,7 @@ def _import_odpt(db_session: Session, dry_run: bool = True) -> dict:
         # odpt- catalog in one run so that is avoided by setting newly_deprecated empty
         newly_deprecated = []
         logger.warning(
-            "Skipping stale-feed deprecation sweep: fetch returned zero feeds; "
+            "Skipping stale-feed deprecation sweep: the portal returned zero feeds; "
             "refusing to deprecate the entire odpt- catalog."
         )
 

functions-python/tasks_executor/tests/tasks/data_import/test_odpt_import.py

@@ -28,8 +28,21 @@
 from tasks.data_import.odpt.import_odpt_feeds import (
     import_odpt_handler,
     _get_license_url,
+    STATIC_FEEDS,
+    URLS_TO_ENTITY_TYPES_MAP,
 )
 
+
+def _static_rt_entries():
+    """(stable_id_base, entity_type, url) for every RT endpoint STATIC_FEEDS declares."""
+    return [
+        (f"odpt-{feed['org_label']}-{feed['dataset_label']}", entity_type, feed[field])
+        for feed in STATIC_FEEDS
+        for field, entity_type in URLS_TO_ENTITY_TYPES_MAP.items()
+        if feed.get(field)
+    ]
+
+
 GTFS_ENDPOINT_TMPL = (
     "https://api-public.odpt.org/api/v4/files/odpt/{}/{}.zip?date=current"
 )
@@ -313,6 +326,11 @@ def test_import_creates_gtfs_rt_and_location(self, db_session: Session):
             return_value=_FakeSessionOK(),
         ), patch(
             "tasks.data_import.odpt.import_odpt_feeds.REQUEST_TIMEOUT_S", 0.01
+        ), patch(
+            # This test is about the portal payload; keep the static feeds out of its
+            # counts. TestStaticFeeds below covers them directly.
+            "tasks.data_import.odpt.import_odpt_feeds.STATIC_FEEDS",
+            [],
         ), patch(
             # commit_changes (and the side effects it triggers) now lives in
             # data_import_utils, shared with jbda/tdg -- patch it there, not on
@@ -458,6 +476,11 @@ def test_import_dry_run_true_does_not_write_to_db(self, db_session: Session):
                 return_value=_FakeSessionDryRun(),
             ), patch(
                 "tasks.data_import.odpt.import_odpt_feeds.REQUEST_TIMEOUT_S", 0.01
+            ), patch(
+                # This test is about the portal payload; keep the static feeds out of
+                # its counts. TestStaticFeeds below covers them directly.
+                "tasks.data_import.odpt.import_odpt_feeds.STATIC_FEEDS",
+                [],
             ), patch(
                 "tasks.data_import.data_import_utils.trigger_dataset_download",
                 mock_trigger,
@@ -544,6 +567,11 @@ def test_reappeared_feed_is_reactivated(self, db_session: Session):
                 return_value=_FakeSessionReactivate(),
             ), patch(
                 "tasks.data_import.odpt.import_odpt_feeds.REQUEST_TIMEOUT_S", 0.01
+            ), patch(
+                # This test is about the portal payload; keep the static feeds out of
+                # its counts. TestStaticFeeds below covers them directly.
+                "tasks.data_import.odpt.import_odpt_feeds.STATIC_FEEDS",
+                [],
             ), patch(
                 # Not a sweep test; keep it off other tests' rows.
                 "tasks.data_import.odpt.import_odpt_feeds.deprecate_stale_feeds",
@@ -620,6 +648,11 @@ def test_withdrawn_rt_subfeed_is_deprecated_without_touching_siblings(
                 return_value=_FakeSessionRtGranularity(),
             ), patch(
                 "tasks.data_import.odpt.import_odpt_feeds.REQUEST_TIMEOUT_S", 0.01
+            ), patch(
+                # This test is about the portal payload; keep the static feeds out of
+                # its counts. TestStaticFeeds below covers them directly.
+                "tasks.data_import.odpt.import_odpt_feeds.STATIC_FEEDS",
+                [],
             ), patch(
                 "tasks.data_import.data_import_utils.trigger_dataset_download",
                 MagicMock(),
@@ -680,6 +713,11 @@ def test_empty_fetch_does_not_deprecate_the_whole_catalog(
                 return_value=_FakeSessionEmpty(),
             ), patch(
                 "tasks.data_import.odpt.import_odpt_feeds.REQUEST_TIMEOUT_S", 0.01
+            ), patch(
+                # This test is about the portal payload; keep the static feeds out of
+                # its counts. TestStaticFeeds below covers them directly.
+                "tasks.data_import.odpt.import_odpt_feeds.STATIC_FEEDS",
+                [],
             ), patch(
                 "tasks.data_import.data_import_utils.trigger_dataset_download",
                 MagicMock(),
@@ -722,5 +760,214 @@ def test_import_http_failure_graceful(self, db_session: Session):
         self.assertEqual(out["total_processed_items"], 0)
 
 
+# ─────────────────────────────────────────────────────────────────────────────
+# Static (non-portal) feed tests
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestStaticFeeds(unittest.TestCase):
+    """
+    STATIC_FEEDS covers feeds ODPT hosts but never lists in ODPT_METADATA_API, so
+    they cannot be discovered and their URLs cannot be derived from
+    PUBLIC_GTFS_ENDPOINT. Every test here drives the real constant rather than a
+    fixture: the point is that the shipped entries behave correctly.
+    """
+
+    def test_static_feeds_are_well_formed(self):
+        """
+        Guards the contract _process_feed relies on: a literal gtfs_endpoint (there is
+        no template that can build these paths), a license_type _get_license_url can
+        resolve, and a unique org/dataset pair, since that pair becomes the stable_id.
+        """
+        self.assertTrue(STATIC_FEEDS)
+        seen = set()
+        for feed in STATIC_FEEDS:
+            with self.subTest(feed=feed.get("dataset_label")):
+                self.assertTrue(feed["org_label"])
+                self.assertTrue(feed["dataset_label"])
+                self.assertTrue(feed["gtfs_endpoint"].startswith("https://"))
+                self.assertIsNotNone(_get_license_url(feed["license_type"]))
+                # RT fields are optional, but a declared one must be a real URL --
+                # a stray falsy-but-present value would silently skip the sub-feed.
+                for field in URLS_TO_ENTITY_TYPES_MAP:
+                    url = feed[field]
+                    if url is not None:
+                        self.assertTrue(url.startswith("https://"))
+                key = (feed["org_label"], feed["dataset_label"])
+                self.assertNotIn(key, seen)
+                seen.add(key)
+
+    @with_db_session(db_url=default_db_url)
+    def test_static_feeds_are_imported_with_their_literal_urls(
+        self, db_session: Session
+    ):
+        """
+        Driven with an empty portal response so the only feeds processed are the static
+        ones -- if these rows exist afterwards, they came from STATIC_FEEDS alone.
+        """
+        try:
+            with patch(
+                "tasks.data_import.odpt.import_odpt_feeds.requests.Session",
+                return_value=_FakeSessionEmpty(),
+            ), patch(
+                "tasks.data_import.odpt.import_odpt_feeds.REQUEST_TIMEOUT_S", 0.01
+            ), patch(
+                "tasks.data_import.data_import_utils.trigger_dataset_download",
+                MagicMock(),
+            ), patch(
+                "tasks.data_import.data_import_utils.create_web_revalidation_task",
+                MagicMock(),
+            ), patch.dict(
+                os.environ, {"ENVIRONMENT": "test"}, clear=False
+            ):
+                result = import_odpt_handler({"dry_run": False})
+
+            rt_entries = _static_rt_entries()
+            self.assertEqual(result["total_processed_items"], len(STATIC_FEEDS))
+            self.assertEqual(result["created_gtfs"], len(STATIC_FEEDS))
+            # Derived from the constant rather than hardcoded, so adding or removing
+            # an RT endpoint doesn't silently invalidate this test.
+            self.assertEqual(result["created_gtfs_rt"], len(rt_entries))
+            self.assertEqual(result["linked_refs"], len(rt_entries))
+
+            db_session.expire_all()
+            for feed in STATIC_FEEDS:
+                stable_id = f"odpt-{feed['org_label']}-{feed['dataset_label']}"
+                with self.subTest(stable_id=stable_id):
+                    row = (
+                        db_session.query(Gtfsfeed)
+                        .filter(Gtfsfeed.stable_id == stable_id)
+                        .one()
+                    )
+                    row = db_session.merge(row)
+                    # The literal URL must survive untouched: the whole reason these
+                    # entries exist is that PUBLIC_GTFS_ENDPOINT cannot express them.
+                    self.assertEqual(row.producer_url, feed["gtfs_endpoint"])
+                    self.assertNotIn("/files/odpt/", row.producer_url)
+                    self.assertEqual(row.feed_name, feed["dataset_name_ja"])
+                    self.assertEqual(row.provider, feed["org_name_ja"])
+                    self.assertEqual(
+                        row.license_url, _get_license_url(feed["license_type"])
+                    )
+                    self.assertEqual(row.status, "active")
+                    self.assertEqual(row.operational_status, "published")
+
+                    externalids = list(row.externalids)
+                    self.assertEqual(len(externalids), 1)
+                    self.assertEqual(externalids[0].source, "odpt")
+                    self.assertEqual(
+                        externalids[0].associated_id,
+                        f"{feed['org_label']}-{feed['dataset_label']}",
+                    )
+
+                    locations = list(row.locations)
+                    self.assertEqual(len(locations), 1)
+                    self.assertEqual(locations[0].country, "Japan")
+
+            for stable_id_base, entity_type, url in _static_rt_entries():
+                rt_stable_id = f"{stable_id_base}-{entity_type}"
+                with self.subTest(stable_id=rt_stable_id):
+                    rt = (
+                        db_session.query(Gtfsrealtimefeed)
+                        .filter(Gtfsrealtimefeed.stable_id == rt_stable_id)
+                        .one()
+                    )
+                    rt = db_session.merge(rt)
+                    self.assertEqual(rt.producer_url, url)
+                    self.assertEqual([et.name for et in rt.entitytypes], [entity_type])
+                    # Must be linked back to its schedule feed, otherwise the RT feed
+                    # is orphaned in the catalog.
+                    self.assertEqual(
+                        [sched.stable_id for sched in rt.gtfs_feeds], [stable_id_base]
+                    )
+        finally:
+            _delete_feeds_like(db_session, "odpt-Toei-%")
+
+    @with_db_session(db_url=default_db_url)
+    def test_static_feeds_are_marked_as_seen_for_the_stale_sweep(
+        self, db_session: Session
+    ):
+        """
+        The static feeds are absent from the portal by definition, so unless they are
+        registered as processed the sweep would deprecate them on every single run.
+        """
+        mock_sweep = MagicMock(return_value=[])
+        with patch(
+            "tasks.data_import.odpt.import_odpt_feeds.requests.Session",
+            return_value=_FakeSessionOK(),
+        ), patch(
+            "tasks.data_import.odpt.import_odpt_feeds.REQUEST_TIMEOUT_S", 0.01
+        ), patch(
+            "tasks.data_import.odpt.import_odpt_feeds.deprecate_stale_feeds",
+            mock_sweep,
+        ):
+            import_odpt_handler({"dry_run": True})
+
+        mock_sweep.assert_called_once()
+        processed_stable_ids = mock_sweep.call_args.args[2]
+        for feed in STATIC_FEEDS:
+            self.assertIn(
+                f"odpt-{feed['org_label']}-{feed['dataset_label']}",
+                processed_stable_ids,
+            )
+        # The RT sub-feeds are swept by the same prefix match, so they need marking
+        # just as much as the schedule feeds do.
+        for stable_id_base, entity_type, _url in _static_rt_entries():
+            self.assertIn(f"{stable_id_base}-{entity_type}", processed_stable_ids)
+
+    @with_db_session(db_url=default_db_url)
+    def test_static_feeds_do_not_defeat_the_empty_fetch_guard(
+        self, db_session: Session
+    ):
+        """
+        Regression guard. STATIC_FEEDS makes the merged feed list permanently
+        non-empty, so gating the sweep on it -- rather than on the portal response --
+        would silently re-open the catastrophic case the guard exists to prevent: an
+        empty-but-successful portal response deprecating every odpt-* feed except the
+        static ones.
+        """
+        stable_id = "odpt-EmptyFetchStaticOrg-survivor_dataset"
+        try:
+            _seed_feed(
+                db_session,
+                Gtfsfeed,
+                stable_id,
+                "gtfs",
+                status="active",
+                operational_status="published",
+            )
+            db_session.commit()
+
+            with patch(
+                "tasks.data_import.odpt.import_odpt_feeds.requests.Session",
+                return_value=_FakeSessionEmpty(),
+            ), patch(
+                "tasks.data_import.odpt.import_odpt_feeds.REQUEST_TIMEOUT_S", 0.01
+            ), patch(
+                "tasks.data_import.data_import_utils.trigger_dataset_download",
+                MagicMock(),
+            ), patch(
+                "tasks.data_import.data_import_utils.create_web_revalidation_task",
+                MagicMock(),
+            ), patch.dict(
+                os.environ, {"ENVIRONMENT": "test"}, clear=False
+            ):
+                result = import_odpt_handler({"dry_run": False})
+
+            # The static feeds were processed...
+            self.assertEqual(result["total_processed_items"], len(STATIC_FEEDS))
+            # ...but the portal said nothing, so nothing may be swept.
+            self.assertEqual(result["deprecated"], 0)
+
+            db_session.expire_all()
+            survivor = db_session.query(Feed).filter(Feed.stable_id == stable_id).one()
+            self.assertEqual(
+                (survivor.status, survivor.operational_status), ("active", "published")
+            )
+        finally:
+            _delete_feeds_like(db_session, "odpt-EmptyFetchStaticOrg-%")
+            _delete_feeds_like(db_session, "odpt-Toei-%")
+
+
 if __name__ == "__main__":
     unittest.main()