feat: add datasets MD5 hash (#1665)

Author: davidgamez

api/src/shared/db_models/gtfs_dataset_impl.py

@@ -47,6 +47,7 @@ def from_orm(cls, gtfs_dataset: Gtfsdataset | None) -> GtfsDataset | None:
             note=gtfs_dataset.note,
             downloaded_at=gtfs_dataset.downloaded_at,
             hash=gtfs_dataset.hash,
+            hash_md5=gtfs_dataset.hash_md5,
             bounding_box=BoundingBoxImpl.from_orm(gtfs_dataset.bounding_box),
             validation_report=cls.from_orm_latest_validation_report(gtfs_dataset.validation_reports),
             service_date_range_start=gtfs_dataset.service_date_range_start,

api/src/shared/db_models/latest_dataset_impl.py

@@ -48,6 +48,7 @@ def from_orm(cls, dataset: Gtfsdataset | None) -> LatestDataset | None:
             service_date_range_end=dataset.service_date_range_end,
             agency_timezone=dataset.agency_timezone,
             hash=dataset.hash,
+            hash_md5=dataset.hash_md5,
             validation_report=validation_report,
             unzipped_folder_size_mb=round(dataset.unzipped_size_bytes / 1024**2, 2)
             if dataset.unzipped_size_bytes

docs/DatabaseCatalogAPI.yaml

@@ -922,9 +922,13 @@ components:
           example: 2023-07-10T22:06:00Z
           format: date-time
         hash:
-          description: A hash of the dataset.
+          description: SHA-256 hash of the dataset.
           type: string
           example: ad3805c4941cd37881ff40c342e831b5f5224f3d8a9a2ec3ac197d3652c78e42
+        hash_md5:
+          description: MD5 hash of the dataset.
+          type: string
+          example: 098f6bcd4621d373cade4e832627b4f6
         service_date_range_start:
           description: The start date of the service date range for the dataset in UTC. Timing starts at 00:00:00 of the day.
           type: string
@@ -1173,9 +1177,13 @@ components:
               example: 2023-07-10T22:06:00Z
               format: date-time
             hash:
-              description: A hash of the dataset.
+              description: SHA-256 hash of the dataset.
               type: string
               example: 6497e85e34390b8b377130881f2f10ec29c18a80dd6005d504a2038cdd00aa71
+            hash_md5:
+              description: MD5 hash of the dataset.
+              type: string
+              example: 098f6bcd4621d373cade4e832627b4f6
             bounding_box:
               $ref: "#/components/schemas/BoundingBox"
             validation_report:

docs/OperationsAPI.yaml

@@ -859,9 +859,13 @@ components:
           example: 2023-07-10T22:06:00Z
           format: date-time
         hash:
-          description: A hash of the dataset.
+          description: SHA-256 hash of the dataset.
           type: string
           example: ad3805c4941cd37881ff40c342e831b5f5224f3d8a9a2ec3ac197d3652c78e42
+        hash_md5:
+          description: MD5 hash of the dataset.
+          type: string
+          example: 098f6bcd4621d373cade4e832627b4f6
         service_date_range_start:
           description: The start date of the service date range for the dataset in UTC. Timing starts at 00:00:00 of the day.
           type: string
@@ -1103,9 +1107,13 @@ components:
               example: 2023-07-10T22:06:00Z
               format: date-time
             hash:
-              description: A hash of the dataset.
+              description: SHA-256 hash of the dataset.
               type: string
               example: 6497e85e34390b8b377130881f2f10ec29c18a80dd6005d504a2038cdd00aa71
+            hash_md5:
+              description: MD5 hash of the dataset.
+              type: string
+              example: 098f6bcd4621d373cade4e832627b4f6
             bounding_box:
               $ref: "#/components/schemas/BoundingBox"
             validation_report:
@@ -1495,6 +1503,7 @@ components:
 
 
 
+
                 * vp - vehicle positions
                 * tu - trip updates
                 * sa - service alerts
@@ -1617,6 +1626,7 @@ components:
 
 
 
+
                 * vp - vehicle positions
                 * tu - trip updates
                 * sa - service alerts
@@ -1696,6 +1706,7 @@ components:
 
 
 
+
           * `active` Feed should be used in public trip planners.
           * `deprecated` Feed is explicitly deprecated and should not be used in public trip planners.
           * `inactive` Feed hasn't been recently updated and should be used at risk of providing outdated information.
@@ -1717,6 +1728,7 @@ components:
 
 
 
+
           * `gtfs` GTFS feed.
           * `gtfs_rt` GTFS-RT feed.
           * `gbfs` GBFS feed.

functions-python/batch_process_dataset/src/main.py

@@ -65,6 +65,7 @@ class DatasetFile:
     stable_id: str
     extracted_files: List[Gtfsfile] = None
     file_sha256_hash: Optional[str] = None
+    file_md5_hash: Optional[str] = None
     hosted_url: Optional[str] = None
     zipped_size: Optional[int] = None
 
@@ -156,22 +157,27 @@ def upload_dataset_zip_to_storage(
         source_file_path,
         dataset_stable_id,
         public=True,
-    ):
+    ) -> Optional[str]:
         """
         Uploads the dataset zip file to GCP storage as latest.zip and versioned zip.
+        Returns the MD5 hash (hex) of the uploaded file as provided by GCS.
         """
         bucket = storage.Client().get_bucket(self.bucket_name)
         target_paths = [
             f"{self.feed_stable_id}/latest.zip",
             f"{self.feed_stable_id}/{dataset_stable_id}/{dataset_stable_id}.zip",
         ]
 
