ops_utils.gcp_cloud_functions

Module to interact with Google Cloud Functions API.

 1"""Module to interact with Google Cloud Functions API."""
 2from google.auth import default
 3from googleapiclient.discovery import build
 4import logging
 5import json
 6from typing import Optional
 7
 8ERROR_KEY = "error"
 9
10class GCPCloudFunctionCaller:
11    """Class to call a GCP Cloud Function programmatically."""
12
13    def __init__(self, project: Optional[str] = None) -> None:
14        """
15        Initialize the GCPCloudFunctionCaller class.
16
17        **Args:**
18        - project (str, optional): The GCP project ID. Defaults to the active project.
19        """
20        credentials, default_project = default()
21        self.project = project or default_project
22        self.service = build("cloudfunctions", "v1", credentials=credentials)
23
24    def call_function(
25            self,
26            function_name: str,
27            data: dict,
28            check_error: bool = True,
29    ) -> dict:
30        """
31        Call a GCP Cloud Function.
32
33        **Args:**
34        - function_name (str): The name of the Cloud Function to call.
35        - data (dict): The payload to send to the Cloud Function.
36        - check_error (bool): Whether to check for errors in the response. Defaults to True.
37
38        **Returns:**
39        - dict: The response from the Cloud Function.
40        """
41        function_path = f"projects/{self.project}/locations/us-central1/functions/{function_name}"
42        request = self.service.projects().locations().functions().call(
43            name=function_path, body={"data": json.dumps(data)}
44        )
45
46        response = request.execute()
47        if check_error and ERROR_KEY in response:
48            raise RuntimeError(f"Error calling Cloud Function {function_name}: {response[ERROR_KEY]}")
49        logging.info(f"Cloud Function {function_name} called successfully.")
50        return response
ERROR_KEY = 'error'
class GCPCloudFunctionCaller: 
11class GCPCloudFunctionCaller:
12    """Class to call a GCP Cloud Function programmatically."""
13
14    def __init__(self, project: Optional[str] = None) -> None:
15        """
16        Initialize the GCPCloudFunctionCaller class.
17
18        **Args:**
19        - project (str, optional): The GCP project ID. Defaults to the active project.
20        """
21        credentials, default_project = default()
22        self.project = project or default_project
23        self.service = build("cloudfunctions", "v1", credentials=credentials)
24
25    def call_function(
26            self,
27            function_name: str,
28            data: dict,
29            check_error: bool = True,
30    ) -> dict:
31        """
32        Call a GCP Cloud Function.
33
34        **Args:**
35        - function_name (str): The name of the Cloud Function to call.
36        - data (dict): The payload to send to the Cloud Function.
37        - check_error (bool): Whether to check for errors in the response. Defaults to True.
38
39        **Returns:**
40        - dict: The response from the Cloud Function.
41        """
42        function_path = f"projects/{self.project}/locations/us-central1/functions/{function_name}"
43        request = self.service.projects().locations().functions().call(
44            name=function_path, body={"data": json.dumps(data)}
45        )
46
47        response = request.execute()
48        if check_error and ERROR_KEY in response:
49            raise RuntimeError(f"Error calling Cloud Function {function_name}: {response[ERROR_KEY]}")
50        logging.info(f"Cloud Function {function_name} called successfully.")
51        return response

Class to call a GCP Cloud Function programmatically.

GCPCloudFunctionCaller(project: Optional[str] = None) 
14    def __init__(self, project: Optional[str] = None) -> None:
15        """
16        Initialize the GCPCloudFunctionCaller class.
17
18        **Args:**
19        - project (str, optional): The GCP project ID. Defaults to the active project.
20        """
21        credentials, default_project = default()
22        self.project = project or default_project
23        self.service = build("cloudfunctions", "v1", credentials=credentials)

Initialize the GCPCloudFunctionCaller class.

Args:

  • project (str, optional): The GCP project ID. Defaults to the active project.
project
service
def call_function(self, function_name: str, data: dict, check_error: bool = True) -> dict: 
25    def call_function(
26            self,
27            function_name: str,
28            data: dict,
29            check_error: bool = True,
30    ) -> dict:
31        """
32        Call a GCP Cloud Function.
33
34        **Args:**
35        - function_name (str): The name of the Cloud Function to call.
36        - data (dict): The payload to send to the Cloud Function.
37        - check_error (bool): Whether to check for errors in the response. Defaults to True.
38
39        **Returns:**
40        - dict: The response from the Cloud Function.
41        """
42        function_path = f"projects/{self.project}/locations/us-central1/functions/{function_name}"
43        request = self.service.projects().locations().functions().call(
44            name=function_path, body={"data": json.dumps(data)}
45        )
46
47        response = request.execute()
48        if check_error and ERROR_KEY in response:
49            raise RuntimeError(f"Error calling Cloud Function {function_name}: {response[ERROR_KEY]}")
50        logging.info(f"Cloud Function {function_name} called successfully.")
51        return response

Call a GCP Cloud Function.

Args:

  • function_name (str): The name of the Cloud Function to call.
  • data (dict): The payload to send to the Cloud Function.
  • check_error (bool): Whether to check for errors in the response. Defaults to True.

Returns:

  • dict: The response from the Cloud Function.