Incremental Fetch

Incremental Fetch

Build data pipelines that never lose progress and never re-fetch existing data.

The Two Watermarks Pattern

Track TWO cursors to support both forward and backward fetching:

A single watermark only fetches forward. Two watermarks enable:

• Regular runs: fetch NEW data (since newest_id)
• Backfill runs: fetch OLD data (until oldest_id)
• No overlap, no gaps

Critical: Data vs Watermark Saving

These are different operations with different timing:

fetch page 1 → save records → fetch page 2 → save records → ... → update watermarks

Workflow Decision Tree

First run (no watermarks)?
├── YES → Full fetch (no since_id, no until_id)
└── NO → Backfill flag set?
    ├── YES → Backfill mode (until_id = oldest_id)
    └── NO → Update mode (since_id = newest_id)

Implementation Checklist

1. Database: Create ingestion_state table (see patterns.md) 2. Fetch loop: Insert records immediately after each API page 3. Watermark tracking: Track newest/oldest IDs seen in this run 4. Watermark update: Save watermarks ONCE at end of successful run 5. Retry: Exponential backoff with jitter 6. Rate limits: Wait for reset or skip and record for next run

Pagination Types

This pattern works best with ID-based pagination (numeric IDs that can be compared). For other pagination types:

See references/patterns.md for schemas and code examples.

Gotchas

• Save watermarks only after full success. If the process crashes mid-run, unsaved watermarks mean the next run re-fetches and deduplicates from scratch — no data loss, but potentially slow. Saving watermarks mid-run causes permanent gaps.
• Newest ID may not equal the highest numeric ID. Some APIs return IDs that are not monotonically increasing (e.g., snowflake IDs with clock drift). Always compare using the API's own ordering guarantees, not numeric comparison.
• Backfill mode must not overwrite the `newest_id`. A backfill run extends history backward; it should update only oldest_id. Overwriting newest_id during backfill causes duplicate fetches on the next forward update run.
• Rate-limit headers vary by API. Twitter uses x-rate-limit-reset; others use Retry-After. Check the specific API's response headers before implementing wait logic.

incremental-fetch

Build resilient data ingestion pipelines from APIs.

When to Use

• Fetching paginated data from APIs
• Need to track progress and avoid duplicates
• Support both new data and historical backfills
• "ingest from API", "pull tweets", "backfill data"

What It Does

Uses a two-watermark pattern:

Key Rules

1. Save records after each page (resilience) 2. Save watermarks once at end (correctness) 3. Never re-fetch existing data 4. Resume from interruption without data loss

Resources

• references/patterns.md - Schemas and code examples

Incremental Fetch Patterns

Code patterns and schemas for implementing resilient incremental data fetching.

Database Schema

Ingestion State Table

CREATE TABLE IF NOT EXISTS ingestion_state (
    asset_id VARCHAR NOT NULL,      -- entity being fetched (user, account, symbol)
    data_type VARCHAR NOT NULL,     -- watermark type (see below)
    last_id VARCHAR,                -- watermark value (string for large ints)
    last_timestamp TIMESTAMP,       -- optional: timestamp-based watermark
    updated_at TIMESTAMP DEFAULT now(),
    PRIMARY KEY (asset_id, data_type)
);

Why Separate Rows for Each Watermark Type

The data_type column stores different watermark types as separate rows, not columns:

asset_id    | data_type       | last_id
------------|-----------------|--------
@elonmusk   | tweets          | 1234567  (newest fetched)
@elonmusk   | tweets_oldest   | 1000000  (oldest fetched)
BTCUSD      | prices_1h       | 9876543
BTCUSD      | prices_1h_oldest| 8000000

This design allows:

• N watermark types per entity without schema changes
• Different data types (tweets, prices_1h, prices_1d) tracked independently
• Bidirectional fetching (newest + oldest) per data type

Watermark Functions

def get_ingestion_state(conn, asset_id: str, data_type: str) -> dict | None:
    result = conn.execute("""
        SELECT last_id, last_timestamp, updated_at
        FROM ingestion_state
        WHERE asset_id = ? AND data_type = ?
    """, [asset_id, data_type]).fetchone()

    if not result:
        return None
    return {"last_id": result[0], "last_timestamp": result[1]}


def update_ingestion_state(conn, asset_id: str, data_type: str,
                           last_id: str = None, last_timestamp = None):
    conn.execute("""
        INSERT INTO ingestion_state (asset_id, data_type, last_id, last_timestamp, updated_at)
        VALUES (?, ?, ?, ?, now())
        ON CONFLICT (asset_id, data_type) DO UPDATE SET
            last_id = COALESCE(EXCLUDED.last_id, ingestion_state.last_id),
            last_timestamp = COALESCE(EXCLUDED.last_timestamp, ingestion_state.last_timestamp),
            updated_at = now()
    """, [asset_id, data_type, last_id, last_timestamp])

