updating triggers

Author: cka-y

functions-python/batch_process_dataset/src/pipeline_tasks.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import os
+from typing import Iterable, List
 
 from google.cloud import tasks_v2
 from sqlalchemy.orm import Session
@@ -66,41 +67,57 @@ def create_http_pmtiles_builder_task(
 
 
 @with_db_session
-def has_file_changed(dataset: Gtfsdataset, file_name: str, db_session: Session) -> bool:
+def get_changed_files(
+    dataset: Gtfsdataset,
+    db_session: Session,
+) -> List[str]:
     """
-    Check if a file has changed in the dataset.
+    Return the subset of `file_names` whose content hash changed compared to the
+    previous dataset for the same feed.
+      - If there is no previous dataset → any file that exists in the new dataset is considered "changed".
+      - If the file existed before and now is missing → NOT considered changed.
+      - If the file did not exist before but exists now → considered changed.
+      - If hashes differ → considered changed.
     """
     previous_dataset = (
         db_session.query(Gtfsdataset)
         .filter(
             Gtfsdataset.feed_id == dataset.feed_id,
             Gtfsdataset.id != dataset.id,
-            Gtfsdataset.latest.is_(False),
         )
         .order_by(Gtfsdataset.downloaded_at.desc())
         .first()
     )
+
+    new_files = list(dataset.gtfsfiles)
+
+    # No previous dataset -> everything that exists now is "changed"
     if not previous_dataset:
-        return True
-    existing_file = next(
-        (file for file in previous_dataset.gtfsfiles if file.file_name == file_name),
-        None,
-    )
-    if not existing_file:
-        return True
-    new_dataset_file = next(
-        (file for file in dataset.gtfsfiles if file.file_name == file_name), None
-    )
-    if not new_dataset_file:
-        return True
-    return existing_file.hash != new_dataset_file.hash
+        return [f.file_name for f in new_files]
+
+    prev_map = {
+        f.file_name: getattr(f, "hash", None) for f in previous_dataset.gtfsfiles
+    }
+
+    changed_files = []
+    for f in new_files:
+        new_hash = getattr(f, "hash", None)
+        old_hash = prev_map.get(f.file_name)
+
+        if old_hash is None or old_hash != new_hash:
+            changed_files.append(f)
+            logging.info(f"Changed file {f.file_name} from {old_hash} to {new_hash}")
+
+    return [f.file_name for f in changed_files]
 
 
 @with_db_session
 def create_pipeline_tasks(dataset: Gtfsdataset, db_session: Session) -> None:
     """
     Create pipeline tasks for a dataset.
     """
+    changed_files = get_changed_files(dataset, db_session=db_session)
+
     stable_id = dataset.feed.stable_id
     dataset_stable_id = dataset.stable_id
     gtfs_files = dataset.gtfsfiles
@@ -110,7 +127,7 @@ def create_pipeline_tasks(dataset: Gtfsdataset, db_session: Session) -> None:
     stops_url = stops_file.hosted_url if stops_file else None
 
     # Create reverse geolocation task
-    if stops_url and has_file_changed(dataset, "stops.txt", db_session):
+    if stops_url and "stops.txt" in changed_files:
         create_http_reverse_geolocation_processor_task(
             stable_id, dataset_stable_id, stops_url
         )
@@ -119,10 +136,23 @@ def create_pipeline_tasks(dataset: Gtfsdataset, db_session: Session) -> None:
         (file for file in gtfs_files if file.file_name == "routes.txt"), None
     )
     # Create PMTiles builder task
+    required_files = {"stops.txt", "routes.txt", "trips.txt", "stop_times.txt"}
+    if not required_files.issubset(set(f.file_name for f in gtfs_files)):
+        logging.info(
+            f"Skipping PMTiles task for dataset {dataset_stable_id} due to missing required files. Required files: "
+            f"{required_files}, available files: {[f.file_name for f in gtfs_files]}"
+        )
+    expected_file_change: Iterable[str] = {
+        "stops.txt",
+        "trips.txt",
+        "routes.txt",
+        "stop_times.txt",
+        "shapes.txt",
+    }
     if (
         routes_file
         and 0 < routes_file.file_size_bytes < 1_000_000
-        and has_file_changed(dataset, "routes.txt", db_session)
+        and not set(changed_files).isdisjoint(expected_file_change)
     ):
         create_http_pmtiles_builder_task(stable_id, dataset_stable_id)
     elif routes_file:

