added refresh materialized view cloud function

Author: qcdyx

functions-python/refresh_materialized_view/.coveragerc

@@ -0,0 +1,76 @@
+# Refresh Materialized View Cloud Function
+
+This Google Cloud Function refreshes a materialized view using the `CONCURRENTLY` command to avoid table locks.
+
+## Purpose
+
+The function allows you to refresh PostgreSQL materialized views without blocking other database operations. It uses the `REFRESH MATERIALIZED VIEW CONCURRENTLY` command, which requires the materialized view to have a unique index.
+
+## Usage
+
+### HTTP Request
+
+The function accepts both GET and POST requests:
+
+#### GET Request
+
+```
+GET /refresh-materialized-view?view_name=your_view_name
+```
+
+#### POST Request
+
+```
+POST /refresh-materialized-view
+Content-Type: application/json
+
+{
+  "view_name": "your_view_name"
+}
+```
+
+### Parameters
+
+- `view_name` (required): The name of the materialized view to refresh. Can include schema prefix (e.g., `schema_name.view_name`).
+
+### Response
+
+#### Success Response
+
+```json
+{
+  "message": "Successfully refreshed materialized view: your_view_name"
+}
+```
+
+#### Error Response
+
+```json
+{
+  "error": "Error refreshing materialized view"
+}
+```
+
+## Security
+
+The function validates the view name to prevent SQL injection attacks. Only alphanumeric characters, underscores, and dots are allowed in view names.
+
+## Requirements
+
+- The materialized view must have a unique index for the `CONCURRENTLY` option to work
+- The database user must have appropriate permissions to refresh materialized views
+
+## Environment Variables
+
+- `FEEDS_DATABASE_URL`: PostgreSQL database connection string
+
+## Error Handling
+
+The function handles various error scenarios:
+
+- Missing view name parameter
+- Invalid view name format
+- Database connection issues
+- View refresh failures
+
+All errors are logged and returned with appropriate HTTP status codes.

functions-python/refresh_materialized_view/README.md

@@ -0,0 +1,20 @@
+{
+  "name": "refresh-materialized-view",
+  "description": "Refreshes a materialized view using the CONCURRENTLY command to avoid table locks",
+  "entry_point": "refresh_materialized_view_function",
+  "timeout": 1800,
+  "memory": "512Mi",
+  "trigger_http": true,
+  "include_folders": ["helpers"],
+  "include_api_folders": ["database_gen", "database", "common"],
+  "secret_environment_variables": [
+    {
+      "key": "FEEDS_DATABASE_URL"
+    }
+  ],
+  "ingress_settings": "ALLOW_INTERNAL_AND_GCLB",
+  "max_instance_request_concurrency": 1,
+  "max_instance_count": 3,
+  "min_instance_count": 0,
+  "available_cpu": 1
+}

functions-python/refresh_materialized_view/function_config.json

@@ -0,0 +1,21 @@
+# 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
+
+# Flask for HTTP request handling
+Flask==3.0.0
+
+# Configuration
+python-dotenv~=1.0.0

functions-python/refresh_materialized_view/requirements.txt

@@ -0,0 +1,8 @@
+# Development packages
+pytest~=7.4.0
+pytest-cov~=4.1.0
+coverage~=7.2.0
+black~=23.0.0
+flake8~=6.0.0
+pre-commit~=3.3.0
+requests-mock~=1.11.0

functions-python/refresh_materialized_view/requirements_dev.txt

@@ -0,0 +1,41 @@
+import logging
+import functions_framework
+from flask import Request
+from shared.helpers.logger import init_logger
+from shared.database.database import with_db_session, refresh_materialized_view
+
+init_logger()
+
+
+@with_db_session
+@functions_framework.http
+def refresh_materialized_view_function(request: Request, db_session):
+    """
+    Refreshes a materialized view using the CONCURRENTLY command to avoid
+    table locks.
+
+    Returns:
+        tuple: (response_message, status_code)
+    """
+    try:
+        logging.info("Starting materialized view refresh function.")
+
+        view_name = "my_materialized_view"
+        logging.info(f"Refreshing materialized view: {view_name}")
+
+        # Call the refresh function
+        success = refresh_materialized_view(db_session, view_name)
+
+        if success:
+            success_msg = f"Successfully refreshed materialized view: {view_name}"
+            logging.info(success_msg)
+            return {"message": success_msg}, 200
+        else:
+            error_msg = f"Failed to refresh materialized view: {view_name}"
+            logging.error(error_msg)
+            return {"error": error_msg}, 500
+
+    except Exception as error:
+        error_msg = f"Error refreshing materialized view: {error}"
+        logging.error(error_msg)
+        return {"error": error_msg}, 500

