ops_utils.google_sheets_util

  1"""Module to interact with Google Sheets API."""
  2from typing import Optional
  3from google.auth import default
  4from google.auth.transport.requests import Request
  5import gspread
  6
  7
  8class GoogleSheets:
  9    """Class to interact with Google Sheets API."""
 10
 11    _SCOPES = ['https://www.googleapis.com/auth/spreadsheets']
 12
 13    def __init__(self, service_account_info: Optional[dict] = None):
 14        """
 15        Initialize the GoogleSheets instance using the service account or user credentials.
 16
 17        This method sets up the Google Sheets client using either the provided service account
 18        credentials or the application-default credentials. If no service account information
 19        is provided, ensure that the application-default credentials are properly configured.
 20
 21        **Args:**
 22        - service_account_info (Optional[dict]): A dictionary containing the service account credentials.
 23
 24        **Example:**
 25        To use application-default credentials, run the following command:
 26        ```
 27        gcloud auth application-default login \
 28        --scopes=https://www.googleapis.com/auth/spreadsheets,https://www.googleapis.com/auth/cloud-platform
 29        ```
 30        """
 31        if service_account_info:
 32            self.gc = gspread.service_account_from_dict(service_account_info)
 33        else:
 34            # This assumes gcloud auth application-default login has been run
 35            creds, _ = default(scopes=self._SCOPES)
 36            creds.refresh(Request())
 37            self.gc = gspread.Client(auth=creds)
 38
 39    def _open_worksheet(self, spreadsheet_id: str, worksheet_name: str) -> gspread.Worksheet:
 40        """
 41        Open a spreadsheet by its ID.
 42
 43        **Args:**
 44        - spreadsheet_id (str): The ID of the Google Sheet.
 45        """
 46        spreadsheet = self.gc.open_by_key(spreadsheet_id)
 47        return spreadsheet.worksheet(worksheet_name)
 48
 49    def update_cell(self, spreadsheet_id: str, worksheet_name: str, cell: str, value: str) -> None:
 50        """
 51        Update a specific cell in the sheet.
 52
 53        **Args:**
 54        - spreadsheet_id (str): Spreadsheet ID.
 55        - worksheet_name (str): Sheet/tab name.
 56        - cell (str): A1-style cell notation.
 57        - value (str): Value to insert.
 58        """
 59        worksheet = self._open_worksheet(spreadsheet_id, worksheet_name)
 60        worksheet.update([[value]], range_name=cell)
 61
 62    def get_cell_value(self, spreadsheet_id: str, worksheet_name: str, cell: str) -> str:
 63        """
 64        Get the value of a specific cell.
 65
 66        **Args:**
 67        - spreadsheet_id (str): Spreadsheet ID.
 68        - worksheet_name (str): Sheet/tab name.
 69        - cell (str): A1-style cell reference.
 70
 71        **Returns:**
 72        - str or None: Cell value or None if empty.
 73        """
 74        ws = self._open_worksheet(spreadsheet_id, worksheet_name)
 75        return ws.acell(cell).value
 76
 77    def get_last_row(self, spreadsheet_id: str, worksheet_name: str) -> int:
 78        """
 79        Get the last non-empty row in the specified column, accounting for trailing empty rows.
 80
 81        **Args:**
 82        - spreadsheet_id (str): Spreadsheet ID.
 83        - worksheet_name (str): Sheet/tab name.
 84
 85        **Returns:**
 86        - int: The last non-empty row number.
 87        """
 88        ws = self._open_worksheet(spreadsheet_id, worksheet_name)
 89        col_values = ws.col_values(1)  # Get all values in the first column
 90        for row_index in range(len(col_values), 0, -1):  # Iterate from the last row to the first
 91            if col_values[row_index - 1]:  # Check if the cell is not empty
 92                return row_index
 93        return 0  # Return 0 if all rows are empty
 94
 95    def get_column_values(self, spreadsheet_id: str, worksheet_name: str, column: str) -> list:
 96        """
 97        Get all values in a specific column in order of row.
 98
 99        **Args:**
100        - spreadsheet_id (str): Spreadsheet ID.
101        - worksheet_name (str): Sheet/tab name.
102        - column (str): Column identifier (e.g., "A" or "1").
103
104        **Returns:**
105        - list: List of values in the column.
106        """
107        ws = self._open_worksheet(spreadsheet_id, worksheet_name)
108
109        # Convert column letter to number if it's a letter
110        if column.isalpha():
111            # gspread uses 1-based indexing for columns
112            column_index = 0
113            for char in column.upper():
114                column_index = column_index * 26 + (ord(char) - ord('A') + 1)
115        else:
116            # If column is already a number
117            column_index = int(column)
118
119        return ws.col_values(column_index)
120
121    def get_worksheet_as_dict(self, spreadsheet_id: str, worksheet_name: str) -> list[dict]:
122        """
123        Get all data from a worksheet as a list of dictionaries.
124
125        The first row is used as the header/keys for the dictionaries.
126        Each subsequent row becomes a dictionary with the header values as keys.
127
128        **Args:**
129        - spreadsheet_id (str): Spreadsheet ID.
130        - worksheet_name (str): Sheet/tab name.
131
132        **Returns:**
133        - list[dict]: List of dictionaries, where each dictionary represents a row
134                      with column headers as keys.
135        """
136        ws = self._open_worksheet(spreadsheet_id, worksheet_name)
137        return ws.get_all_records()
138
139    def batch_update_cells(self, spreadsheet_id: str, worksheet_name: str, updates: list[dict[str, str]]) -> None:
140        """
141        Update multiple cells in a single batch request.
142
143        This method is more efficient than calling update_cell multiple times
144        as it sends all updates in a single API request.
145
146        **Args:**
147        - spreadsheet_id (str): Spreadsheet ID.
148        - worksheet_name (str): Sheet/tab name.
149        - updates (list[dict[str, str]]): List of dictionaries, each containing:
150            - 'cell': A1-style cell notation (e.g., 'A1', 'B5')
151            - 'value': Value to insert
152
153        **Example:**
154        ```python
155        updates = [
156            {'cell': 'A1', 'value': 'Name'},
157            {'cell': 'B1', 'value': 'Age'},
158            {'cell': 'A2', 'value': 'John'},
159            {'cell': 'B2', 'value': '30'}
160        ]
161        gs.batch_update_cells(spreadsheet_id, worksheet_name, updates)
162        ```
163        """
164        worksheet = self._open_worksheet(spreadsheet_id, worksheet_name)
165
166        # Build the batch update data
167        batch_data = []
168        for update in updates:
169            batch_data.append({
170                'range': update['cell'],
171                'values': [[update['value']]]
172            })
173
174        # Perform batch update
175        worksheet.batch_update(batch_data)