feat: google sheets api integration

app/api/server.py

@@ -149,6 +149,8 @@ async def lifespan(app: FastAPI):
     await close_gcs_service()
     from app.api.v1.gmail import close_gmail_service
     await close_gmail_service()
+    from app.api.v1.sheets import close_sheets_service
+    await close_sheets_service()
     from app.services.media_storage_service import close_storage_service
     await close_storage_service()
     from app.utils.http_utils import close_shared_aiohttp_sessions

app/api/v1/router.py

@@ -25,6 +25,7 @@ from app.api.v1 import (
     scopes,
     scraper,
     semantic_router,
+    sheets,
     sql_validator,
     system,
     token_counter,
@@ -62,6 +63,7 @@ api_v1_router.include_router(google_oauth.router, tags=["Google OAuth"])
 api_v1_router.include_router(scopes.router, tags=["Google Scopes"])
 api_v1_router.include_router(gcs.router, tags=["Google Cloud Storage"])
 api_v1_router.include_router(gmail.router, tags=["Gmail"])
+api_v1_router.include_router(sheets.router, tags=["Google Sheets"])
 api_v1_router.include_router(media_convert.router, tags=["Media-to-Media Conversion"])
 api_v1_router.include_router(json_extract.router, tags=["JSON Extractor"])
 api_v1_router.include_router(keys_extract.router, prefix="/json", tags=["Keys Extractor"])

app/api/v1/sheets.py

@@ -0,0 +1,364 @@
+from __future__ import annotations
+
+import time
+from typing import Any, Dict, Optional
+
+from fastapi import APIRouter, Depends, Header, HTTPException
+
+from app.config import get_settings
+from app.core.logger import get_logger
+from app.models.schemas import (
+    SheetsAppendRequest,
+    SheetsBatchUpdateRequest,
+    SheetsCreateSpreadsheetRequest,
+    SheetsGenericResponse,
+    SheetsRefreshResponse,
+    SheetsValueUpdateRequest,
+)
+from app.services.sheets_service import (
+    SheetsAPIError,
+    SheetsCredentials,
+    SheetsService,
+)
+
+router = APIRouter(prefix="/google/sheets", tags=["Google Sheets"])
+_logger = get_logger(__name__)
+_settings = get_settings()
+
+_sheets_service = SheetsService()
+
+
+def get_sheets_service() -> SheetsService:
+    return _sheets_service
+
+
+async def close_sheets_service() -> None:
+    await _sheets_service.close()
+
+
+def _credentials(
+    x_access_token: Optional[str] = Header(None, alias="X-Access-Token"),
+    x_refresh_token: Optional[str] = Header(None, alias="X-Refresh-Token"),
+    x_client_id: Optional[str] = Header(None, alias="X-Client-Id"),
+    x_client_secret: Optional[str] = Header(None, alias="X-Client-Secret"),
+    x_token_expires_at: Optional[float] = Header(None, alias="X-Token-Expires-At"),
+) -> SheetsCredentials:
+    if not x_access_token:
+        raise HTTPException(
+            status_code=401,
+            detail="X-Access-Token header is required.",
+        )
+    return SheetsCredentials(
+        access_token=x_access_token,
+        refresh_token=x_refresh_token,
+        client_id=x_client_id,
+        client_secret=x_client_secret,
+        expires_at=x_token_expires_at,
+    )
+
+
+def _http_error(exc: SheetsAPIError) -> HTTPException:
+    return HTTPException(status_code=exc.status_code, detail=exc.message)
+
+
+def _elapsed_ms(start: float) -> float:
+    return round((time.perf_counter() - start) * 1000, 2)
+
+
+def _ok(
+    start: float,
+    creds: SheetsCredentials,
+    data: Any = None,
+    *,
+    error: Optional[str] = None,
+) -> SheetsGenericResponse:
+    return SheetsGenericResponse(
+        success=error is None,
+        time_ms=_elapsed_ms(start),
+        data=data,
+        refreshed_access_token=creds.refreshed_access_token,
+        error=error,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Service metadata & token lifecycle
+# ---------------------------------------------------------------------------
+
+@router.get("/scopes", response_model=SheetsGenericResponse,
+            summary="List all supported Google Sheets OAuth scopes")
+async def list_scopes(
+    service: SheetsService = Depends(get_sheets_service),
+):
+    start = time.perf_counter()
+    scopes = await service.list_available_scopes()
+    _logger.info("Listed %d Sheets scopes (%.2fms)", len(scopes), _elapsed_ms(start))
+    return SheetsGenericResponse(success=True, time_ms=_elapsed_ms(start), data=scopes)
+
+
+@router.post("/token/refresh", response_model=SheetsRefreshResponse,
+             summary="Refresh an access token (stateless, returned to the client)")
+async def token_refresh(
+    body: Dict[str, str],
+    service: SheetsService = Depends(get_sheets_service),
+):
+    start = time.perf_counter()
+    client_id = body.get("client_id", "")
+    client_secret = body.get("client_secret", "")
+    refresh_token = body.get("refresh_token", "")
+    if not (client_id and client_secret and refresh_token):
+        raise HTTPException(
+            status_code=400,
+            detail="client_id, client_secret and refresh_token are required.",
+        )
+    try:
+        tokens = await service.refresh_access_token(
+            client_id=client_id,
+            client_secret=client_secret,
+            refresh_token=refresh_token,
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets token refreshed (%.2fms)", _elapsed_ms(start))
+    return SheetsRefreshResponse(
+        success=True,
+        access_token=tokens.get("access_token"),
+        expires_in=int(tokens.get("expires_in", 0)),
+        token_type=tokens.get("token_type", "Bearer"),
+        scope=tokens.get("scope"),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Spreadsheets
+# ---------------------------------------------------------------------------
+
+@router.post("/spreadsheets", response_model=SheetsGenericResponse,
+             summary="Create a new spreadsheet")
+async def create_spreadsheet(
+    body: SheetsCreateSpreadsheetRequest,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.create_spreadsheet(
+            creds,
+            title=body.title,
+            sheets=body.sheets,
+            properties=body.properties,
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets spreadsheet '%s' created (%.2fms)", body.title, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/spreadsheets/{spreadsheet_id}", response_model=SheetsGenericResponse,
+            summary="Get spreadsheet metadata (spreadsheets.get)")
+async def get_spreadsheet(
+    spreadsheet_id: str,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+    ranges: Optional[str] = None,
+    include_grid_data: bool = False,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_spreadsheet(
+            creds,
+            spreadsheet_id,
+            ranges=ranges.split(",") if ranges else None,
+            include_grid_data=include_grid_data,
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets spreadsheet %s fetched (%.2fms)", spreadsheet_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/spreadsheets/{spreadsheet_id}/sheets/{sheet_id}",
+            response_model=SheetsGenericResponse,
+            summary="Get a single sheet's metadata (spreadsheets.sheets.get)")
+async def get_sheet(
+    spreadsheet_id: str,
+    sheet_id: int,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+    ranges: Optional[str] = None,
+    include_grid_data: bool = False,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_sheet(
+            creds,
+            spreadsheet_id,
+            sheet_id,
+            ranges=ranges.split(",") if ranges else None,
+            include_grid_data=include_grid_data,
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info(
+        "Sheets sheet %d of %s fetched (%.2fms)", sheet_id, spreadsheet_id, _elapsed_ms(start)
+    )
+    return _ok(start, creds, data)
+
+
+@router.post("/spreadsheets/{spreadsheet_id}:batchUpdate",
+             response_model=SheetsGenericResponse,
+             summary="Apply batch mutations (addSheet, updateCells, etc.)")
+async def batch_update(
+    spreadsheet_id: str,
+    body: SheetsBatchUpdateRequest,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.batch_update(
+            creds, spreadsheet_id, requests=body.requests
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets spreadsheet %s batch-updated (%.2fms)", spreadsheet_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+# ---------------------------------------------------------------------------
+# Values
+# ---------------------------------------------------------------------------
+
+@router.get("/spreadsheets/{spreadsheet_id}/values/{range}",
+            response_model=SheetsGenericResponse,
+            summary="Read cell values from a range (values.get)")
+async def get_values(
+    spreadsheet_id: str,
+    range: str,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+    major_dimension: Optional[str] = None,
+    value_render_option: Optional[str] = None,
+    date_time_render_option: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_values(
+            creds,
+            spreadsheet_id,
+            range,
+            major_dimension=major_dimension,
+            value_render_option=value_render_option,
+            date_time_render_option=date_time_render_option,
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets values %s!%s fetched (%.2fms)", spreadsheet_id, range, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/spreadsheets/{spreadsheet_id}/values:batchGet",
+            response_model=SheetsGenericResponse,
+            summary="Read multiple ranges in one call (values.batchGet)")
+async def batch_get_values(
+    spreadsheet_id: str,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+    ranges: Optional[str] = None,
+    major_dimension: Optional[str] = None,
+    value_render_option: Optional[str] = None,
+    date_time_render_option: Optional[str] = None,
+):
+    start = time.perf_counter()
+    if not ranges:
+        raise HTTPException(status_code=400, detail="At least one 'ranges' value is required.")
+    try:
+        data = await service.batch_get_values(
+            creds,
+            spreadsheet_id,
+            ranges=ranges.split(","),
+            major_dimension=major_dimension,
+            value_render_option=value_render_option,
+            date_time_render_option=date_time_render_option,
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets batchGet %s fetched (%.2fms)", spreadsheet_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.put("/spreadsheets/{spreadsheet_id}/values/{range}",
+            response_model=SheetsGenericResponse,
+            summary="Overwrite cell values in a range (values.update)")
+async def update_values(
+    spreadsheet_id: str,
+    range: str,
+    body: SheetsValueUpdateRequest,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+    value_input_option: str = "RAW",
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_values(
+            creds,
+            spreadsheet_id,
+            range,
+            values=body.values,
+            major_dimension=body.major_dimension,
+            value_input_option=value_input_option,
+            include_values_in_response=body.include_values_in_response,
+            value_render_option=body.value_render_option,
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets values %s!%s updated (%.2fms)", spreadsheet_id, range, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/spreadsheets/{spreadsheet_id}/values/{range}:append",
+             response_model=SheetsGenericResponse,
+             summary="Append rows to a range (values.append)")
+async def append_values(
+    spreadsheet_id: str,
+    range: str,
+    body: SheetsAppendRequest,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+    value_input_option: str = "USER_ENTERED",
+    insert_data_option: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.append_values(
+            creds,
+            spreadsheet_id,
+            range,
+            values=body.values,
+            major_dimension=body.major_dimension,
+            value_input_option=value_input_option,
+            insert_data_option=insert_data_option,
+            include_values_in_response=False,
+        )
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets values %s!%s appended (%.2fms)", spreadsheet_id, range, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/spreadsheets/{spreadsheet_id}/values/{range}:clear",
+             response_model=SheetsGenericResponse,
+             summary="Clear cell values in a range (values.clear)")
+async def clear_values(
+    spreadsheet_id: str,
+    range: str,
+    creds: SheetsCredentials = Depends(_credentials),
+    service: SheetsService = Depends(get_sheets_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.clear_values(creds, spreadsheet_id, range)
+    except SheetsAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Sheets values %s!%s cleared (%.2fms)", spreadsheet_id, range, _elapsed_ms(start))
+    return _ok(start, creds, data)

app/config.py

@@ -141,6 +141,16 @@ class Settings(BaseSettings):
     gmail_default_scope: str = "https://www.googleapis.com/auth/gmail.modify"
     gmail_refresh_buffer_seconds: int = 60
 
+    # Google Sheets API settings
+    sheets_api_base_url: str = "https://sheets.googleapis.com/v4"
+    sheets_timeout: float = 30.0
+    sheets_max_retries: int = 3
+    sheets_base_backoff_seconds: float = 1.0
+    sheets_max_backoff_seconds: float = 32.0
+    sheets_retryable_statuses: str = "429,500,502,503,504"
+    sheets_default_scope: str = "https://www.googleapis.com/auth/spreadsheets"
+    sheets_refresh_buffer_seconds: int = 60
+
     # Media-to-Media conversion settings
     media_output_dir: str = "./data/media-convert"
     media_max_workers: int = 4

app/models/schemas.py

@@ -1996,3 +1996,79 @@ class GmailRefreshResponse(BaseModel):
     token_type: Optional[str] = None
     scope: Optional[str] = None
     error: Optional[str] = None
+
+
+# ---------------------------------------------------------------------------
+# Google Sheets API (stateless, client-supplied OAuth credentials)
+# ---------------------------------------------------------------------------
+
+class SheetsScope(Enum):
+    """All Sheets-specific OAuth scopes (developers.google.com/sheets/api/auth).
+
+    Drive scopes are intentionally excluded here — they belong to the separate
+    Google Drive API integration.
+    """
+
+    READONLY = "https://www.googleapis.com/auth/spreadsheets.readonly"
+    SPREADSHEETS = "https://www.googleapis.com/auth/spreadsheets"
+
+    @property
+    def permission_level(self) -> str:
+        return "sensitive"
+
+
+class SheetsCreateSpreadsheetRequest(BaseModel):
+    """Body for creating a spreadsheet (spreadsheets.create)."""
+
+    title: str = Field(..., min_length=1, max_length=255, description="Spreadsheet title")
+    sheets: Optional[List[Dict[str, Any]]] = Field(
+        None,
+        description="Initial sheet definitions, e.g. [{'properties': {'title': 'Data', 'index': 0}}]",
+    )
+    properties: Optional[Dict[str, Any]] = Field(
+        None,
+        description="Spreadsheet properties (locale, autoRecalc, defaultFormat, etc.)",
+    )
+
+
+class SheetsValueUpdateRequest(BaseModel):
+    """Body for overwriting a range of cells (values.update)."""
+
+    values: List[List[Any]] = Field(..., min_length=1, description="2D array of cell values (rows of columns)")
+    major_dimension: Optional[str] = Field(None, pattern="^(ROWS|COLUMNS)$", description="Major dimension of the values array")
+    include_values_in_response: Optional[bool] = Field(None, description="Whether the response should include updated values")
+    value_render_option: Optional[str] = Field(None, pattern="^(FORMATTED_VALUE|UNFORMATTED_VALUE|FORMULA)$", description="Render option for returned values")
+
+
+class SheetsAppendRequest(BaseModel):
+    """Body for appending rows to a range (values.append)."""
+
+    values: List[List[Any]] = Field(..., min_length=1, description="2D array of cell values to append")
+    major_dimension: Optional[str] = Field(None, pattern="^(ROWS|COLUMNS)$", description="Major dimension of the values array")
+
+
+class SheetsBatchUpdateRequest(BaseModel):
+    """Body for batch mutation requests (spreadsheets.batchUpdate)."""
+
+    requests: List[Dict[str, Any]] = Field(..., min_length=1, description="List of request objects (addSheet, updateCells, etc.)")
+
+
+class SheetsGenericResponse(BaseModel):
+    """Standard Google Sheets operation envelope."""
+
+    success: bool
+    time_ms: float = 0.0
+    data: Optional[Any] = Field(None, description="Raw Sheets API payload")
+    refreshed_access_token: Optional[str] = Field(None, description="New access token when an automatic refresh occurred; persist it client-side")
+    error: Optional[str] = Field(None, description="Human-readable error when success is false")
+
+
+class SheetsRefreshResponse(BaseModel):
+    """Stateless token refresh result returned to the client for its own storage."""
+
+    success: bool
+    access_token: Optional[str] = None
+    expires_in: Optional[int] = None
+    token_type: Optional[str] = None
+    scope: Optional[str] = None
+    error: Optional[str] = None

app/services/sheets_service.py

@@ -0,0 +1,563 @@
+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,
+    ) -> Any:
+        """Send a Sheets 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
+            )
+            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,
+    ) -> httpx.Response:
+        client = await self._get_client()
+        url = f"{_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)
+
+    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={},
+        )