How do I automate daily pulls from a data API into Snowflake?

A solid way to automate daily API pulls into Snowflake is:

1. Extract data from the API on a schedule
2. Land the raw response in cloud storage or directly in Snowflake staging
3. Load it into Snowflake
4. Transform it with SQL/dbt if needed

Common patterns

Option 1: Simple and reliable

Use a scheduled job (cron, Airflow, Dagster, Prefect, GitHub Actions, etc.) that:

• calls the API daily
• writes JSON/CSV files to S3/GCS/Azure Blob
• triggers a Snowflake COPY INTO load

Best for:

• predictable daily loads
• moderate data volumes
• easy retries/auditing

Option 2: Direct to Snowflake

A scheduled script:

• calls the API
• inserts rows directly into Snowflake via Python connector or Snowpark

Best for:

• smaller datasets
• quick setup

Option 3: Orchestrated pipeline

Use Airflow/Prefect/Dagster:

• fetch API data
• store raw data
• load to Snowflake
• run transformations and data quality checks

Best for:

• production pipelines
• multiple APIs/tables
• dependencies and retries

---

Recommended architecture

API → Python job → cloud storage → Snowflake stage → raw table → transformed tables

Why this is good:

• keeps a raw copy of source data
• easier debugging and replay
• Snowflake load is fast and robust
• avoids partial-load issues

---

Example implementation

1) Fetch data in Python

import requests
import json
from datetime import date

API_URL = "https://api.example.com/data"
API_TOKEN = "your_token"

headers = {"Authorization": f"Bearer {API_TOKEN}"}
response = requests.get(API_URL, headers=headers)
response.raise_for_status()

data = response.json()

with open(f"data_{date.today()}.json", "w") as f:
    json.dump(data, f)

2) Upload to cloud storage

For example, to S3 using boto3:

import boto3

s3 = boto3.client("s3")
s3.upload_file(f"data_{date.today()}.json", "my-bucket", f"raw/data_{date.today()}.json")

3) Load into Snowflake

Create a file format and stage:

CREATE OR REPLACE FILE FORMAT my_json_format
  TYPE = JSON;

CREATE OR REPLACE STAGE my_stage
  URL = 's3://my-bucket/raw/'
  STORAGE_INTEGRATION = my_s3_integration
  FILE_FORMAT = my_json_format;

Load into a raw table:

CREATE OR REPLACE TABLE raw_api_data (
  payload VARIANT,
  load_ts TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

COPY INTO raw_api_data (payload)
FROM @my_stage
FILE_FORMAT = (TYPE = JSON)
ON_ERROR = 'CONTINUE';

---

If you prefer direct insert from Python

import snowflake.connector
import requests
import json

conn = snowflake.connector.connect(
    user="USER",
    password="PASSWORD",
    account="ACCOUNT",
    warehouse="WAREHOUSE",
    database="DB",
    schema="SCHEMA",
    role="ROLE"
)

cur = conn.cursor()

response = requests.get("https://api.example.com/data")
response.raise_for_status()
records = response.json()

for record in records:
    cur.execute(
        "INSERT INTO raw_api_data (payload) VALUES (%s)",
        (json.dumps(record),)
    )

cur.close()
conn.close()

This is easy, but for larger datasets, bulk loading is better.

---

Scheduling

To run daily:

• cron on a VM or server
• Airflow DAG with daily schedule
• Prefect/Dagster flow
• GitHub Actions scheduled workflow
• AWS Lambda + EventBridge if lightweight

Example cron:

0 2 * * * /usr/bin/python3 /path/to/pipeline.py

---

Best practices

• Use incremental loading if the API supports updated_since, cursor pagination, or timestamps
• Store raw data before transforming
• Handle rate limits with retries/backoff
• Track loaded files or watermarks to avoid duplicates
• Use Snowflake streams/tasks if you want downstream automation
• Log failures and alert on errors
• Secure secrets with a secrets manager, not hardcoded values

---

Minimal production setup

If you want the simplest robust setup:

• Python script fetches API data
• Save JSON to S3/GCS
• Use Snowflake COPY INTO
• Schedule with cron or Airflow
• Transform with dbt

If you want, I can give you:

• a full working Python + Snowflake example
• an Airflow DAG
• or a serverless AWS setup for daily API ingestion