Blog

Google Sheets as a REST API with FastAPI and Python

Use SheetsAPI to treat any Google Sheet as a typed REST endpoint - async reads, appends, dependency injection, and cache invalidation with FastAPI.

Jun 26, 20265 min read

#python #fastapi #google-sheets #httpx #pydantic

Google Sheets is already where a lot of operational data lives - event registrations, product catalogs, content queues, support tickets logged by a Zapier workflow. The friction comes when you want to expose that data through a proper API. SheetsAPI removes the OAuth dance and Apps Script glue by giving you a plain HTTP interface to any sheet: GET rows, POST new ones, filter by column value. This post walks through wiring it into a FastAPI service with async I/O, Pydantic response models, dependency injection, and lightweight caching.

The SheetsAPI contract

Every request targets a single URL pattern:

https://sheetsapi.io/api/spreadsheets/{userKey}/{sheetName}

GET returns rows. Supports limit, offset, and filter[col]=val query params.
POST appends a new row. The body is a JSON object where keys match column headers.
Auth is a bearer token passed as Authorization: Bearer sk_....

A successful GET response looks like this:

{
  "data": [{ "id": "1", "name": "Alice", "status": "active" }],
  "meta": { "total": 142, "limit": 20, "offset": 0 }
}

Project setup

pip install fastapi uvicorn httpx pydantic

Pydantic models

Define the response shapes first. This keeps validation at the boundary and gives you clear types throughout the codebase.

# models.py
from __future__ import annotations
from typing import Generic, TypeVar
from pydantic import BaseModel
 
T = TypeVar("T")
 
 
class Meta(BaseModel):
    total: int
    limit: int
    offset: int
 
 
class SheetResponse(BaseModel, Generic[T]):
    data: list[T]
    meta: Meta
 
 
# Concrete row shape for a "contacts" sheet
class ContactRow(BaseModel):
    id: str | None = None
    name: str
    email: str
    status: str = "pending"

SheetResponse is generic so the same model works for every sheet: SheetResponse[ContactRow], SheetResponse[OrderRow], and so on. FastAPI and Pydantic v2 resolve the generic at route definition time and generate the correct OpenAPI schema for each.

Async client with dependency injection

httpx.AsyncClient is the right tool here. It supports connection pooling and HTTP/2, and it slots naturally into FastAPI's async request handlers. Create one client instance at startup and share it across all requests via Depends.

# client.py
import httpx
 
SHEETS_API_BASE = "https://sheetsapi.io/api/spreadsheets"
 
 
class SheetsClient:
    def __init__(self, user_key: str, api_key: str) -> None:
        self._base = f"{SHEETS_API_BASE}/{user_key}"
        self._http = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10.0,
        )
 
    async def get_rows(
        self,
        sheet: str,
        *,
        limit: int = 20,
        offset: int = 0,
        filters: dict[str, str] | None = None,
    ) -> dict:
        params: dict[str, str | int] = {"limit": limit, "offset": offset}
        if filters:
            for col, val in filters.items():
                params[f"filter[{col}]"] = val
        resp = await self._http.get(f"{self._base}/{sheet}", params=params)
        resp.raise_for_status()
        return resp.json()
 
    async def append_row(self, sheet: str, row: dict) -> dict:
        resp = await self._http.post(f"{self._base}/{sheet}", json=row)
        resp.raise_for_status()
        return resp.json()
 
    async def aclose(self) -> None:
        await self._http.aclose()

Wire the client into FastAPI's lifespan and expose it through a dependency function:

# dependencies.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from client import SheetsClient
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    client = SheetsClient(
        user_key=os.environ["SHEETS_USER_KEY"],
        api_key=os.environ["SHEETS_API_KEY"],
    )
    app.state.sheets = client
    yield
    await client.aclose()
 
 
def get_sheets_client(request: Request) -> SheetsClient:
    return request.app.state.sheets

Storing the client on app.state inside the lifespan context means it is created once, reused across all requests, and cleanly closed on shutdown. No global variables, no module-level imports that break under multiple workers.

Simple TTL cache

functools.lru_cache works for pure functions but never expires. A small TTL dict is enough for most sheet use cases - store (timestamp, payload) and evict on read when stale:

# cache.py
import time
from typing import Any
 
