Add maximum executions check

Author: davidgamez

functions-python/helpers/utils.py

@@ -17,10 +17,12 @@
 import logging
 import os
 import ssl
+from datetime import date, datetime
+from logging import Logger
+from typing import Optional
 
 import requests
 import urllib3
-from google.cloud import storage
 from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry
 from urllib3.util.ssl_ import create_urllib3_context
@@ -32,6 +34,8 @@ def create_bucket(bucket_name):
     Creates GCP storage bucket if it doesn't exist
     :param bucket_name: name of the bucket to create
     """
+    from google.cloud import storage
+
     storage_client = storage.Client()
     bucket = storage_client.lookup_bucket(bucket_name)
     if bucket is None:
@@ -53,6 +57,8 @@ def download_from_gcs(bucket_name: str, blob_path: str, local_path: str) -> str:
     Returns:
         The absolute path to the downloaded file.
     """
+    from google.cloud import storage
+
     storage_client = storage.Client()
     bucket = storage_client.bucket(bucket_name)
     blob = bucket.blob(blob_path)
@@ -211,3 +217,53 @@ def create_http_task(
         task_time=proto_time,
         http_method=tasks_v2.HttpMethod.POST,
     )
+
+
+def get_execution_id(json_payload: dict, stable_id: Optional[str]) -> str:
+    """
+    Extracts the execution_id from the JSON payload.
+    If not present, defaults to today's date in YYYY-MM-DD format followed by a hyphen and the stable_id if provided.
+    """
+    execution_id = json_payload.get("execution_id")
+    if not execution_id:
+        execution_id = f"{str(date.today())}"
+        if stable_id:
+            execution_id += f"-{stable_id}"
+        else:
+            # Even this should not happen, but just in case we are defaulting it to the current time
+            execution_id += f"-{datetime.now().strftime('%H:%M:%S')}"
+    return execution_id
+
+
+def check_maximum_executions(
+    execution_id: str, stable_id: str, logger: Logger, maximum_executions: int = 1
+) -> str:
+    """
+    Checks if the dataset has been executed more than the maximum allowed times.
+    If it has, returns an error message; otherwise, returns None.
+    :param execution_id: The ID of the execution.
+    :param stable_id: The stable ID of the dataset.
+    :param logger: Logger instance to log messages.
+    :param maximum_executions: The maximum number of allowed executions.
+    :return: Error message if the maximum executions are exceeded, otherwise None.
+    """
+    from shared.dataset_service.main import DatasetTraceService
+
+    trace_service = DatasetTraceService()
+    trace = trace_service.get_by_execution_and_stable_ids(execution_id, stable_id)
+    logger.info(f"Dataset trace: {trace}")
+    executions = len(trace) if trace else 0
+    logger.info(
+        f"Dataset executed times={executions}/{maximum_executions} "
+        f"in execution=[{execution_id}] "
+    )
+
+    if executions > 0:
+        if executions >= maximum_executions:
+            error_message = (
+                f"Function already executed maximum times "
+                f"in execution: [{execution_id}]"
+            )
+            logger.error(error_message)
+            return error_message
+    return None

functions-python/reverse_geolocation/function_config.json

@@ -5,7 +5,7 @@
   "timeout": 540,
   "available_memory": "4Gi",
   "trigger_http": true,