+        md5_hash_hex = None
         for target_path in target_paths:
             blob = bucket.blob(target_path)
             blob.upload_from_filename(source_file_path)
             if public:
                 blob.make_public()
             self.logger.info(f"Uploaded {blob.public_url}")
+            if blob.md5_hash:
+                md5_hash_hex = base64.b64decode(blob.md5_hash).hex()
+        return md5_hash_hex
 
     def _extract_and_upload_single_file(
         self,
@@ -320,7 +326,7 @@ def transfer_dataset(self, feed_id, public=True) -> DatasetFile or None:
                 self.logger.info(
                     f"Creating file {dataset_full_path} in bucket {self.bucket_name}"
                 )
-                self.upload_dataset_zip_to_storage(
+                file_md5_hash = self.upload_dataset_zip_to_storage(
                     temp_file_path,
                     dataset_stable_id,
                     public=public,
@@ -336,6 +342,7 @@ def transfer_dataset(self, feed_id, public=True) -> DatasetFile or None:
                 return DatasetFile(
                     stable_id=dataset_stable_id,
                     file_sha256_hash=file_sha256_hash,
+                    file_md5_hash=file_md5_hash,
                     hosted_url=f"{self.public_hosted_datasets_url}/{dataset_full_path}",
                     extracted_files=extracted_files,
                     zipped_size=(
@@ -459,6 +466,7 @@ def create_dataset_entities(
                     bounding_box=None,
                     note=None,
                     hash=dataset_file.file_sha256_hash,
+                    hash_md5=dataset_file.file_md5_hash,
                     downloaded_at=func.now(),
                     hosted_url=dataset_file.hosted_url,
                     gtfsfiles=(

functions-python/batch_process_dataset/tests/test_batch_process_dataset_main.py

@@ -72,12 +72,8 @@ def test_upload_dataset_diff_hash(
         """
         Test upload_dataset method of DatasetProcessor class with different hash from the latest one
         """
-        mock_blob = MagicMock()
-        mock_blob.public_url = public_url
-        mock_blob.path = public_url
-
-        # Mock the new methods used in transfer_dataset
-        mock_upload_dataset_zip.return_value = mock_blob
+        mock_md5_hex = "098f6bcd4621d373cade4e832627b4f6"
+        mock_upload_dataset_zip.return_value = mock_md5_hex
         mock_extracted_files = []  # Empty list of extracted files
         mock_extract_and_upload.return_value = mock_extracted_files
         mock_download_url_content.return_value = file_hash, True
@@ -105,6 +101,7 @@ def test_upload_dataset_diff_hash(
             f"/feed_stable_id-mocked_timestamp.zip",
         )
         self.assertEqual(result.file_sha256_hash, file_hash)
+        self.assertEqual(result.file_md5_hash, mock_md5_hex)
         # Verify the new methods were called
         self.assertEqual(mock_upload_dataset_zip.call_count, 1)
         self.assertEqual(mock_extract_and_upload.call_count, 1)
@@ -905,3 +902,77 @@ def test_create_dataset_entities_update_existing_no_files(self, db_session):
             self.assertIsNone(result_dataset.unzipped_size_bytes)  # None when no files
 
             mock_refresh_task.assert_called_once()
+
+
+class TestUploadDatasetZipToStorage(unittest.TestCase):
+    @patch("main.storage.Client")
+    def test_md5_hash_read_from_gcs_blob_after_upload(self, mock_storage_client):
+        """
+        upload_dataset_zip_to_storage should return the hex MD5 hash read from the GCS
+        blob's md5_hash attribute (base64-encoded) after upload.
+        """
+        import base64
+        import tempfile
+
+        raw_md5 = b"\x09\x8f\x6b\xcd\x46\x21\xd3\x73\xca\xde\x4e\x83\x26\x27\xb4\xf6"
+        b64_md5 = base64.b64encode(raw_md5).decode()
+        expected_hex = raw_md5.hex()
+
+        mock_blob = MagicMock()
+        mock_blob.md5_hash = b64_md5
+        mock_blob.public_url = "https://storage.googleapis.com/bucket/path.zip"
+        mock_bucket = MagicMock()
+        mock_bucket.blob.return_value = mock_blob
+        mock_storage_client.return_value.get_bucket.return_value = mock_bucket
+
+        processor = DatasetProcessor(
+            producer_url="https://example.com/feed.zip",
+            feed_id="feed_id",
+            feed_stable_id="feed_stable",
+            execution_id="exec_id",
+            latest_hash="hash",
+            bucket_name="test-bucket",
+            authentication_type=0,
+            api_key_parameter_name=None,
+            public_hosted_datasets_url="https://public.example.com",
+        )
+
+        with tempfile.NamedTemporaryFile(suffix=".zip") as tmp:
+            result = processor.upload_dataset_zip_to_storage(
+                tmp.name, "dataset_stable_id"
+            )
+
+        self.assertEqual(result, expected_hex)
+
+    @patch("main.storage.Client")
+    def test_md5_hash_none_when_blob_has_no_md5(self, mock_storage_client):
+        """
+        upload_dataset_zip_to_storage should return None if the GCS blob provides no md5_hash.
+        """
+        import tempfile
+
+        mock_blob = MagicMock()
+        mock_blob.md5_hash = None
+        mock_blob.public_url = "https://storage.googleapis.com/bucket/path.zip"
+        mock_bucket = MagicMock()
+        mock_bucket.blob.return_value = mock_blob
+        mock_storage_client.return_value.get_bucket.return_value = mock_bucket
+
+        processor = DatasetProcessor(
+            producer_url="https://example.com/feed.zip",
+            feed_id="feed_id",
+            feed_stable_id="feed_stable",
+            execution_id="exec_id",
+            latest_hash="hash",
+            bucket_name="test-bucket",
+            authentication_type=0,
+            api_key_parameter_name=None,
+            public_hosted_datasets_url="https://public.example.com",
+        )
+
+        with tempfile.NamedTemporaryFile(suffix=".zip") as tmp:
+            result = processor.upload_dataset_zip_to_storage(
+                tmp.name, "dataset_stable_id"
+            )
+
+        self.assertIsNone(result)

functions-python/tasks_executor/README.md

@@ -93,6 +93,27 @@ To populate licenses:
 }
 ```
 
+To backfill MD5 hashes for existing GTFS datasets (reads the MD5 from the GCS object metadata):
+
+```json
+{
+  "task": "backfill_dataset_hash_md5",
+  "payload": {
+    "dry_run": true,
+    "only_latest": true,
+    "only_missing_hashes": true,
+    "limit": 10
+  }
+}
+```
+
+| Parameter | Type | Default | Description |
+|---|---|---|---|
+| `dry_run` | bool | `true` | Log changes without writing to the database |
+| `only_latest` | bool | `true` | Process only datasets that are the current latest for their feed |
+| `only_missing_hashes` | bool | `true` | Skip datasets that already have `hash_md5` set |
+| `limit` | int \| null | `10` | Maximum number of datasets to process; omit or pass `null` for no limit |
+
 ## Response Content Type
 
 When the request includes the header `Accept: text/csv`, the server returns the response as a CSV file generated from the handler’s output.

functions-python/tasks_executor/src/main.py

@@ -29,6 +29,9 @@
 from tasks.dataset_files.rebuild_missing_dataset_files import (
     rebuild_missing_dataset_files_handler,
 )
+from tasks.dataset_files.backfill_dataset_hash_md5 import (
+    backfill_dataset_hash_md5_handler,
+)
 from tasks.licenses.license_matcher import match_license_handler
 from tasks.missing_bounding_boxes.rebuild_missing_bounding_boxes import (
     rebuild_missing_bounding_boxes_handler,
@@ -109,6 +112,14 @@
         "description": "Rebuilds missing dataset files for GTFS datasets.",
         "handler": rebuild_missing_dataset_files_handler,
     },
+    "backfill_dataset_hash_md5": {
+        "description": (
+            "Backfills the MD5 hash for existing GTFS datasets by reading it from GCS blob metadata. "
+            "Parameters: dry_run (default true), only_latest (default true), "
+            "only_missing_hashes (default true), limit (default 10)."
+        ),
+        "handler": backfill_dataset_hash_md5_handler,
+    },
     "update_geojson_files": {
         "description": "Iterate over bucket looking for {feed_stable_id}/geolocation.geojson and update precision.",
         "handler": update_geojson_files_precision_handler,