Feat: update feed status python function (#944)

* update feed status python function

Author: Alessandro100

functions-python/update_feed_status/.coveragerc

@@ -0,0 +1,10 @@
+[run]
+omit =
+    */test*/*
+    */database_gen/*
+    */dataset_service/*
+    */helpers/*
+
+[report]
+exclude_lines =
+    if __name__ == .__main__.:

functions-python/update_feed_status/.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}}

functions-python/update_feed_status/README.md

@@ -0,0 +1,21 @@
+# Update Feed Status
+This directory contains the GCP serverless function that will update all the feed statuses according to their associated latest dataset service date range.
+
+It will exclude Feeds with exisiting status 'deprecated' and 'development'
+
+## 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 feeds which have latest dataset with exisitng values for the service date range
+3. **Feed Update Query**: Update the feed status based on the service date range values comparing vs current date
+
+## 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 update_feed_status`
+
+In postman or similar service, with a `POST` call `v1/update_feed_status`

functions-python/update_feed_status/function_config.json

@@ -0,0 +1,20 @@
+{
+  "name": "update-feed-status",
+  "description": "Update the feed status with the corresponding latest dataset service date range",
+  "entry_point": "update_feed_status",
+  "timeout": 3600,
+  "memory": "1Gi",
+  "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
+}

functions-python/update_feed_status/requirements.txt

@@ -0,0 +1,18 @@
+# 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

functions-python/update_feed_status/requirements_dev.txt

@@ -0,0 +1,2 @@
+Faker
+pytest~=7.4.3

functions-python/update_feed_status/src/__init__.py

@@ -0,0 +1,92 @@
+import logging
+import os
+import functions_framework
+from datetime import date
+from shared.helpers.logger import Logger
+from shared.helpers.database import Database
+from typing import TYPE_CHECKING
+from sqlalchemy import case, text
+from shared.database_gen.sqlacodegen_models import Gtfsdataset, Feed
+
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+logging.basicConfig(level=logging.INFO)
+
+
+#  query to update the status of the feeds based on the service date range of the latest dataset
+def update_feed_statuses_query(session: "Session"):
+    today = date.today()
+
+    latest_dataset_subq = (
+        session.query(
+            Gtfsdataset.feed_id,
+            Gtfsdataset.service_date_range_start,
+            Gtfsdataset.service_date_range_end,
+        )
+        .filter(
+            Gtfsdataset.latest.is_(True),
+            Gtfsdataset.service_date_range_start.isnot(None),
+            Gtfsdataset.service_date_range_end.isnot(None),
+        )
+        .subquery()
+    )
+
+    new_status = case(
+        (
+            latest_dataset_subq.c.service_date_range_end < today,
+            text("'inactive'::status"),
+        ),
+        (
+            latest_dataset_subq.c.service_date_range_start > today,
+            text("'future'::status"),
+        ),
+        (
+            (latest_dataset_subq.c.service_date_range_start <= today)
+            & (latest_dataset_subq.c.service_date_range_end >= today),
+            text("'active'::status"),
+        ),
+    )
+
+    try:
+        updated_count = (
+            session.query(Feed)
+            .filter(
+                Feed.status != text("'deprecated'::status"),
+                Feed.status != text("'development'::status"),
+                Feed.id == latest_dataset_subq.c.feed_id,
+            )
+            .update({Feed.status: new_status}, synchronize_session=False)
+        )
+    except Exception as e:
+        logging.error(f"Error updating feed statuses: {e}")
+        raise Exception(f"Error updating feed statuses: {e}")
+
+    try:
+        session.commit()
+        logging.info("Feed Database changes committed.")
+        session.close()
+        return updated_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 update_feed_status(_):
+    """Updates the Feed status based on the latets dataset service date range."""
+    Logger.init_logger()
+    db = Database(database_url=os.getenv("FEEDS_DATABASE_URL"))
+    update_count = 0
+    try:
+        with db.start_db_session() as session:
+            logging.info("Database session started.")
+            update_count = update_feed_statuses_query(session)
+
+    except Exception as error:
+        logging.error(f"Error updating the feed statuses: {error}")
+        return f"Error updating the feed statuses: {error}", 500
+
+    return f"Script executed successfully. {update_count} feeds updated", 200

functions-python/update_feed_status/src/main.py

@@ -0,0 +1,76 @@
+from unittest.mock import patch, MagicMock
+from test_shared.test_utils.database_utils import default_db_url
+from main import update_feed_status, update_feed_statuses_query
+from datetime import date, timedelta
+
+import os
+
+
+def test_update_feed_status_return():
+    mock_session = MagicMock()
+
+    today = date(2025, 3, 1)
+
+    mock_subquery = MagicMock()
+    mock_subquery.c.feed_id = 1
+    mock_subquery.c.service_date_range_start = today - timedelta(days=10)
+    mock_subquery.c.service_date_range_end = today + timedelta(days=10)
+
+    mock_query = mock_session.query.return_value
+    mock_query.filter.return_value.subquery.return_value = mock_subquery
+
+    mock_update_query = mock_session.query.return_value.filter.return_value
+    mock_update_query.update.return_value = 3
+
+    updated_count = update_feed_statuses_query(mock_session)
+
+    assert updated_count == 3
+    mock_session.commit.assert_called_once()
+
+
+def test_update_feed_status_failed_query():
+    mock_session = MagicMock()
+
+    today = date(2025, 3, 1)
+
+    mock_subquery = MagicMock()
+    mock_subquery.c.feed_id = 1
+    mock_subquery.c.service_date_range_start = today - timedelta(days=10)
+    mock_subquery.c.service_date_range_end = today + timedelta(days=10)
+
+    mock_query = mock_session.query.return_value
+    mock_query.filter.return_value.subquery.return_value = mock_subquery
+
+    mock_update_query = mock_session.query.return_value.filter.return_value
+    mock_update_query.update.side_effect = Exception("Mocked exception")
+
+    try:
+        update_feed_statuses_query(mock_session)
+    except Exception as e:
+        assert str(e) == "Error updating feed statuses: Mocked exception"
+
+
+@patch("main.Logger", autospec=True)
+@patch("main.update_feed_statuses_query")
+def test_updated_feed_status(mock_update_query, mock_logger):
+    mock_update_query.return_value = 5
+
+    with patch.dict(os.environ, {"FEEDS_DATABASE_URL": default_db_url}):
+        response_body, status_code = update_feed_status(None)
+
+    mock_update_query.asser_called_once()
+    assert response_body == "Script executed successfully. 5 feeds updated"
+    assert status_code == 200
+
+
+@patch("main.Logger", autospec=True)
+@patch("main.update_feed_statuses_query")
+def test_updated_feed_status_error_raised(mock_update_query, mock_logger):
+    mock_update_query.side_effect = Exception("Mocked exception")
+
+    with patch.dict(os.environ, {"FEEDS_DATABASE_URL": default_db_url}):
+        response_body, status_code = update_feed_status(None)
+
+    mock_update_query.asser_called_once()
+    assert response_body == "Error updating the feed statuses: Mocked exception"
+    assert status_code == 500

functions-python/update_feed_status/tests/test_update_feed_status_main.py

@@ -56,6 +56,9 @@ locals {
   function_backfill_dataset_service_date_range_config = jsondecode(file("${path.module}/../../functions-python/backfill_dataset_service_date_range/function_config.json"))
   function_backfill_dataset_service_date_range_zip = "${path.module}/../../functions-python/backfill_dataset_service_date_range/.dist/backfill_dataset_service_date_range.zip"
 
+  function_update_feed_status_config = jsondecode(file("${path.module}/../../functions-python/update_feed_status/function_config.json"))
+  function_update_feed_status_zip = "${path.module}/../../functions-python/update_feed_status/.dist/update_feed_status.zip"
+
   function_export_csv_config = jsondecode(file("${path.module}/../../functions-python/export_csv/function_config.json"))
   function_export_csv_zip = "${path.module}/../../functions-python/export_csv/.dist/export_csv.zip"
 }
@@ -69,6 +72,7 @@ locals {
     [for x in local.function_process_validation_report_config.secret_environment_variables : x.key],
     [for x in local.function_update_validation_report_config.secret_environment_variables : x.key],
     [for x in local.function_backfill_dataset_service_date_range_config.secret_environment_variables : x.key],
+    [for x in local.function_update_feed_status_config.secret_environment_variables : x.key],
     [for x in local.function_export_csv_config.secret_environment_variables : x.key]
   )
 
@@ -180,14 +184,20 @@ resource "google_storage_bucket_object" "backfill_dataset_service_date_range_zip
   source = local.function_backfill_dataset_service_date_range_zip
 }
 
-
 # 10. Export CSV
 resource "google_storage_bucket_object" "export_csv_zip" {
   bucket = google_storage_bucket.functions_bucket.name
   name   = "export-csv-${substr(filebase64sha256(local.function_export_csv_zip), 0, 10)}.zip"
   source = local.function_export_csv_zip
 }
 
+# 11. Update Feed Status
+resource "google_storage_bucket_object" "update_feed_status_zip" {
+  bucket = google_storage_bucket.functions_bucket.name
+  name   = "backfill-dataset-service-date-range-${substr(filebase64sha256(local.function_update_feed_status_zip), 0, 10)}.zip"
+  source = local.function_update_feed_status_zip
+}
+
 # Secrets access
 resource "google_secret_manager_secret_iam_member" "secret_iam_member" {
   for_each = local.unique_secret_keys
@@ -893,6 +903,53 @@ resource "google_cloudfunctions2_function" "export_csv" {
   }
 }
 
+# 11. functions/update_feed_status cloud function
+# Updates the Feed statuses based on latest dataset service date range
+
+resource "google_cloudfunctions2_function" "update_feed_status" {
+  name        = local.function_update_feed_status_config.name
+  description = local.function_update_feed_status_config.description
+  location    = var.gcp_region
+  depends_on = [google_secret_manager_secret_iam_member.secret_iam_member]
+  project = var.project_id
+  build_config {
+    runtime     = var.python_runtime
+    entry_point = local.function_update_feed_status_config.entry_point
+    source {
+      storage_source {
+        bucket = google_storage_bucket.functions_bucket.name
+        object = google_storage_bucket_object.update_feed_status_zip.name
+      }
+    }
+  }
+  service_config {
+    available_memory = local.function_update_feed_status_config.memory
+    available_cpu    = local.function_update_feed_status_config.available_cpu
+    timeout_seconds  = local.update_feed_status_config.timeout
+    vpc_connector = data.google_vpc_access_connector.vpc_connector.id
+    vpc_connector_egress_settings = "PRIVATE_RANGES_ONLY"
+
+    environment_variables = {
+      # prevents multiline logs from being truncated on GCP console
+      PYTHONNODEBUGRANGES = 0
+    }
+    dynamic "secret_environment_variables" {
+      for_each = local.function_update_feed_status_config.secret_environment_variables
+      content {
+        key        = secret_environment_variables.value["key"]
+        project_id = var.project_id
+        secret     = lookup(secret_environment_variables.value, "secret", "${upper(var.environment)}_${secret_environment_variables.value["key"]}")
+        version    = "latest"
+      }
+    }
+    service_account_email            = google_service_account.functions_service_account.email
+    max_instance_request_concurrency = local.function_update_feed_status_config.max_instance_request_concurrency
+    max_instance_count               = local.function_update_feed_status_config.max_instance_count
+    min_instance_count               = local.function_update_feed_status_config.min_instance_count
+  }
+}
+
+
 resource "google_cloud_scheduler_job" "export_csv_scheduler" {
   name = "export-csv-scheduler-${var.environment}"
   description = "Schedule the export_csv function"