updated code

Author: qcdyx

functions-python/refresh_materialized_view/requirements.txt

@@ -1,9 +1,26 @@
-fastapi==0.110.0
-uvicorn==0.29.0
-sqlalchemy==2.0.23
-geoalchemy2==0.14.7
-python-dotenv~=1.0.0
-requests
-psycopg2-binary
+# Common packages
+functions-framework==3.*
 google-cloud-logging
-aiohttp
+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
+fastapi>=0.68.0
+uvicorn>=0.15.0
+functions-framework>=3.0.0
+asgiref>=3.4.1
+google-cloud-tasks>=2.11.0
+
+# SQL Alchemy and Geo Alchemy
+SQLAlchemy==2.0.23
+geoalchemy2==0.14.7
+
+# Google specific packages for this function
+cloudevents~=1.10.1
+
+# Configuration
+python-dotenv==1.0.0

functions-python/refresh_materialized_view/src/main.py

@@ -1,29 +1,30 @@
 import logging
 import os
-import functions_framework
-from flask import Request, jsonify
+import json
 from google.cloud import tasks_v2
 from datetime import datetime
+import functions_framework
 from shared.helpers.logger import init_logger
 from shared.database.database import with_db_session
 
 init_logger()
 
 
 @functions_framework.http
-def refresh_materialized_view_function(request: Request):
+def refresh_materialized_view_function(request):
     """
-    Enqueues a Cloud Task to refresh a materialized view asynchronously.
+    Enqueues a Cloud Task to asynchronously refresh a materialized view.
+    Ensures deduplication by generating a unique task name.
 
     Returns:
-        tuple: (response_message, status_code)
+        dict: Response message and status code.
     """
     try:
         logging.info("Starting materialized view refresh function.")
 
         # Generate deduplication key based on current timestamp
         timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M")
-        task_name = f"refresh-feed-search-{timestamp}"
+        task_name = f"refresh-materialized-view-{timestamp}"
 
         # Cloud Tasks client setup
         client = tasks_v2.CloudTasksClient()
@@ -43,7 +44,7 @@ def refresh_materialized_view_function(request: Request):
                 "http_method": tasks_v2.HttpMethod.POST,
                 "url": url,
                 "headers": {"Content-Type": "application/json"},
-                "body": jsonify(payload).data,
+                "body": json.dumps(payload).encode(),
             },
         }
 
@@ -60,19 +61,18 @@ def refresh_materialized_view_function(request: Request):
 
 
 @with_db_session
-@functions_framework.http
-def refresh_materialized_view_task(request: Request, db_session):
+def refresh_materialized_view_task(request, db_session):
     """
     Refreshes a materialized view using the CONCURRENTLY command to avoid
     table locks. This function is triggered by a Cloud Task.
 
     Returns:
-        tuple: (response_message, status_code)
+        dict: Response message and status code.
     """
     try:
         logging.info("Starting materialized view refresh task.")
 
-        data = request.get_json()
+        data = request.json()
         view_name = data.get("view_name")
         deduplication_key = data.get("deduplication_key")
 
@@ -111,7 +111,7 @@ def refresh_materialized_view(db_session, view_name):
         bool: True if refresh was successful, False otherwise.
     """
     try:
-        db_session.execute(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {view_name};")
+        db_session.execute(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {view_name}")
         db_session.commit()
         return True
     except Exception as error:

infra/functions-python/main.tf

@@ -1503,16 +1503,16 @@ resource "google_project_iam_member" "service_account_workflow_act_as_binding" {
 }
 
 # 15. functions/refresh_materialized_view
+# Cloud Function for refresh_materialized_view
 resource "google_cloudfunctions2_function" "refresh_materialized_view" {
-  name        = local.function_refresh_materialized_view_config.name
-  description = local.function_refresh_materialized_view_config.description
+  name        = "refresh-materialized-view-${var.environment}"
+  description = "Function to refresh materialized views asynchronously."
   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_refresh_materialized_view_config.entry_point
+    entry_point = "refresh_materialized_view_function"
     source {
       storage_source {
         bucket = google_storage_bucket.functions_bucket.name
@@ -1522,29 +1522,34 @@ resource "google_cloudfunctions2_function" "refresh_materialized_view" {
   }
 
   service_config {
-    available_memory                 = local.function_refresh_materialized_view_config.memory
-    timeout_seconds                  = local.function_refresh_materialized_view_config.timeout
-    available_cpu                    = local.function_refresh_materialized_view_config.available_cpu
-    max_instance_request_concurrency = local.function_refresh_materialized_view_config.max_instance_request_concurrency
-    max_instance_count               = local.function_refresh_materialized_view_config.max_instance_count
-    min_instance_count               = local.function_refresh_materialized_view_config.min_instance_count
+    environment_variables = {
+      PROJECT_ID  = var.project_id
+      QUEUE_NAME  = google_cloud_tasks_queue.refresh_materialized_view_queue.name
+      LOCATION    = var.gcp_region
+      FUNCTION_URL = "https://${google_cloudfunctions2_function.refresh_materialized_view.name}.cloudfunctions.net"
+    }
+    available_memory                 = "512Mi"
+    timeout_seconds                  = 60
+    available_cpu                    = 1
+    max_instance_request_concurrency = 10
+    max_instance_count               = 5
+    min_instance_count               = 0
     service_account_email            = google_service_account.functions_service_account.email
-    ingress_settings                 = local.function_refresh_materialized_view_config.ingress_settings
+    ingress_settings                 = "ALLOW_ALL"
     vpc_connector                    = data.google_vpc_access_connector.vpc_connector.id
     vpc_connector_egress_settings    = "PRIVATE_RANGES_ONLY"
-
-    dynamic "secret_environment_variables" {
-      for_each = local.function_refresh_materialized_view_config.secret_environment_variables
-      content {
-        key        = secret_environment_variables.value["key"]
-        project_id = var.project_id
-        secret     = "${upper(var.environment)}_${secret_environment_variables.value["key"]}"
-        version    = "latest"
-      }
-    }
   }
 }
 
+# IAM entry for invoking the refresh_materialized_view function
+resource "google_cloudfunctions2_function_iam_member" "refresh_materialized_view_invoker" {
+  project        = var.project_id
+  location       = var.gcp_region
+  cloud_function = google_cloudfunctions2_function.refresh_materialized_view.name
+  role           = "roles/cloudfunctions.invoker"
+  member         = "serviceAccount:${google_service_account.functions_service_account.email}"
+}
+
 # Task queue to invoke refresh_materialized_view function
 resource "google_cloud_tasks_queue" "refresh_materialized_view_queue" {
   name     = "refresh-materialized-view-queue-${var.environment}"