Blog

Exporting Google Sheets data to Redshift, BigQuery, and other data warehouses

How to get data out of Google Sheets and into Redshift, BigQuery, Snowflake, or any other warehouse - using REST API exports instead of manual CSV downloads.

Jul 1, 20266 min read

#google-sheets #redshift #bigquery #data-warehouse #sheets-api #export

Your ops team manages a lead tracker in Google Sheets. Your marketing team updates a campaign budget sheet every Monday. Your data team wants both in Redshift so they can join them against event data and build dashboards. The workflow that emerges is always the same: someone manually downloads a CSV, uploads it to S3, and runs a COPY command. It works once. It breaks the second week when someone forgets, renames a column, or downloads the wrong tab.

The better path is to treat the sheet as a data source with a stable URL. Pull from it programmatically, on a schedule, with no human in the loop.

The SheetsAPI approach

SheetsAPI exposes any Google Sheet as a REST endpoint. Connect your sheet once in the dashboard, and you get a URL that returns your data as CSV, JSON, TSV, or XML - with filtering, sorting, and pagination built in.

The base endpoint shape:

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

Every example in this post uses plain HTTP. No Google Cloud Console, no OAuth, no service accounts. See the docs for authentication details.

Step 1: Pull data from the sheet

For warehouse loading, format=csv is usually the right choice. It streams directly into most bulk-load tools without a JSON-to-relational conversion step.

curl "https://api.sheetsapi.io/api/spreadsheets/abc123/leads?format=csv" \
  -H "Authorization: Bearer sk_YOUR_KEY"

For workflows where you want typed data or need to inspect the shape before loading, format=json returns the standard envelope:

curl "https://api.sheetsapi.io/api/spreadsheets/abc123/leads?format=json&limit=500" \
  -H "Authorization: Bearer sk_YOUR_KEY"

Sheets larger than 1,000 rows use limit and offset for pagination:

# Page 1
curl "...?format=csv&limit=1000&offset=0"
 
# Page 2
curl "...?format=csv&limit=1000&offset=1000"

The meta.total field in JSON responses tells you the full row count so you can calculate the number of pages before you start. For CSV, loop until you receive fewer rows than your limit.

Loading into Redshift

Redshift's fastest ingest path is COPY from S3. For smaller sheets or frequent incremental loads, a direct INSERT via psycopg2 is simpler and avoids the S3 round-trip.

import io
import csv
import requests
import psycopg2
 
SHEETS_URL = "https://api.sheetsapi.io/api/spreadsheets/abc123/leads"
SHEETS_KEY = "sk_YOUR_KEY"
 
REDSHIFT_DSN = "postgresql://user:pass@cluster.region.redshift.amazonaws.com:5439/analytics"
 