functions-python/batch_process_dataset/tests/test_batch_process_dataset_main.py

@@ -451,6 +451,7 @@ def test_process_dataset_missing_stable_id(self, mock_dataset_trace):
         )
 
     @patch.dict(os.environ, {"DATASETS_BUCKET_NAME": "test-bucket"})
+    @patch("main.create_pipeline_tasks")
     @patch("main.DatasetProcessor.create_dataset_entities")
     @patch("main.DatasetProcessor.upload_files_to_storage")
     @patch("main.DatasetProcessor.unzip_files")
@@ -461,6 +462,7 @@ def test_process_from_bucket_latest_happy_path(
         mock_unzip_files,
         mock_upload_files_to_storage,
         mock_create_dataset_entities,
+        _,
     ):
         # Arrange
         mock_blob = MagicMock()

functions-python/batch_process_dataset/tests/test_pipeline_tasks.py

@@ -6,7 +6,7 @@
 from pipeline_tasks import (
     create_http_reverse_geolocation_processor_task,
     create_http_pmtiles_builder_task,
-    has_file_changed,
+    get_changed_files,
     create_pipeline_tasks,
 )
 
@@ -152,9 +152,8 @@ def test_no_previous_dataset_returns_true(self):
         )
         mock_session = self._make_mock_session_chain(previous_dataset=None)
 
-        # Use __wrapped__ to bypass @with_db_session
-        result = has_file_changed(dataset, "stops.txt", db_session=mock_session)
-        self.assertTrue(result)
+        result = get_changed_files(dataset, db_session=mock_session)
+        self.assertTrue("stops.txt" in result)
 
     def test_previous_without_target_file_returns_true(self):
         prev = SimpleDataset(
@@ -173,10 +172,10 @@ def test_previous_without_target_file_returns_true(self):
         )
         mock_session = self._make_mock_session_chain(previous_dataset=prev)
 
-        result = has_file_changed(dataset, "stops.txt", db_session=mock_session)
-        self.assertTrue(result)
+        result = get_changed_files(dataset, db_session=mock_session)
+        self.assertTrue("stops.txt" in result)
 
-    def test_new_dataset_missing_target_file_returns_true(self):
+    def test_new_dataset_missing_target_file_returns_false(self):
         prev = SimpleDataset(
             feed_id=1,
             dataset_id=9,
@@ -193,8 +192,8 @@ def test_new_dataset_missing_target_file_returns_true(self):
         )
         mock_session = self._make_mock_session_chain(previous_dataset=prev)
 
-        result = has_file_changed(dataset, "stops.txt", db_session=mock_session)
-        self.assertTrue(result)
+        result = get_changed_files(dataset, db_session=mock_session)
+        self.assertFalse("stops.txt" in result)
 
     def test_hash_diff_returns_true(self):
         prev = SimpleDataset(
@@ -213,8 +212,8 @@ def test_hash_diff_returns_true(self):
         )
         mock_session = self._make_mock_session_chain(previous_dataset=prev)
 
-        result = has_file_changed(dataset, "stops.txt", db_session=mock_session)
-        self.assertTrue(result)
+        result = get_changed_files(dataset, db_session=mock_session)
+        self.assertTrue("stops.txt" in result)
 
     def test_hash_same_returns_false(self):
         prev = SimpleDataset(
@@ -233,8 +232,8 @@ def test_hash_same_returns_false(self):
         )
         mock_session = self._make_mock_session_chain(previous_dataset=prev)
 
-        result = has_file_changed(dataset, "stops.txt", db_session=mock_session)
-        self.assertFalse(result)
+        result = get_changed_files(dataset, db_session=mock_session)
+        self.assertFalse("stops.txt" in result)
 
     class TestCreatePipelineTasks(unittest.TestCase):
         @patch.dict(