How to extract incremental changes from Data APIs: step by step

Learn how to extract only the new or changed records from a Data API — useful to reduce traffic, speed up ETL and keep a local copy up to date. We will explain why and show a Python example that detects and processes incremental changes.

Prerequisites

• Python 3.8+ installed
• Libraries: requests and sqlite3 (bundled with Python)
• A REST API that exposes a date/timestamp field or an incremental field (e.g.: updated_at or id)
• Basic knowledge of HTTP and JSON

Step 1: Understand the API change model

Before writing code, check if the API has a field that indicates changes (for example updated_at) or an incremental identifier. If it has updated_at in UTC, you can request only records after a date. If there is only an incremental id, you can request id > last_id. Knowing this avoids pulling all data and is the basis of incremental extraction.

Step 2: Structure a control storage (state)

You need to store the last timestamp or last id extracted. A simple file or a small SQLite database works well for a simple ETL flow. Here we use SQLite for safe persistence.

import sqlite3

def init_state_db(path='state.db'):
    conn = sqlite3.connect(path)
    cur = conn.cursor()
    cur.execute('''CREATE TABLE IF NOT EXISTS etl_state (
                    key TEXT PRIMARY KEY,
                    value TEXT
                  )''')
    conn.commit()
    return conn

def get_state(conn, key):
    cur = conn.cursor()
    cur.execute('SELECT value FROM etl_state WHERE key=?', (key,))
    row = cur.fetchone()
    return row[0] if row else None

def set_state(conn, key, value):
    cur = conn.cursor()
    cur.execute('REPLACE INTO etl_state (key, value) VALUES (?, ?)', (key, value))
    conn.commit()

Step 3: Make a conditional API call

Build the query using the stored value. Example with an updated_after parameter (many APIs accept something similar). Handle common errors like 400/500 and rate limit issues; if filtering is not supported, you will need to use pagination and filter locally.

import requests
from datetime import datetime

API_URL = 'https://api.exemplo.com/items'

def fetch_incremental(updated_after=None, page=1):
    params = {'page': page}
    if updated_after:
        params['updated_after'] = updated_after  # parameter name depends on the API
    resp = requests.get(API_URL, params=params, timeout=10)
    resp.raise_for_status()
    return resp.json()  # assumes JSON with list and metadata

Step 4: Process results and update the state

When receiving the records, process them (write to your local database, transform, etc.) and compute the new state value: max(updated_at) or max(id). Only after successful persistence update the state to avoid data loss.

def process_and_update_state(conn, items):
    # Minimal example: write to a file or local DB (full write implementation omitted)
    # Assumes each item has 'id' and 'updated_at' in ISO 8601
    if not items:
        return None
    max_ts = max(item['updated_at'] for item in items)
    # here we would persist the items in a table or files
    set_state(conn, 'last_updated_at', max_ts)
    return max_ts

Step 5: Put it all together in a safe loop

Run a loop that reads the state, calls the API in pages until there are no more results, processes and updates the state at the end. Handle transient errors with simple retries and respect rate limits with linear or exponential backoff.

import time

def run_once(conn):
    last = get_state(conn, 'last_updated_at')
    page = 1
    all_items = []
    while True:
        try:
            data = fetch_incremental(updated_after=last, page=page)
        except requests.HTTPError as e:
            print('HTTP Error:', e)
            break
        items = data.get('items', [])
        if not items:
            break
        # processing logic here; we accumulate to update state at the end
        all_items.extend(items)
        page += 1
        # simple safeguard: if the API doesn't support pagination by updated_after, you may need to stop for safety
        if page > 1000:
            break
        time.sleep(0.2)  # smooth out calls

    if all_items:
        new_state = process_and_update_state(conn, all_items)
        print('State updated to', new_state)
    else:
        print('No changes')

if __name__ == '__main__':
    conn = init_state_db()
    run_once(conn)

Verify the result

Confirm that the state field was written to the etl_state table (SELECT * FROM etl_state). Also check that only records with updated_at > previous last_updated_at were imported. Run twice: the first run should import data and write the state; the second should import nothing if there are no changes.

Conclusion

With a simple state and conditional calls you can turn a Data API into an efficient incremental flow: less traffic, faster ETL and lower latency. Next steps: add retries with exponential backoff, detailed logging and automated tests. Tip: start by validating the API timestamp format to avoid timezone and parsing errors.