functions-python/refresh_materialized_view/src/main.py

@@ -0,0 +1,77 @@
+import pytest
+from unittest.mock import Mock, patch
+import sys
+import os
+
+# Add the src directory to the path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
+
+from main import refresh_materialized_view_function  # noqa: E402
+
+
+class TestRefreshMaterializedView:
+    """Test class for the refresh materialized view function."""
+
+    @patch("main.refresh_materialized_view")
+    def test_refresh_view_success(self, mock_refresh):
+        """Test successful view refresh."""
+        mock_refresh.return_value = True
+
+        # Mock request object
+        request = Mock()
+        request.method = "GET"
+
+        # Mock database session
+        db_session = Mock()
+
+        # Call the function
+        response, status_code = refresh_materialized_view_function(request, db_session)
+
+        # Assertions
+        assert "Successfully refreshed materialized view" in response["message"]
+        mock_refresh.assert_called_once_with(db_session, "my_materialized_view")
+
+    @patch("main.refresh_materialized_view")
+    def test_refresh_view_failure(self, mock_refresh):
+        """Test failure to refresh the view."""
+        mock_refresh.return_value = False
+
+        # Mock request object
+        request = Mock()
+        request.method = "GET"
+
+        # Mock database session
+        db_session = Mock()
+
+        # Call the function
+        response, status_code = refresh_materialized_view_function(request, db_session)
+
+        # Assertions
+        assert status_code == 500
+        assert "Failed to refresh materialized view" in response["error"]
+        mock_refresh.assert_called_once_with(db_session, "my_materialized_view")
+
+    @patch("main.refresh_materialized_view")
+    def test_refresh_view_exception(self, mock_refresh):
+        """Test exception during view refresh."""
+        mock_refresh.side_effect = Exception("Database connection failed")
+
+        # Mock request object
+        request = Mock()
+        request.method = "GET"
+
+        # Mock database session
+        db_session = Mock()
+
+        # Call the function
+        response, status_code = refresh_materialized_view_function(request, db_session)
+
+        # Assertions
+        assert status_code == 500
+        assert "Error refreshing materialized view" in response["error"]
+        assert "Database connection failed" in response["error"]
+        mock_refresh.assert_called_once_with(db_session, "my_materialized_view")
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])

functions-python/refresh_materialized_view/tests/test_main.py

@@ -68,6 +68,9 @@ locals {
 
   function_tasks_executor_config = jsondecode(file("${path.module}/../../functions-python/tasks_executor/function_config.json"))
   function_tasks_executor_zip = "${path.module}/../../functions-python/tasks_executor/.dist/tasks_executor.zip"
+
+  function_refresh_materialized_view_config = jsondecode(file("${path.module}/../../functions-python/refresh_materialized_view/function_config.json"))
+  function_refresh_materialized_view_zip = "${path.module}/../../functions-python/refresh_materialized_view/.dist/refresh_materialized_view.zip"
 }
 
 locals {
@@ -81,7 +84,8 @@ locals {
     [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],
-    [for x in local.function_tasks_executor_config.secret_environment_variables : x.key]
+    [for x in local.function_tasks_executor_config.secret_environment_variables : x.key],
+    [for x in local.function_refresh_materialized_view_config.secret_environment_variables : x.key]
   )
 
   # Convert the list to a set to ensure uniqueness
@@ -227,6 +231,13 @@ resource "google_storage_bucket_object" "tasks_executor_zip" {
   source = local.function_tasks_executor_zip
 }
 
+# 15. Refresh Materialized View
+resource "google_storage_bucket_object" "refresh_materialized_view_zip" {
+  bucket = google_storage_bucket.functions_bucket.name
+  name   = "refresh-materialized-view-${substr(filebase64sha256(local.function_refresh_materialized_view_zip), 0, 10)}.zip"
+  source = local.function_refresh_materialized_view_zip
+}
+
 # Secrets access
 resource "google_secret_manager_secret_iam_member" "secret_iam_member" {
   for_each = local.unique_secret_keys
@@ -1469,6 +1480,14 @@ resource "google_cloudfunctions2_function_iam_member" "update_feed_status_invoke
   member         = "serviceAccount:${google_service_account.functions_service_account.email}"
 }
 
+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}"
+}
+
 # Grant permissions to the service account to create bigquery jobs
 resource "google_project_iam_member" "bigquery_job_user" {
   project = var.project_id