-  "include_folders": ["helpers"],
+  "include_folders": ["helpers", "dataset_service"],
   "include_api_folders": ["database_gen", "database", "common"],
   "environment_variables": [],
   "secret_environment_variables": [

functions-python/reverse_geolocation/src/parse_request.py

@@ -80,7 +80,21 @@ def parse_request_parameters(
     else:
         use_cache = data_type == "gtfs"
         logging.info("No use_cache provided, using(%s): %s", data_type, use_cache)
-    return df, stable_id, dataset_id, data_type, urls, public, strategy, use_cache
+    if "maximum_executions" in request_json:
+        maximum_executions = int(request_json["maximum_executions"])
+    else:
+        maximum_executions = 1
+    return (
+        df,
+        stable_id,
+        dataset_id,
+        data_type,
+        urls,
+        public,
+        strategy,
+        use_cache,
+        maximum_executions,
+    )
 
 
 def parse_request_parameters_gtfs(

functions-python/reverse_geolocation/src/reverse_geolocation_processor.py

@@ -36,6 +36,7 @@
 from shared.helpers.locations import ReverseGeocodingStrategy
 from shared.helpers.logger import get_logger
 from shared.helpers.runtime_metrics import track_metrics
+from shared.helpers.utils import check_maximum_executions, get_execution_id
 from strategy_extraction_per_point import extract_location_aggregates_per_point
 from strategy_extraction_per_polygon import extract_location_aggregates_per_polygon
 
@@ -272,10 +273,21 @@ def reverse_geolocation_process(
             public,
             strategy,
             use_cache,
+            maximum_executions,
         ) = parse_request_parameters(request)
 
         logger = get_logger(__name__, stable_id)
 
+        # Check for maximum executions to avoid repeated processing during the same day
+        request_json = request.get_json(silent=True)
+        execution_id = get_execution_id(request_json, stable_id)
+        max_execution_error = check_maximum_executions(
+            execution_id, stable_id, logger, maximum_executions
+        )
+        if max_execution_error:
+            logger.warning(max_execution_error)
+            return max_execution_error, ERROR_STATUS_CODE
+
         # Remove duplicate lat/lon points
         stops_df["stop_lat"] = pd.to_numeric(stops_df["stop_lat"], errors="coerce")
         stops_df["stop_lon"] = pd.to_numeric(stops_df["stop_lon"], errors="coerce")

functions-python/reverse_geolocation/src/scripts/reverse_geolocation_process_verifier.py

@@ -1,3 +1,8 @@
+# This script is used to verify the reverse geolocation process
+# Before running this script, ensure you have the necessary environment set up:
+# 1. Google DataStore emulator running on localhost:8081 by running:
+#       gcloud beta emulators datastore start --project=your-project-id
+
 import json
 import logging
 import os
@@ -234,12 +239,14 @@ def create_test_data(feed_stable_id: str, feed_dict: Dict, db_session: Session =
         "data_type": feed_dict["data_type"],
         "use_cache": False,
         "public": False,
+        "maximum_executions": 1000,
     }
 
     try:
         os.environ["STORAGE_EMULATOR_HOST"] = f"http://{HOST}:{PORT}"
         os.environ["DATASETS_BUCKET_NAME_GBFS"] = BUCKET_NAME
         os.environ["DATASETS_BUCKET_NAME_GTFS"] = BUCKET_NAME
+        os.environ["DATASTORE_EMULATOR_HOST"] = "localhost:8081"
         server = create_server(
             host=HOST, port=PORT, in_memory=False, default_bucket=BUCKET_NAME
         )

functions-python/reverse_geolocation/tests/test_reverse_geolocation_processor.py

@@ -59,6 +59,7 @@ def test_parse_request_parameters(self, requests_mock):
             public,
             strategy,
             use_cache,
+            maximum_executions,
         ) = parse_request_parameters(request)
         self.assertEqual("test_stable_id", stable_id)
         self.assertEqual("test_dataset_id", dataset_id)
@@ -67,6 +68,8 @@ def test_parse_request_parameters(self, requests_mock):
         self.assertEqual(["test_url"], urls)
         self.assertEqual(True, public)
         self.assertEqual("per-point", strategy)
+        self.assertEqual(True, use_cache)
+        self.assertEqual(1, maximum_executions)
 
         # Exception should be raised
         requests_mock.get.return_value.content = None
@@ -106,6 +109,7 @@ def test_parse_request_parameters_gbfs_station_information(self, requests_mock):
             public,
             strategy,
             use_cache,
+            maximum_executions,
         ) = parse_request_parameters(request)
 
         self.assertEqual("stable123", stable_id)
@@ -115,6 +119,9 @@ def test_parse_request_parameters_gbfs_station_information(self, requests_mock):
         self.assertEqual((2, 2), df.shape)
         self.assertEqual("per-polygon", strategy)
         self.assertEqual(True, public)
+        # Cache is disabled for GBFS data by default
+        self.assertEqual(False, use_cache)
+        self.assertEqual(1, maximum_executions)
 
     @patch("parse_request.requests")
     def test_parse_request_parameters_gbfs_vehicle_status(self, requests_mock):
@@ -136,6 +143,7 @@ def test_parse_request_parameters_gbfs_vehicle_status(self, requests_mock):
             "vehicle_status_url": "http://dummy.vehicle",
             "data_type": "gbfs",
             "public": "False",
+            "maximum_executions": 10,
         }
 
         (
@@ -147,6 +155,7 @@ def test_parse_request_parameters_gbfs_vehicle_status(self, requests_mock):
             public,
             strategy,
             use_cache,
+            maximum_executions,
         ) = parse_request_parameters(request)
 
         self.assertEqual("stable456", stable_id)