def fetch_csv(offset=0, limit=1000):
    resp = requests.get(
        SHEETS_URL,
        headers={"Authorization": f"Bearer {SHEETS_KEY}"},
        params={"format": "csv", "limit": limit, "offset": offset},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.text
 
def load_to_redshift(csv_text: str, conn):
    reader = csv.DictReader(io.StringIO(csv_text))
    rows = list(reader)
    if not rows:
        return 0
 
    cols = ", ".join(rows[0].keys())
    placeholders = ", ".join(["%s"] * len(rows[0]))
    sql = f"INSERT INTO staging.leads ({cols}) VALUES ({placeholders})"
 
    with conn.cursor() as cur:
        cur.executemany(sql, [list(r.values()) for r in rows])
    conn.commit()
    return len(rows)
 
conn = psycopg2.connect(REDSHIFT_DSN)
 
# Truncate staging table before full refresh
with conn.cursor() as cur:
    cur.execute("TRUNCATE TABLE staging.leads")
conn.commit()
 
offset = 0
total = 0
while True:
    csv_text = fetch_csv(offset=offset)
    loaded = load_to_redshift(csv_text, conn)
    total += loaded
    if loaded < 1000:
        break
    offset += 1000
 
conn.close()
print(f"Loaded {total} rows into staging.leads")

After staging, merge into your production table with a DELETE + INSERT or a Redshift MERGE statement based on a row key.

Loading into BigQuery

BigQuery's load_table_from_json accepts a list of dicts directly, which maps cleanly to SheetsAPI's JSON output.

import requests
from google.cloud import bigquery
 
SHEETS_URL = "https://api.sheetsapi.io/api/spreadsheets/abc123/leads"
SHEETS_KEY = "sk_YOUR_KEY"
BQ_TABLE = "my-project.analytics.leads_staging"
 
client = bigquery.Client()
 
def fetch_all_rows():
    rows = []
    offset = 0
    limit = 1000
    while True:
        resp = requests.get(
            SHEETS_URL,
            headers={"Authorization": f"Bearer {SHEETS_KEY}"},
            params={"format": "json", "limit": limit, "offset": offset},
            timeout=30,
        )
        resp.raise_for_status()
        body = resp.json()
        batch = body["data"]
        rows.extend(batch)
        if len(batch) < limit:
            break
        offset += limit
    return rows
 
rows = fetch_all_rows()
 
job_config = bigquery.LoadJobConfig(
    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    autodetect=True,
)
 
job = client.load_table_from_json(rows, BQ_TABLE, job_config=job_config)
job.result()  # wait for completion
 
print(f"Loaded {job.output_rows} rows into {BQ_TABLE}")

autodetect=True infers the schema from the first batch of rows. In production, define an explicit schema so a column type change in the sheet doesn't silently corrupt downstream queries.

Loading into Snowflake

Snowflake's connector supports staging a local file directly, which keeps the script simple.

import io
import requests
import snowflake.connector
 
SHEETS_URL = "https://api.sheetsapi.io/api/spreadsheets/abc123/leads"
SHEETS_KEY = "sk_YOUR_KEY"
 
SF_CONFIG = {
    "account": "xy12345.us-east-1",
    "user": "LOADER_USER",
    "password": "...",
    "warehouse": "LOADING_WH",
    "database": "ANALYTICS",
    "schema": "STAGING",
}
 
def fetch_full_csv():
    pages = []
    offset = 0
    limit = 1000
    while True:
        resp = requests.get(
            SHEETS_URL,
            headers={"Authorization": f"Bearer {SHEETS_KEY}"},
            params={"format": "csv", "limit": limit, "offset": offset},
            timeout=30,
        )
        resp.raise_for_status()
        text = resp.text
        # Skip header on subsequent pages
        if offset > 0:
            text = "\n".join(text.splitlines()[1:])
        pages.append(text)
        if text.count("\n") < limit:
            break
        offset += limit
    return "\n".join(pages)
 
csv_text = fetch_full_csv()
csv_bytes = io.BytesIO(csv_text.encode("utf-8"))
 
ctx = snowflake.connector.connect(**SF_CONFIG)
cs = ctx.cursor()
 
cs.execute("CREATE OR REPLACE STAGE tmp_leads_stage FILE_FORMAT = (TYPE = CSV FIELD_OPTIONALLY_ENCLOSED_BY = '\"' SKIP_HEADER = 1)")
ctx.cursor().execute("PUT file://leads.csv @tmp_leads_stage", file_stream=csv_bytes)
cs.execute("COPY INTO leads FROM @tmp_leads_stage PURGE = TRUE")
 
ctx.close()

Any SQLAlchemy-compatible warehouse

If your warehouse has a SQLAlchemy dialect - DuckDB, Postgres, MySQL, Databricks, CockroachDB - the pandas path works without any warehouse-specific code:

import io
import requests
import pandas as pd
from sqlalchemy import create_engine
 
SHEETS_URL = "https://api.sheetsapi.io/api/spreadsheets/abc123/leads"
SHEETS_KEY = "sk_YOUR_KEY"
DB_URL = "postgresql+psycopg2://user:pass@host:5432/analytics"
 
def fetch_dataframe():
    pages = []
    offset = 0
    limit = 1000
    while True:
        resp = requests.get(
            SHEETS_URL,
            headers={"Authorization": f"Bearer {SHEETS_KEY}"},
            params={"format": "csv", "limit": limit, "offset": offset},
            timeout=30,
        )
        resp.raise_for_status()
        df = pd.read_csv(io.StringIO(resp.text))
        pages.append(df)
        if len(df) < limit:
            break
        offset += limit
    return pd.concat(pages, ignore_index=True)
 
engine = create_engine(DB_URL)
df = fetch_dataframe()
 
df.to_sql(
    "leads_staging",
    con=engine,
    schema="staging",
    if_exists="replace",
    index=False,
    chunksize=500,
    method="multi",
)
 
print(f"Loaded {len(df)} rows")

The CSV to JSON tool is useful for inspecting the column shapes before you commit to a target schema. If you are new to SheetsAPI's query parameters, the Python guide covers search, sort, and field filtering in detail.

Incremental loads

Full-refresh every run is fine for sheets under a few thousand rows. For larger sheets or tight scheduling windows, use the search parameter to pull only rows modified since the last run.

If your sheet has an updated_at column:

# Rows updated on or after 2026-07-01
curl "...?search=updated_at:2026-07-01&sort=-updated_at"

To always fetch the most recent 100 rows by insertion order:

curl "...?sort=-created_at&limit=100"

Store the timestamp of the last successful run in a file or a metadata table, and pass it as the search value on the next execution.

Using SheetsAPI as a dbt external source

If your data team already uses dbt, you can wire the SheetsAPI CSV URL directly as an external source and skip the Python loader entirely. Add a source definition to your dbt_project.yml:

# dbt_project.yml
sources:
  - name: sheets
    description: "Live data from Google Sheets via SheetsAPI"
    tables:
      - name: leads
        external:
          location: "https://api.sheetsapi.io/api/spreadsheets/abc123/leads?format=csv"
          options:
            format: csv
            skip_leading_rows: 1

BigQuery's external table feature and Snowflake's external stage both accept a URL directly. For Redshift, use a Lambda or Airflow task to stage the file to S3 first, then reference the S3 path in your dbt source.

Scheduling the load

The script runs anywhere Python runs. Common options:

Cron on a VM or container

# /etc/cron.d/sheets-to-redshift
0 6 * * * ubuntu /usr/bin/python3 /opt/etl/load_leads.py >> /var/log/etl/leads.log 2>&1

AWS Lambda - package the script with requests and psycopg2-binary, trigger on EventBridge schedule. Cold-start time is negligible for HTTP + INSERT workloads.

Cloud Functions / Cloud Run - same pattern. The BigQuery client library is pre-installed in the Python 3.11 base image so the deployment package stays small.

Airflow / Prefect - wrap fetch_dataframe() and df.to_sql() in a task. Set retries=3 and retry_delay=timedelta(minutes=2) so transient network errors don't fail the DAG.

dbt source freshness - if you are using the dbt external source approach, run dbt source freshness on a schedule to alert when the sheet has not been updated within an expected window.

Comparison: SheetsAPI export vs managed connectors vs manual CSV

	SheetsAPI export	Fivetran / Airbyte / Stitch	Manual CSV
Setup time	~15 minutes	30–60 minutes + connector config	Minutes per run
Ongoing maintenance	None	Connector version updates	Human time every cycle
Cost	See pricing	$0.25–$2 / 1,000 MAR (Fivetran)	Free
Scheduling	Cron / Lambda / Airflow	Managed, configurable	Manual
Incremental support	Yes, via search params	Yes, managed	No
Custom filtering	Yes - search, sort, fields	Limited	No
Works in dbt	Yes, external source	Yes, via generated models	No

Managed connectors earn their cost when you have dozens of sources and a dedicated data team to maintain them. For one or two Sheets-based sources, writing a 40-line Python script against a REST endpoint is cheaper and gives you full control over the load logic.

Next step

Connect your first sheet in the dashboard and copy the base URL. The SheetsAPI docs have the full parameter reference, authentication guide, and rate limit details. If you are evaluating GKit for your team, the about page covers what else is in the toolkit alongside SheetsAPI.

The free tools include a CSV to JSON converter and several other utilities that are useful when inspecting export payloads before loading.

All posts