_store: dict[str, tuple[float, Any]] = {}
TTL = 60  # seconds
 
 
def cache_get(key: str) -> Any | None:
    entry = _store.get(key)
    if entry is None:
        return None
    ts, value = entry
    if time.monotonic() - ts > TTL:
        del _store[key]
        return None
    return value
 
 
def cache_set(key: str, value: Any) -> None:
    _store[key] = (time.monotonic(), value)
 
 
def cache_invalidate(prefix: str) -> None:
    keys = [k for k in _store if k.startswith(prefix)]
    for k in keys:
        del _store[k]

Route handlers

# routes/contacts.py
from typing import Annotated
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
from client import SheetsClient
from dependencies import get_sheets_client
from models import ContactRow, SheetResponse
from cache import cache_get, cache_invalidate, cache_set
 
router = APIRouter(prefix="/contacts", tags=["contacts"])
 
SHEET = "contacts"
 
 
@router.get("/", response_model=SheetResponse[ContactRow])
async def list_contacts(
    client: Annotated[SheetsClient, Depends(get_sheets_client)],
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0),
    status: str | None = Query(None),
) -> SheetResponse[ContactRow]:
    cache_key = f"{SHEET}:{limit}:{offset}:{status}"
    cached = cache_get(cache_key)
    if cached is not None:
        return cached
 
    filters = {"status": status} if status else None
    raw = await client.get_rows(SHEET, limit=limit, offset=offset, filters=filters)
 
    result = SheetResponse[ContactRow].model_validate(raw)
    cache_set(cache_key, result)
    return result
 
 
@router.post("/", response_model=ContactRow, status_code=201)
async def create_contact(
    body: ContactRow,
    background_tasks: BackgroundTasks,
    client: Annotated[SheetsClient, Depends(get_sheets_client)],
) -> ContactRow:
    try:
        await client.append_row(SHEET, body.model_dump(exclude_none=True))
    except Exception as exc:
        raise HTTPException(status_code=502, detail=str(exc)) from exc
 
    background_tasks.add_task(cache_invalidate, SHEET)
    return body

A few decisions worth noting:

Cache invalidation runs in a background task. BackgroundTasks executes after the response is sent, so the client gets a 201 immediately and the stale cache is cleared without adding latency to the write path.

response_model is set on the route, not just on the return annotation. FastAPI uses response_model for output filtering and OpenAPI schema generation. Setting it explicitly ensures the generic resolves correctly in both Pydantic v2 and the generated docs.

exclude_none=True on POST. SheetsAPI maps JSON keys to column headers. Sending null for optional columns writes empty cells rather than skipping them. Excluding None fields keeps the appended row clean.

Application entry point

# main.py
from fastapi import FastAPI
from dependencies import lifespan
from routes.contacts import router as contacts_router
 
app = FastAPI(title="Sheets-backed API", lifespan=lifespan)
app.include_router(contacts_router)

Run it:

SHEETS_USER_KEY=your_key SHEETS_API_KEY=sk_... uvicorn main:app --reload

The interactive docs are available at http://localhost:8000/docs with the full schema for SheetResponse[ContactRow] inlined.

What to reach for next

This setup covers the common path. A few extensions worth considering:

Redis for the cache. The TTL dict is process-local and breaks under multiple Uvicorn workers. Swap in redis.asyncio and replace cache_get/cache_set with await redis.get / await redis.setex. The cache_invalidate function becomes await redis.delete(*keys) after a SCAN.

Retry logic. Pass transport=httpx.AsyncHTTPTransport(retries=3) when constructing the AsyncClient, or use tenacity for exponential backoff on 5xx responses from SheetsAPI.

Multiple sheets, one client. SheetsClient is sheet-agnostic - the sheet name is just a path segment. Add a second router for a different tab, pass the sheet name as a parameter, and reuse the same injected client. No new infrastructure required.

Stronger typing on filters. The filters dict accepts any string key. If you want to prevent callers from filtering on arbitrary columns, replace it with an explicit query param per filterable column and build the dict from those named params in the route handler.

The core pattern - typed models at the boundary, a single pooled async client injected via Depends, and background-task cache invalidation - scales cleanly to any number of sheets without changing the architecture.

All posts