@@ -156,6 +165,9 @@ def test_parse_request_parameters_gbfs_vehicle_status(self, requests_mock):
         self.assertEqual((2, 2), df.shape)
         self.assertEqual("per-polygon", strategy)
         self.assertEqual(False, public)
+        # Cache is disabled for GBFS data by default
+        self.assertEqual(False, use_cache)
+        self.assertEqual(10, maximum_executions)
 
     @patch("parse_request.requests")
     def test_parse_request_parameters_invalid_request(self, requests_mock):
@@ -406,15 +418,21 @@ def test_update_dataset_bounding_box_exception(self, db_session):
     @patch("reverse_geolocation_processor.update_dataset_bounding_box")
     @patch("reverse_geolocation_processor.reverse_geolocation")
     @patch("reverse_geolocation_processor.create_geojson_aggregate")
+    @patch("reverse_geolocation_processor.check_maximum_executions")
+    @patch("reverse_geolocation_processor.get_execution_id")
     def test_valid_request(
         self,
+        mock_get_execution_id,
+        mock_check_maximum_executions,
         mock_create_geojson_aggregate,
         mock_reverse_geolocation,
         mock_update_bounding_box,
         mock_parse_request_parameters,
     ):
         from reverse_geolocation_processor import reverse_geolocation_process
 
+        mock_get_execution_id.return_value = "test_execution_id"
+        mock_check_maximum_executions.return_value = None
         # Mocking the parsed request parameters
         mock_parse_request_parameters.return_value = (
             pd.DataFrame({"stop_lat": [1.0], "stop_lon": [1.0]}),
@@ -425,6 +443,7 @@ def test_valid_request(
             True,
             "per-point",
             False,
+            1,
         )
         mock_update_bounding_box.return_value = MagicMock()
         mock_reverse_geolocation.return_value = {"group_id": MagicMock()}
@@ -464,14 +483,21 @@ def test_invalid_request(self, mock_parse_request_parameters):
     @patch("reverse_geolocation_processor.parse_request_parameters")
     @patch("reverse_geolocation_processor.update_dataset_bounding_box")
     @patch("reverse_geolocation_processor.reverse_geolocation")
+    @patch("reverse_geolocation_processor.check_maximum_executions")
+    @patch("reverse_geolocation_processor.get_execution_id")
     def test_exception_handling(
         self,
+        mock_check_get_execution_id,
+        mock_check_maximum_executions,
         mock_reverse_geolocation,
         mock_update_bounding_box,
         mock_parse_request_parameters,
     ):
         from reverse_geolocation_processor import reverse_geolocation_process
 
+        mock_check_get_execution_id.return_value = "test_execution_id"
+        mock_check_maximum_executions.return_value = None
+        # mock_dataset_service.get_by_execution_and_stable_ids.return_value = 0
         # Mocking the parsed request parameters
         mock_parse_request_parameters.return_value = (
             pd.DataFrame({"stop_lat": [1.0], "stop_lon": [1.0]}),
@@ -482,6 +508,7 @@ def test_exception_handling(
             True,
             "per-point",
             False,
+            1,
         )
         mock_update_bounding_box.side_effect = Exception("Unexpected error")
 
@@ -499,12 +526,18 @@ def test_exception_handling(
         mock_reverse_geolocation.assert_not_called()
 
     @patch("reverse_geolocation_processor.parse_request_parameters")
+    @patch("reverse_geolocation_processor.check_maximum_executions")
+    @patch("reverse_geolocation_processor.get_execution_id")
     def test_valid_request_empty_stops(
         self,
+        mock_get_execution_id,
+        mock_check_maximum_executions,
         mock_parse_request_parameters,
     ):
         from reverse_geolocation_processor import reverse_geolocation_process
 
+        mock_get_execution_id.return_value = "test_execution_id"
+        mock_check_maximum_executions.return_value = None
         # Mocking the parsed request parameters
         mock_parse_request_parameters.return_value = (
             pd.DataFrame({"stop_lat": [], "stop_lon": []}),
@@ -515,6 +548,7 @@ def test_valid_request_empty_stops(
             True,
             "per-point",
             False,
+            1,
         )
 
         # Mocking a Flask request