merge: main

cka-y · cka-y · commit 5a985096243d · 2025-01-25T15:15:33.000-05:00
diff --git a/api/src/feeds/impl/models/gtfs_dataset_impl.py b/api/src/feeds/impl/models/gtfs_dataset_impl.py
@@ -1,12 +1,11 @@
 from functools import reduce
 from typing import List
 
-from packaging.version import Version
-
 from database_gen.sqlacodegen_models import Gtfsdataset, Validationreport
 from feeds.impl.models.bounding_box_impl import BoundingBoxImpl
 from feeds.impl.models.validation_report_impl import ValidationReportImpl
 from feeds_gen.models.gtfs_dataset import GtfsDataset
+from utils.model_utils import compare_java_versions
 
 
 class GtfsDatasetImpl(GtfsDataset):
@@ -30,7 +29,8 @@ def from_orm_latest_validation_report(
         """
         if validation_reports:
             latest_report = reduce(
-                lambda a, b: a if Version(a.validator_version) > Version(b.validator_version) else b, validation_reports
+                lambda a, b: a if compare_java_versions(a.validator_version, b.validator_version) == 1 else b,
+                validation_reports,
             )
             return ValidationReportImpl.from_orm(latest_report)
         return None
diff --git a/api/src/feeds/impl/models/latest_dataset_impl.py b/api/src/feeds/impl/models/latest_dataset_impl.py
@@ -1,12 +1,11 @@
 from functools import reduce
 
-from packaging.version import Version
-
 from database_gen.sqlacodegen_models import Gtfsdataset
 from feeds.impl.models.bounding_box_impl import BoundingBoxImpl
 from feeds.impl.models.validation_report_impl import ValidationReportImpl
 from feeds_gen.models.latest_dataset import LatestDataset
 from feeds_gen.models.latest_dataset_validation_report import LatestDatasetValidationReport
+from utils.model_utils import compare_java_versions
 
 
 class LatestDatasetImpl(LatestDataset):
@@ -28,7 +27,7 @@ def from_orm(cls, dataset: Gtfsdataset | None) -> LatestDataset | None:
         validation_report: LatestDatasetValidationReport | None = None
         if dataset.validation_reports:
             latest_report = reduce(
-                lambda a, b: a if Version(a.validator_version) > Version(b.validator_version) else b,
+                lambda a, b: a if compare_java_versions(a.validator_version, b.validator_version) == 1 else b,
                 dataset.validation_reports,
             )
             (
diff --git a/api/src/utils/model_utils.py b/api/src/utils/model_utils.py
@@ -0,0 +1,26 @@
+from packaging.version import Version
+
+
+def compare_java_versions(v1: str | None, v2: str | None):
+    """
+    Compare two version strings v1 and v2.
+    Returns 1 if v1 > v2, -1 if v1 < v2,
+    otherwise 0.
+    The version strings are expected to be in the format of
+    major.minor.patch[-SNAPSHOT]
+    """
+    if v1 is None and v2 is None:
+        return 0
+    if v1 is None:
+        return -1
+    if v2 is None:
+        return 1
+    # clean version strings replacing the SNAPSHOT suffix with .dev0
+    v1 = v1.replace("-SNAPSHOT", ".dev0")
+    v2 = v2.replace("-SNAPSHOT", ".dev0")
+    if Version(v1) > Version(v2):
+        return 1
+    elif Version(v1) < Version(v2):
+        return -1
+    else:
+        return 0
diff --git a/api/tests/utils/test_compare_java_versions.py b/api/tests/utils/test_compare_java_versions.py
@@ -0,0 +1,27 @@
+import unittest
+from utils.model_utils import compare_java_versions
+
+
+class TestCompareJavaVersions(unittest.TestCase):
+    def test_compare_versions_equal(self):
+        self.assertEqual(compare_java_versions("1.0.0", "1.0.0"), 0)
+        self.assertEqual(compare_java_versions("1.0.0-SNAPSHOT", "1.0.0-SNAPSHOT"), 0)
+
+    def test_compare_versions_v1_greater(self):
+        self.assertEqual(compare_java_versions("1.0.1", "1.0.0"), 1)
+        self.assertEqual(compare_java_versions("1.0.0", "0.9.9"), 1)
+        self.assertEqual(compare_java_versions("1.0.0", "1.0.0-SNAPSHOT"), 1)
+
+    def test_compare_versions_v2_greater(self):
+        self.assertEqual(compare_java_versions("1.0.0", "1.0.1"), -1)
+        self.assertEqual(compare_java_versions("0.9.9", "1.0.0"), -1)
+        self.assertEqual(compare_java_versions("1.0.0-SNAPSHOT", "1.0.0"), -1)
+
+    def test_compare_versions_with_none(self):
+        self.assertEqual(compare_java_versions(None, None), 0)
+        self.assertEqual(compare_java_versions(None, "1.0.0"), -1)
+        self.assertEqual(compare_java_versions("1.0.0", None), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/functions-python/backfill_dataset_service_date_range/.coveragerc b/functions-python/backfill_dataset_service_date_range/.coveragerc
@@ -0,0 +1,10 @@
+[run]
+omit =
+    */test*/*
+    */database_gen/*
+    */dataset_service/*
+    */helpers/*
+
+[report]
+exclude_lines =
+    if __name__ == .__main__.:
diff --git a/functions-python/backfill_dataset_service_date_range/.env.rename_me b/functions-python/backfill_dataset_service_date_range/.env.rename_me
@@ -0,0 +1,2 @@
+# Environment variables for the validation report information extraction to run locally
+export FEEDS_DATABASE_URL=${{FEEDS_DATABASE_URL}}
diff --git a/functions-python/backfill_dataset_service_date_range/README.md b/functions-python/backfill_dataset_service_date_range/README.md
@@ -0,0 +1,21 @@
+# Backfill Dataset Service Date Range
+This directory contains the GCP serverless function that will backfill the GTFS Datasets to include the service date range based off their latest validation report.
+It will only fill `NULL` values and not over write existing values
+
+## Function Workflow
+1. **HTTP Request Trigger**: The function is invoked through an HTTP request that includes identifiers for a dataset and feed.
+2. **Dataset Query**: Retreives all gtfs datasets which have a missing service date range value
+3. **Validation Report Retrieval**: For each dataset, get download the latest validaiton json report to retrieve service date ranges
+4. **Database Update**: Updates the dataset with the values retrieved from the validation report
+
+## Function Configuration
+The function depends on several environment variables:
+- `FEEDS_DATABASE_URL`: The database URL for connecting to the database containing GTFS datasets and related entities.
+
+## Local Development
+Follow standard practices for local development of GCP serverless functions. Refer to the main [README.md](../README.md) for general setup instructions for the development environment.
+
+## Testing
+To run it locally `./scripts/function-python-run.sh --function_name backfill_dataset_service_date_range`
+
+In postman or similar service, with a `POST` call `v1/backfill-dataset-service-date-range`
diff --git a/functions-python/backfill_dataset_service_date_range/function_config.json b/functions-python/backfill_dataset_service_date_range/function_config.json
@@ -0,0 +1,20 @@
+{
+  "name": "backfill-dataset-service-date-range",
+  "description": "For each dataset, backfill the dataset service with the date range of the dataset",
+  "entry_point": "backfill_dataset_service_date_range",
+  "timeout": 3600,
+  "memory": "2Gi",
+  "trigger_http": true,
+  "include_folders": ["helpers"],
+  "include_api_folders": ["database_gen"],
+  "secret_environment_variables": [
+    {
+      "key": "FEEDS_DATABASE_URL"
+    }
+  ],
+  "ingress_settings": "ALLOW_INTERNAL_AND_GCLB",
+  "max_instance_request_concurrency": 1,
+  "max_instance_count": 1,
+  "min_instance_count": 0,
+  "available_cpu": 1
+}
diff --git a/functions-python/backfill_dataset_service_date_range/requirements.txt b/functions-python/backfill_dataset_service_date_range/requirements.txt
@@ -0,0 +1,19 @@
+# Common packages
+functions-framework==3.*
+google-cloud-logging
+psycopg2-binary==2.9.6
+aiohttp~=3.10.5
+asyncio~=3.4.3
+urllib3~=2.2.2
+requests~=2.32.3
+attrs~=23.1.0
+pluggy~=1.3.0
+certifi~=2024.7.4
+
+# SQL Alchemy and Geo Alchemy
+SQLAlchemy==2.0.23
+geoalchemy2==0.14.7
+
+# Google specific packages for this function
+cloudevents~=1.10.1
+google-cloud-storage
diff --git a/functions-python/backfill_dataset_service_date_range/requirements_dev.txt b/functions-python/backfill_dataset_service_date_range/requirements_dev.txt
@@ -0,0 +1,2 @@
+Faker
+pytest~=7.4.3
diff --git a/functions-python/backfill_dataset_service_date_range/src/__init__.py b/functions-python/backfill_dataset_service_date_range/src/__init__.py
diff --git a/functions-python/backfill_dataset_service_date_range/src/main.py b/functions-python/backfill_dataset_service_date_range/src/main.py
@@ -0,0 +1,193 @@
+import logging
+import os
+import functions_framework
+
+from shared.helpers.logger import Logger
+
+from shared.helpers.database import Database
+
+from typing import TYPE_CHECKING
+from sqlalchemy.orm import joinedload
+from sqlalchemy import or_, func
+
+from shared.database_gen.sqlacodegen_models import Gtfsdataset, Validationreport
+
+import requests
+import json
+from datetime import datetime
+
+from google.cloud import storage
+
+env = os.getenv("ENV", "dev").lower()
+bucket_name = f"mobilitydata-datasets-{env}"
+
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+logging.basicConfig(level=logging.INFO)
+
+
+def is_version_gte(target_version: str, version_field):
+    target_parts = func.string_to_array(target_version, ".")
+    version_parts = func.string_to_array(version_field, ".")
+    return func.array_to_string(version_parts, ".") >= func.array_to_string(
+        target_parts, "."
+    )
+
+
+# function backfills the service date range columns in the gtfsdataset table that are null
+# this will not overwrite any existing values
+def backfill_datasets(session: "Session"):
+    # Only care about datasets where service_date_range_start or service_date_range_end are NULL
+    changes_count = 0
+    total_changes_count = 0
+    elements_per_commit = 100
+    # blob setup
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+
+    datasets = (
+        session.query(Gtfsdataset)
+        .options(joinedload(Gtfsdataset.validation_reports))
+        .filter(
+            or_(
+                Gtfsdataset.service_date_range_start.is_(None),
+                Gtfsdataset.service_date_range_end.is_(None),
+            )
+        )
+        .filter(
+            Gtfsdataset.validation_reports.any(
+                is_version_gte("6.0.0", Validationreport.validator_version)
+            )
+        )
+    ).all()
+
+    logging.info(f"Found {len(datasets)} datasets to process.")
+
+    for dataset in datasets:
+        logging.info(f"Processing gtfsdataset ID {dataset.stable_id}")
+        gtfsdataset_id = dataset.stable_id
+        feed_stable_id = "-".join(gtfsdataset_id.split("-")[0:2])
+        # Get the latest validation report for the dataset
+        latest_validation_report = max(
+            dataset.validation_reports,
+            key=lambda report: report.validated_at,
+            default=None,
+        )
+
+        if not latest_validation_report:
+            logging.info(
+                f"Skipping gtfsdataset ID {gtfsdataset_id}: no validation reports found."
+            )
+            continue
+
+        json_report_url = latest_validation_report.json_report
+
+        try:
+            # Download the JSON report
+            blob_url = f"{feed_stable_id}/{gtfsdataset_id}/report_{latest_validation_report.validator_version}.json"
+            logging.info("Blob URL: " + blob_url)
+            dataset_blob = bucket.blob(blob_url)
+            if not dataset_blob.exists():
+                logging.info("Blob not found, downloading from URL")
+                response = requests.get(json_report_url)
+                response.raise_for_status()
+                json_data = response.json()
+            else:
+                logging.info("Blob found, downloading from blob")
+                json_data = json.loads(dataset_blob.download_as_string())
+
+            extracted_service_start_date = (
+                json_data.get("summary", {})
+                .get("feedInfo", {})
+                .get("feedServiceWindowStart", None)
+            )
+            extracted_service_end_date = (
+                json_data.get("summary", {})
+                .get("feedInfo", {})
+                .get("feedServiceWindowEnd", None)
+            )
+
+            try:
+                datetime.strptime(extracted_service_start_date, "%Y-%m-%d")
+            except ValueError:
+                logging.error(
+                    f"""
+                    Key 'summary.feedInfo.feedStartDate' not found or bad value in
+                    JSON for gtfsdataset ID {gtfsdataset_id}. value: {extracted_service_start_date}
+                    """
+                )
+                continue
+
+            try:
+                datetime.strptime(extracted_service_end_date, "%Y-%m-%d")
+            except ValueError:
+                logging.error(
+                    f"""
+                    Key 'summary.feedInfo.feedEndDate' not found or bad value in
+                    JSON for gtfsdataset ID {gtfsdataset_id}. value: {extracted_service_end_date}
+                    """
+                )
+                continue
+
+            dataset.service_date_range_start = extracted_service_start_date
+            dataset.service_date_range_end = extracted_service_end_date
+
+            formatted_dates = (
+                extracted_service_start_date + " - " + extracted_service_end_date
+            )
+            logging.info(
+                f"Updated gtfsdataset ID {gtfsdataset_id} with value: {formatted_dates}"
+            )
+            total_changes_count += 1
+            changes_count += 1
+            if changes_count >= elements_per_commit:
+                try:
+                    changes_count = 0
+                    session.commit()
+                    logging.info(f"{changes_count} elements committed.")
+                except Exception as e:
+                    logging.error("Error committing changes:", e)
+                    session.rollback()
+                    session.close()
+                    raise Exception(f"Error creating dataset: {e}")
+
+        except requests.RequestException as e:
+            logging.error(
+                f"Error downloading JSON for gtfsdataset ID {gtfsdataset_id}: {e}"
+            )
+        except json.JSONDecodeError as e:
+            logging.error(
+                f"Error parsing JSON for gtfsdataset ID {gtfsdataset_id}: {e}"
+            )
+        except Exception as e:
+            logging.error(f"Error processing gtfsdataset ID {gtfsdataset_id}: {e}")
+
+    try:
+        session.commit()
+        logging.info("Database changes committed.")
+        session.close()
+        return total_changes_count
+    except Exception as e:
+        logging.error("Error committing changes:", e)
+        session.rollback()
+        session.close()
+        raise Exception(f"Error creating dataset: {e}")
+
+
+@functions_framework.http
+def backfill_dataset_service_date_range(_):
+    """Fills gtfs dataset service date range from the latest validation report."""
+    Logger.init_logger()
+    db = Database(database_url=os.getenv("FEEDS_DATABASE_URL"))
+    change_count = 0
+    try:
+        with db.start_db_session() as session:
+            logging.info("Database session started.")
+            change_count = backfill_datasets(session)
+
+    except Exception as error:
+        logging.error(f"Error setting the datasets service date range values: {error}")
+        return f"Error setting the datasets service date range values: {error}", 500
+
+    return f"Script executed successfully. {change_count} datasets updated", 200
diff --git a/functions-python/backfill_dataset_service_date_range/tests/test_backfill_dataset_service_date_range_main.py b/functions-python/backfill_dataset_service_date_range/tests/test_backfill_dataset_service_date_range_main.py
diff --git a/infra/functions-python/main.tf b/infra/functions-python/main.tf

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+# Environment variables for the validation report information extraction to run locally`
	`2`	`+export FEEDS_DATABASE_URL=${{FEEDS_DATABASE_URL}}`