from __future__ import annotations import asyncio import random import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional import httpx from app.config import get_settings from app.core.logger import get_logger from app.utils.http_utils import SharedAsyncClient _logger = get_logger(__name__) _settings = get_settings() _RETRYABLE_STATUS = frozenset({ int(code) for code in _settings.sheets_retryable_statuses.split(",") if code.strip().isdigit() }) class SheetsAPIError(Exception): """Raised for upstream Google Sheets API failures that map to a client-facing error.""" def __init__(self, message: str, status_code: int = 400, *, reason: str = "") -> None: super().__init__(message) self.message = message self.status_code = status_code self.reason = reason @dataclass class SheetsCredentials: """Client-supplied OAuth credentials (never persisted by this backend). ``refreshed_access_token`` is populated whenever the service transparently refreshes the access token, so the stateless client can persist it itself. """ access_token: str refresh_token: Optional[str] = None client_id: Optional[str] = None client_secret: Optional[str] = None expires_at: Optional[float] = None refreshed_access_token: Optional[str] = field(default=None, repr=False) @property def can_refresh(self) -> bool: return bool(self.refresh_token and self.client_id and self.client_secret) @property def needs_proactive_refresh(self) -> bool: if self.expires_at is None: return False return time.time() >= self.expires_at - _settings.sheets_refresh_buffer_seconds class SheetsService: """Stateless, fully-async abstraction over the Google Sheets REST API. Credentials are supplied by the caller on every invocation. Access tokens are transparently refreshed against Google's token endpoint when a ``refresh_token`` + ``client_id`` + ``client_secret`` are provided, and the resulting token is surfaced via ``SheetsCredentials.refreshed_access_token``. """ def __init__(self) -> None: self._http = SharedAsyncClient( timeout=httpx.Timeout(_settings.sheets_timeout), follow_redirects=False, ) async def _get_client(self) -> httpx.AsyncClient: return await self._http.get() async def close(self) -> None: await self._http.close() # ------------------------------------------------------------------ # Core request pipeline: auth, retry, backoff, error mapping # ------------------------------------------------------------------ async def _request( self, creds: SheetsCredentials, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Any] = None, base_url: Optional[str] = None, ) -> Any: """Send a Sheets/Drive API request with proactive + on-401 refresh and retries.""" if creds.needs_proactive_refresh: await self._refresh(creds, context=f"{method} {path}") refresh_done = False attempt = 0 while True: response = await self._send( creds, method, path, params=params, json_body=json_body, base_url=base_url ) if 200 <= response.status_code < 300: return self._decode(response) if response.status_code == 401 and creds.can_refresh and not refresh_done: _logger.warning( "Sheets 401 for %s %s; refreshing access token once", method, path, ) await self._refresh(creds, context=f"{method} {path}") refresh_done = True continue if response.status_code in _RETRYABLE_STATUS and attempt < _settings.sheets_max_retries: await self._backoff(response, attempt) attempt += 1 continue raise self._map_error(response, creds) async def _send( self, creds: SheetsCredentials, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Any] = None, base_url: Optional[str] = None, ) -> httpx.Response: client = await self._get_client() url = f"{base_url or _settings.sheets_api_base_url}{path}" headers = { "Authorization": f"Bearer {creds.access_token}", "Accept": "application/json", } try: return await client.request( method, url, params=params, json=json_body, headers=headers, ) except httpx.TimeoutException as exc: raise SheetsAPIError( f"Sheets API request timed out for {method} {path}.", status_code=504 ) from exc except httpx.RequestError as exc: raise SheetsAPIError( f"Failed to reach Sheets API: {exc}", status_code=502 ) from exc @staticmethod def _decode(response: httpx.Response) -> Any: if not response.content: return None try: return response.json() except ValueError: return response.text async def _backoff(self, response: httpx.Response, attempt: int) -> None: retry_after = response.headers.get("Retry-After") delay: Optional[float] = None if retry_after: try: delay = float(retry_after) except (TypeError, ValueError): delay = None if delay is None: base = _settings.sheets_base_backoff_seconds * (2 ** attempt) delay = min(base, _settings.sheets_max_backoff_seconds) * (0.5 + random.random()) delay = min(delay, _settings.sheets_max_backoff_seconds + 5.0) _logger.warning( "Sheets transient HTTP %s; retrying in %.2fs (attempt %d/%d)", response.status_code, delay, attempt + 1, _settings.sheets_max_retries, ) await asyncio.sleep(delay) def _map_error(self, response: httpx.Response, creds: SheetsCredentials) -> SheetsAPIError: status = response.status_code reason = "" detail = "" try: body = response.json() error = body.get("error") or {} if isinstance(error, dict): detail = error.get("message", "") errors = error.get("errors") or [] if errors and isinstance(errors[0], dict): reason = errors[0].get("reason", "") except Exception: body = {} detail = response.text[:500] _logger.error( "Sheets API error: HTTP %s reason=%s detail=%s", status, reason, detail ) if status == 401: if creds.can_refresh: return SheetsAPIError( "Access token is invalid or expired and could not be refreshed. " "Re-authorize via POST /api/v1/google/oauth/auth-url.", status_code=401, reason=reason, ) return SheetsAPIError( "Access token is invalid or expired. Refresh it via " "POST /api/v1/google/oauth/refresh (or /api/v1/google/sheets/token/refresh) " "and retry, or supply a fresh access token.", status_code=401, reason=reason, ) if status == 403 and reason == "PERMISSION_DENIED": return SheetsAPIError( "Permission denied. The access token may be missing the required " "Sheets scope (see /api/v1/google/sheets/scopes), or the spreadsheet " "is not shared with the authenticated account.", status_code=403, reason=reason, ) if status == 404: return SheetsAPIError( detail or "Spreadsheet or range not found.", status_code=404, reason=reason, ) if status == 429: return SheetsAPIError( "Sheets API rate limit exceeded. Please retry after a short delay.", status_code=429, reason=reason, ) if status in _RETRYABLE_STATUS: return SheetsAPIError( detail or f"Sheets API service error (HTTP {status}).", status_code=status, reason=reason, ) return SheetsAPIError( detail or f"Sheets API error (HTTP {status}).", status_code=status, reason=reason, ) # ------------------------------------------------------------------ # Token lifecycle # ------------------------------------------------------------------ async def refresh_access_token( self, *, client_id: str, client_secret: str, refresh_token: str, ) -> Dict[str, Any]: """Refresh an access token without touching any spreadsheet (standalone).""" tokens = await self._refresh( creds=None, client_id=client_id, client_secret=client_secret, refresh_token=refresh_token, context="standalone refresh", ) return tokens async def _refresh( self, creds: Optional[SheetsCredentials], *, context: str, client_id: Optional[str] = None, client_secret: Optional[str] = None, refresh_token: Optional[str] = None, ) -> Dict[str, Any]: if creds is not None: client_id = creds.client_id client_secret = creds.client_secret refresh_token = creds.refresh_token if not (client_id and client_secret and refresh_token): raise SheetsAPIError( "Token refresh requires refresh_token, client_id and client_secret.", status_code=400, ) client = await self._get_client() data = { "client_id": client_id, "client_secret": client_secret, "refresh_token": refresh_token, "grant_type": "refresh_token", } try: response = await client.post(_settings.google_oauth_token_url, data=data) except httpx.TimeoutException as exc: raise SheetsAPIError( "Google token refresh timed out.", status_code=504 ) from exc except httpx.RequestError as exc: raise SheetsAPIError( f"Failed to reach Google token endpoint: {exc}", status_code=502 ) from exc if response.status_code == 200: tokens = response.json() new_token = tokens.get("access_token") if not new_token: raise SheetsAPIError( "Google did not return an access token.", status_code=502 ) expires_in = int(tokens.get("expires_in", 3600)) if creds is not None: creds.access_token = new_token creds.expires_at = time.time() + expires_in creds.refreshed_access_token = new_token _logger.info("Sheets access token refreshed (%s)", context) return tokens try: body = response.json() error = body.get("error", "") description = body.get("error_description", "") except Exception: error = "" description = "" body = {} if error == "invalid_grant": raise SheetsAPIError( description or "Refresh token is invalid, expired, or revoked. " "Re-authorize via POST /api/v1/google/oauth/auth-url.", status_code=401, reason=error, ) if error == "invalid_client": raise SheetsAPIError( description or "Invalid client_id or client_secret.", status_code=401, reason=error, ) raise SheetsAPIError( description or f"Google token refresh failed (HTTP {response.status_code}).", status_code=response.status_code if response.status_code >= 400 else 502, reason=error, ) # ------------------------------------------------------------------ # Scopes # ------------------------------------------------------------------ async def list_available_scopes(self) -> List[Dict[str, Any]]: from app.models.schemas import SheetsScope return [ { "name": scope.name, "uri": scope.value, "permission_level": scope.permission_level, } for scope in SheetsScope ] # ------------------------------------------------------------------ # Spreadsheets # ------------------------------------------------------------------ async def create_spreadsheet( self, creds: SheetsCredentials, *, title: str, sheets: Optional[List[Dict[str, Any]]] = None, properties: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"properties": {"title": title}} if properties: body["properties"].update(properties) if sheets: body["sheets"] = sheets return await self._request(creds, "POST", "/spreadsheets", json_body=body) _SPREADSHEET_MIME_TYPE = "application/vnd.google-apps.spreadsheet" async def get_drive_file( self, creds: SheetsCredentials, file_id: str, ) -> Dict[str, Any]: """Read a Drive file's metadata (drive.files.get) — used to validate the target before deletion so arbitrary non-sheet files cannot be removed.""" return await self._request( creds, "GET", f"/files/{file_id}", params={"fields": "id,name,mimeType,trashed"}, base_url=_settings.drive_api_base_url, ) async def list_spreadsheets( self, creds: SheetsCredentials, *, page_size: int = 100, page_token: Optional[str] = None, include_trashed: bool = False, ) -> Dict[str, Any]: """List the user's Google Sheets via drive.files.list. Filters to ``application/vnd.google-apps.spreadsheet`` so only spreadsheets are returned (no arbitrary Drive files). The Sheets API has no list endpoint; Drive's ``files.list`` with a ``q`` filter is the documented way to enumerate spreadsheets. """ query = f"mimeType='{self._SPREADSHEET_MIME_TYPE}'" if not include_trashed: query += " and trashed=false" params: Dict[str, Any] = { "q": query, "pageSize": page_size, "fields": "nextPageToken,files(id,name,mimeType,trashed,modifiedTime)", } if page_token: params["pageToken"] = page_token return await self._request( creds, "GET", "/files", params=params, base_url=_settings.drive_api_base_url, ) async def delete_spreadsheet( self, creds: SheetsCredentials, spreadsheet_id: str, ) -> Dict[str, Any]: """Delete a spreadsheet via drive.files.delete. Pre-validates with drive.files.get that the target file is a Google Sheet (``mimeType == application/vnd.google-apps.spreadsheet``) so this endpoint can never remove an arbitrary Drive file. Requires the ``drive.file`` (or broader ``drive``) OAuth scope — the Sheets API has no delete method. """ file_meta = await self.get_drive_file(creds, spreadsheet_id) mime = (file_meta or {}).get("mimeType") if mime != self._SPREADSHEET_MIME_TYPE: raise SheetsAPIError( f"Refusing to delete '{spreadsheet_id}': it is not a Google Sheet " f"(mimeType={mime or 'unknown'}). Only spreadsheets can be removed " "through the spreadsheet delete API.", status_code=400, reason="INVALID_MIME_TYPE", ) await self._request( creds, "DELETE", f"/files/{spreadsheet_id}", base_url=_settings.drive_api_base_url, ) return { "spreadsheetId": spreadsheet_id, "mimeType": mime, "name": file_meta.get("name"), "deleted": True, } async def get_spreadsheet( self, creds: SheetsCredentials, spreadsheet_id: str, *, ranges: Optional[List[str]] = None, include_grid_data: bool = False, ) -> Dict[str, Any]: params: Dict[str, Any] = {} if ranges: params["ranges"] = ranges if include_grid_data: params["includeGridData"] = "true" return await self._request( creds, "GET", f"/spreadsheets/{spreadsheet_id}", params=params or None ) async def get_sheet( self, creds: SheetsCredentials, spreadsheet_id: str, sheet_id: int, *, ranges: Optional[List[str]] = None, include_grid_data: bool = False, ) -> Dict[str, Any]: params: Dict[str, Any] = {} if ranges: params["ranges"] = ranges if include_grid_data: params["includeGridData"] = "true" return await self._request( creds, "GET", f"/spreadsheets/{spreadsheet_id}/sheets/{sheet_id}", params=params or None, ) async def batch_update( self, creds: SheetsCredentials, spreadsheet_id: str, *, requests: List[Dict[str, Any]], response_include_grid_data: Optional[bool] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"requests": requests} if response_include_grid_data is not None: body["responseIncludeGridData"] = response_include_grid_data return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}:batchUpdate", json_body=body ) # ------------------------------------------------------------------ # Values # ------------------------------------------------------------------ async def get_values( self, creds: SheetsCredentials, spreadsheet_id: str, range_: str, *, major_dimension: Optional[str] = None, value_render_option: Optional[str] = None, date_time_render_option: Optional[str] = None, ) -> Dict[str, Any]: params: Dict[str, Any] = {} if major_dimension: params["majorDimension"] = major_dimension if value_render_option: params["valueRenderOption"] = value_render_option if date_time_render_option: params["dateTimeRenderOption"] = date_time_render_option return await self._request( creds, "GET", f"/spreadsheets/{spreadsheet_id}/values/{range_}", params=params or None, ) async def batch_get_values( self, creds: SheetsCredentials, spreadsheet_id: str, *, ranges: List[str], major_dimension: Optional[str] = None, value_render_option: Optional[str] = None, date_time_render_option: Optional[str] = None, ) -> Dict[str, Any]: params: Dict[str, Any] = {"ranges": ranges} if major_dimension: params["majorDimension"] = major_dimension if value_render_option: params["valueRenderOption"] = value_render_option if date_time_render_option: params["dateTimeRenderOption"] = date_time_render_option return await self._request( creds, "GET", f"/spreadsheets/{spreadsheet_id}/values:batchGet", params=params, ) async def update_values( self, creds: SheetsCredentials, spreadsheet_id: str, range_: str, *, values: List[List[Any]], major_dimension: Optional[str] = None, value_input_option: str = "RAW", include_values_in_response: Optional[bool] = None, value_render_option: Optional[str] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"values": values} if major_dimension: body["majorDimension"] = major_dimension params: Dict[str, Any] = {"valueInputOption": value_input_option} if include_values_in_response is not None: params["includeValuesInResponse"] = "true" if include_values_in_response else "false" if value_render_option: params["valueRenderOption"] = value_render_option return await self._request( creds, "PUT", f"/spreadsheets/{spreadsheet_id}/values/{range_}", params=params, json_body=body, ) async def append_values( self, creds: SheetsCredentials, spreadsheet_id: str, range_: str, *, values: List[List[Any]], major_dimension: Optional[str] = None, value_input_option: str = "USER_ENTERED", insert_data_option: Optional[str] = None, include_values_in_response: Optional[bool] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"values": values} if major_dimension: body["majorDimension"] = major_dimension params: Dict[str, Any] = {"valueInputOption": value_input_option} if insert_data_option: params["insertDataOption"] = insert_data_option if include_values_in_response is not None: params["includeValuesInResponse"] = "true" if include_values_in_response else "false" return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/values/{range_}:append", params=params, json_body=body, ) async def clear_values( self, creds: SheetsCredentials, spreadsheet_id: str, range_: str, ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/values/{range_}:clear", json_body={}, ) async def batch_update_values( self, creds: SheetsCredentials, spreadsheet_id: str, *, data: List[Dict[str, Any]], value_input_option: str = "RAW", include_values_in_response: Optional[bool] = None, value_render_option: Optional[str] = None, response_value_render_option: Optional[str] = None, response_date_time_render_option: Optional[str] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"data": data} if include_values_in_response is not None: body["includeValuesInResponse"] = include_values_in_response if value_render_option: body["valueRenderOption"] = value_render_option if response_value_render_option: body["responseValueRenderOption"] = response_value_render_option if response_date_time_render_option: body["responseDateTimeRenderOption"] = response_date_time_render_option return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/values:batchUpdate", params={"valueInputOption": value_input_option}, json_body=body, ) async def batch_clear_values( self, creds: SheetsCredentials, spreadsheet_id: str, *, ranges: List[str], ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/values:batchClear", json_body={"ranges": ranges}, ) async def get_spreadsheet_by_data_filter( self, creds: SheetsCredentials, spreadsheet_id: str, *, data_filters: List[Dict[str, Any]], include_grid_data: bool = False, ) -> Dict[str, Any]: body: Dict[str, Any] = {"dataFilters": data_filters} if include_grid_data: body["includeGridData"] = True return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}:getByDataFilter", json_body=body, ) async def copy_sheet( self, creds: SheetsCredentials, spreadsheet_id: str, sheet_id: int, *, destination_spreadsheet_id: str, ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/sheets/{sheet_id}:copyTo", json_body={"destinationSpreadsheetId": destination_spreadsheet_id}, ) async def batch_get_values_by_data_filter( self, creds: SheetsCredentials, spreadsheet_id: str, *, data_filters: List[Dict[str, Any]], major_dimension: Optional[str] = None, value_render_option: Optional[str] = None, date_time_render_option: Optional[str] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"dataFilters": data_filters} if major_dimension: body["majorDimension"] = major_dimension if value_render_option: body["valueRenderOption"] = value_render_option if date_time_render_option: body["dateTimeRenderOption"] = date_time_render_option return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/values:batchGetByDataFilter", json_body=body, ) async def batch_update_values_by_data_filter( self, creds: SheetsCredentials, spreadsheet_id: str, *, data: List[Dict[str, Any]], value_input_option: str = "RAW", include_values_in_response: Optional[bool] = None, value_render_option: Optional[str] = None, response_value_render_option: Optional[str] = None, response_date_time_render_option: Optional[str] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"data": data} if include_values_in_response is not None: body["includeValuesInResponse"] = include_values_in_response if value_render_option: body["valueRenderOption"] = value_render_option if response_value_render_option: body["responseValueRenderOption"] = response_value_render_option if response_date_time_render_option: body["responseDateTimeRenderOption"] = response_date_time_render_option return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/values:batchUpdateByDataFilter", params={"valueInputOption": value_input_option}, json_body=body, ) async def batch_clear_values_by_data_filter( self, creds: SheetsCredentials, spreadsheet_id: str, *, data_filters: List[Dict[str, Any]], ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/values:batchClearByDataFilter", json_body={"dataFilters": data_filters}, ) async def get_developer_metadata( self, creds: SheetsCredentials, spreadsheet_id: str, metadata_id: int, ) -> Dict[str, Any]: return await self._request( creds, "GET", f"/spreadsheets/{spreadsheet_id}/developerMetadata/{metadata_id}", ) async def search_developer_metadata( self, creds: SheetsCredentials, spreadsheet_id: str, *, data_filters: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {} if data_filters: body["dataFilters"] = data_filters return await self._request( creds, "POST", f"/spreadsheets/{spreadsheet_id}/developerMetadata:search", json_body=body, )