Fetch Loop Pattern

def fetch_for_asset(asset_id: str, backfill: bool = False, max_pages: int = 50):
    conn = get_connection()

    # Get watermarks (separate rows in DB)
    state = get_ingestion_state(conn, asset_id, "tweets")
    oldest_state = get_ingestion_state(conn, asset_id, "tweets_oldest")

    newest_id = state.get("last_id") if state else None
    oldest_id = oldest_state.get("last_id") if oldest_state else None

    # Determine fetch mode
    if backfill and oldest_id:
        since_id = None
        until_id = oldest_id  # Fetch older than this
    elif newest_id:
        since_id = newest_id  # Fetch newer than this
        until_id = None
    else:
        since_id = None  # Full fetch
        until_id = None

    # Track watermarks for THIS run
    run_newest_id = None
    run_oldest_id = None
    pagination_token = None

    for page in range(max_pages):
        # Fetch one page
        items, next_token = fetch_page(
            asset_id,
            since_id=since_id,
            until_id=until_id,
            pagination_token=pagination_token
        )

        if not items:
            break

        # Track watermarks (compare as INT for numeric IDs)
        for item in items:
            item_id = item["id"]
            if run_newest_id is None or int(item_id) > int(run_newest_id):
                run_newest_id = item_id
            if run_oldest_id is None or int(item_id) < int(run_oldest_id):
                run_oldest_id = item_id

        # SAVE DATA IMMEDIATELY after each page (resilience)
        insert_items(conn, asset_id, items)
        print(f"Page {page + 1}: {len(items)} items saved")

        if not next_token:
            break
        pagination_token = next_token

    # UPDATE WATERMARKS ONCE at end (correctness)
    if run_newest_id:
        if newest_id is None or int(run_newest_id) > int(newest_id):
            update_ingestion_state(conn, asset_id, "tweets", last_id=run_newest_id)

    if run_oldest_id:
        update_ingestion_state(conn, asset_id, "tweets_oldest", last_id=run_oldest_id)

    conn.close()

Retry Pattern

import time
import random

def fetch_with_retry(url: str, params: dict, max_attempts: int = 3):
    for attempt in range(max_attempts):
        try:
            response = client.get(url, params=params, timeout=30.0)

            if response.status_code == 429:  # Rate limit
                reset_ts = response.headers.get("x-rate-limit-reset", "0")
                wait = max(0, int(reset_ts) - int(time.time()) + 5)

                if wait > 120:  # Don't wait more than 2 minutes
                    return None, "rate_limit_skip"

                print(f"Rate limited, waiting {wait}s...")
                time.sleep(wait)
                continue

            if response.status_code != 200:
                return None, f"http_{response.status_code}"

            return response.json(), None

        except TimeoutError:
            # Exponential backoff with jitter
            wait = (2 ** attempt) * 5 + random.uniform(0, 3)
            print(f"Timeout, waiting {wait:.0f}s (attempt {attempt + 1}/{max_attempts})")
            time.sleep(wait)

    return None, "max_retries_exceeded"

Gotchas

1. Compare IDs as integers

Tweet IDs and similar identifiers are numeric but may be stored as strings. Always compare as int:

# WRONG: string comparison ("9" > "10" is True)
if item_id > run_newest_id:

# RIGHT: numeric comparison
if int(item_id) > int(run_newest_id):

2. Data saves vs watermark updates are DIFFERENT

This is the #1 mistake. The timing is different for a reason:

• Data: Save per-page for resilience (crash recovery)
• Watermarks: Save at end for correctness (don't claim progress until complete)

3. Handle backfill-without-data error

If user requests --backfill but no data exists yet, error loudly:

if backfill and not oldest_id:
    oldest_in_db = conn.execute("SELECT MIN(id) FROM items WHERE asset_id = ?", [asset_id]).fetchone()[0]
    if not oldest_in_db:
        raise Error("Cannot backfill - no data exists. Run without --backfill first.")

4. Rate limit skip tracking

When you hit rate limits and skip an asset, record it for next run:

FETCH_STATE_FILE = Path("data/fetch_state.json")

def write_fetch_state(skipped_assets: list, reason: str):
    state = {
        "last_run": datetime.now().isoformat(),
        "skipped_assets": skipped_assets,
        "skip_reason": reason
    }
    FETCH_STATE_FILE.write_text(json.dumps(state, indent=2))

Then prioritize skipped assets on next run.

5. Use ON CONFLICT for upserts

When inserting data, handle duplicates gracefully:

conn.execute("""
    INSERT INTO items (id, asset_id, data, fetched_at)
    VALUES (?, ?, ?, now())
    ON CONFLICT (id) DO UPDATE SET
        data = EXCLUDED.data,
        fetched_at = now()
""", [item_id, asset_id, data])

This allows re-running without duplicate key errors and updates stale data.