used http call to the refresh function in other GCP functions

Author: qcdyx

functions-python/backfill_dataset_service_date_range/src/main.py

@@ -4,7 +4,7 @@
 
 from shared.helpers.logger import init_logger
 
-from shared.database.database import with_db_session, refresh_materialized_view
+from shared.database.database import with_db_session
 
 from sqlalchemy.orm import joinedload, Session
 from sqlalchemy import or_, func
@@ -17,13 +17,14 @@
 from shared.database_gen.sqlacodegen_models import (
     Gtfsdataset,
     Validationreport,
-    t_feedsearch,
 )
 
-import requests
+import requests as http_requests
 import json
 
 from google.cloud import storage
+from google.auth.transport import requests
+from google.oauth2 import id_token
 
 env = os.getenv("ENV", "dev").lower()
 bucket_name = f"mobilitydata-datasets-{env}"
@@ -146,7 +147,27 @@ def backfill_datasets(session: "Session"):
                 try:
                     changes_count = 0
                     session.commit()
-                    refresh_materialized_view(session, t_feedsearch.name)
+                    # Replace direct call to refresh_materialized_view with HTTP request to the refresh function
+                    refresh_url = os.getenv("FUNCTION_URL_REFRESH_MV")
+                    if not refresh_url:
+                        raise ValueError(
+                            "FUNCTION_URL_REFRESH_MV environment variable is not set"
+                        )
+
+                    # Create an authorized request
+                    auth_req = requests.Request()
+
+                    # Get an identity token for the target URL
+                    token = id_token.fetch_id_token(auth_req, refresh_url)
+
+                    # Make the HTTP request with the ID token
+                    headers = {"Authorization": f"Bearer {token}"}
+                    response = http_requests.get(refresh_url, headers=headers)
+
+                    response.raise_for_status()
+                    logging.info(
+                        "Materialized view refresh event triggered successfully."
+                    )
                     logging.info(f"{changes_count} elements committed.")
                 except Exception as e:
                     logging.error("Error committing changes:", e)

functions-python/batch_process_dataset/src/main.py

@@ -18,7 +18,7 @@
 import json
 import os
 import random
-import requests
+import requests as http_requests
 import uuid
 import zipfile
 from dataclasses import dataclass
@@ -30,7 +30,7 @@
 from google.cloud import storage
 from sqlalchemy import func
 
-from shared.database_gen.sqlacodegen_models import Gtfsdataset, t_feedsearch
+from shared.database_gen.sqlacodegen_models import Gtfsdatasets
 from shared.dataset_service.main import DatasetTraceService, DatasetTrace, Status
 from shared.database.database import with_db_session
 import logging
@@ -39,6 +39,10 @@
 from shared.helpers.utils import download_and_get_hash
 from sqlalchemy.orm import Session
 
+from google.auth.transport import requests
+from google.oauth2 import id_token
+
+
 init_logger()
 
 
@@ -250,19 +254,25 @@ def create_dataset(self, dataset_file: DatasetFile, db_session: Session):
             db_session.commit()
             self.logger.info(f"[{self.feed_stable_id}] Dataset created successfully.")
 
-            # Replace direct call to refresh_materialized_view with HTTP request to GCP function
-            function_url = os.getenv("FUNCTION_URL_REFRESH_MV")
-            if not function_url:
+            # Replace direct call to refresh_materialized_view with HTTP request to the refresh function
+            refresh_url = os.getenv("FUNCTION_URL_REFRESH_MV")
+            if not refresh_url:
                 raise ValueError(
                     "FUNCTION_URL_REFRESH_MV environment variable is not set"
                 )
 
-            response = requests.get(f"{function_url}/refresh-materialized-view")
+            # Create an authorized request
+            auth_req = requests.Request()
+
+            # Get an identity token for the target URL
+            token = id_token.fetch_id_token(auth_req, refresh_url)
+
+            # Make the HTTP request with the ID token
+            headers = {"Authorization": f"Bearer {token}"}
+            response = http_requests.get(refresh_url, headers=headers)
+
             response.raise_for_status()
-            self.logger.info(
-                f"[{self.feed_stable_id}] Materialized view refresh event triggered "
-                "successfully."
-            )
+            self.logger.info("Materialized view refresh event triggered successfully.")
         except Exception as e:
             raise Exception(f"Error creating dataset: {e}")
 

functions-python/helpers/feed_status.py

@@ -1,13 +1,17 @@
 import logging
 from datetime import datetime, timezone
 from sqlalchemy import text
-from shared.database_gen.sqlacodegen_models import Gtfsdataset, Feed, t_feedsearch
-from shared.database.database import refresh_materialized_view
+import requests as http_requests
+import os
+from shared.database_gen.sqlacodegen_models import Gtfsdataset, Feed
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
 
+from google.auth.transport import requests
+from google.oauth2 import id_token
+
 
 #  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", stable_feed_ids: list[str]):
@@ -75,7 +79,23 @@ def get_filters(status: str):
 
     try:
         session.commit()
-        refresh_materialized_view(session, t_feedsearch.name)
+        # Replace direct call to refresh_materialized_view with HTTP request to the refresh function
+        refresh_url = os.getenv("FUNCTION_URL_REFRESH_MV")
+        if not refresh_url:
+            raise ValueError("FUNCTION_URL_REFRESH_MV environment variable is not set")
+
+        # Create an authorized request
+        auth_req = requests.Request()
+
+        # Get an identity token for the target URL
+        token = id_token.fetch_id_token(auth_req, refresh_url)
+
+        # Make the HTTP request with the ID token
+        headers = {"Authorization": f"Bearer {token}"}
+        response = http_requests.get(refresh_url, headers=headers)
+
+        response.raise_for_status()
+        logging.info("Materialized view refresh event triggered successfully.")
         logging.info("Feed Database changes for status committed.")
         logging.info("Status Changes: %s", diff_counts)
         session.close()

functions-python/reverse_geolocation/src/reverse_geolocation_processor.py

@@ -32,7 +32,6 @@
     Osmlocationgroup,
     Feedosmlocationgroup,
     Location,
-    t_feedsearch,
     Gtfsdataset,
     Gtfsfeed,
 )
@@ -380,7 +379,7 @@ def extract_location_aggregates(
     # Commit the changes to the database before refreshing the materialized view
     db_session.commit()
 
-    # Replace direct call to refresh_materialized_view with HTTP request to GCP function
+    # Replace direct call to refresh_materialized_view with HTTP request to the refresh function
     refresh_url = os.getenv("FUNCTION_URL_REFRESH_MV")
     if not refresh_url:
         raise ValueError("FUNCTION_URL_REFRESH_MV environment variable is not set")