How to Fetch Data from an API and Store It in a SQL Database with Python

Fetching data from an API and storing it in a SQL database with Python requires five steps: make a reliable HTTPS request, authenticate safely, validate the JSON response, map records to a table, and insert them with parameterized SQL. SQLite is ideal for a self-contained example; PostgreSQL is better suited to shared or production workloads.

The workflow is:

1. Request data from the external API.
2. Check the HTTP status and response format.
3. Extract and validate records.
4. Transform API fields into database columns.
5. Insert or upsert the records inside a transaction.

Key takeaways

• Python Requests supports query parameters, headers, authentication, JSON parsing, timeouts, sessions, and HTTP error handling.
• SQLite is included with Python and is the simplest database for learning or running a small local importer.
• Parameterized SQL, such as SQLite’s ? placeholders or Psycopg’s %s placeholders, prevents values from being interpreted as SQL code.
• Pagination, retries, rate limits, transactions, and duplicate handling are required for a dependable recurring import.
• An API response shape, authentication method, pagination model, and stopping condition are provider-specific and must be adapted from the API documentation.

What does fetching data from an API and storing it in a SQL database mean?

Fetching data from an API and storing it in a SQL database means using a Python application as the bridge between an external HTTP service and a local or hosted relational database.

#	Preview	Product	Price
1		Sandisk 2TB Extreme Portable SSD, Up to 1050MB/s, USB-C, USB 3.2 Gen 2, IP65 Water and Dust...	$289.99	Buy on Amazon
2		Sandisk 1TB Portable SSD, Up to 800MB/s Read Speeds, Black (Old Model)	$188.90	Buy on Amazon
3		SSK Portable SSD 500GB External Solid State Hard Drive USB C Up to 1050MB/s	$89.99	Buy on Amazon
4		Seagate 2TB Portable Hard Drive | USB 3.0 (STGX2000400)	$119.99	Buy on Amazon
5		Samsung T7 Portable SSD 1TB Titan Gray, USB 3.2 Gen 2, Up to 1,050MB/s	$229.99	Buy on Amazon

External API
    ↓ HTTPS request
Python application
    ↓ parse, validate, transform
SQL database
    ↓ queries, reports, application use
Stored data

An API response is normally a temporary representation of data. A SQL database gives the data local persistence, repeatable queries, joins, indexes, constraints, and the ability to build reports or applications without requesting the same records repeatedly.

This workflow is different from querying a database through an API. It is also different from building an API backed by a SQL database, or using an API that exposes SQL queries. The tutorial below reads from an external API and writes the returned records into SQL.

#1 Best Overall

Sandisk 2TB Extreme Portable SSD, Up to 1050MB/s, USB-C, USB 3.2 Gen 2, IP65 Water and Dust Resistance, Updated Firmware, External Solid State Drive, SDSSDE61-2T00-G25

• Get NVMe solid state performance with up to 1050MB/s read and 1000MB/s write speeds in a portable, high-capacity drive(1) (Based on internal testing; performance may be lower depending on host device & other factors. 1MB=1,000,000 bytes.)
• Up to 3-meter drop protection and IP65 water and dust resistance mean this tough drive can take a beating(3) (Previously rated for 2-meter drop protection and IP55 rating. Now qualified for the higher, stated specs.)
• Use the handy carabiner loop to secure it to your belt loop or backpack for extra peace of mind.
• Help keep private content private with the included password protection featuring 256‐bit AES hardware encryption.(3)
• Easily manage files and automatically free up space with the SanDisk Memory Zone app.(5). Non-Operating Temperature -20°C to 85°C

$289.99

Buy on Amazon

What do you need before importing API data into SQL?

You need Python, the target API’s documentation, and a SQL database that your script can reach.

• Use Python 3.10 or newer for the current Requests 2.x documentation’s supported workflow; check the installed package documentation if your environment uses an older Python release.
• Install packages with pip or another Python package manager.
• Obtain the API endpoint, required fields, authentication method, request limits, pagination rules, and error format from the provider’s documentation.
• Use SQLite for a local tutorial or small standalone job.
• Use PostgreSQL, MySQL, or SQL Server when the data must be shared by multiple users or applications.
• Make sure database credentials, network access, and firewall rules are configured for a server database.

Requests can be installed with:

python -m pip install requests

For PostgreSQL with Psycopg 3, install:

python -m pip install "psycopg[binary]"

The binary Psycopg package is convenient for many development environments, but deployment environments may instead prefer system libraries or a source build. SQLAlchemy is another option when an application needs database-engine abstraction, models, migrations, or connection pooling:

python -m pip install sqlalchemy

Python includes the sqlite3 module, so the SQLite example requires no separate database server. The Python sqlite3 documentation covers connections, parameter binding, batch execution, and transactions.

How should you read the API documentation first?

Read the API documentation before writing the importer because API-specific details determine almost every part of the program.

Many modern APIs return JSON, but an API can also return XML, CSV, binary data, or another format. The examples in this article deliberately use a fictional JSON API whose response shape must be adapted to the real provider.

How do you authenticate an API request safely?

Store production credentials outside the source code and send credentials according to the target API’s documented authentication method.

import os

API_TOKEN = os.environ["API_TOKEN"]

headers = {
    "Authorization": f"Bearer {API_TOKEN}",
    "Accept": "application/json",
}

Bearer authentication is only one common pattern. Other APIs use API keys, basic authentication, OAuth flows, signed requests, or provider-specific headers. The Requests authentication documentation describes supported authentication patterns, but the target API’s own documentation remains authoritative.

Use environment variables for deployed jobs. A .env file can be convenient during local development, but the file must be excluded from version control. Hosted applications should use a secret manager when one is available. Apply least privilege, rotate compromised credentials, and never log API tokens, complete authorization headers, passwords, or sensitive raw payload fields.

How do you make a safe API request with Python?

Use query parameters, explicit headers, a timeout, HTTP status validation, and JSON parsing rather than treating a single requests.get() call as a complete ingestion process.

import requests

url = "https://api.example.com/v1/items"

response = requests.get(
    url,
    params={"page": 1, "limit": 100},
    headers={
        "Accept": "application/json",
        # "Authorization": f"Bearer {API_TOKEN}",
    },
    timeout=30,
)

response.raise_for_status()
payload = response.json()

The params argument builds the query string without manually concatenating or escaping values. The headers argument handles content negotiation and often authentication. The timeout prevents a network operation from waiting indefinitely. raise_for_status() turns 4xx and 5xx responses into exceptions. The Requests documentation covers these request, response, timeout, session, and exception features.

A successful HTTP status does not prove that the payload has the expected structure. The server may return a valid JSON error object, an empty body, or a response with renamed fields. Validate the payload before writing anything to SQL.

How do you inspect and validate JSON returned by an API?

Inspect whether the response is a list or an object containing a collection, then check required fields and data types before constructing database rows.

APIs commonly return one of these shapes:

[
  {"id": 1, "name": "Alpha"}
]

{
  "data": [
    {"id": 1, "name": "Alpha"}
  ],
  "next_page": 2
}

{
  "results": [
    {"id": 1, "name": "Alpha"}
  ],
  "pagination": {
    "next": "https://api.example.com/v1/items?cursor=abc"
  }
}

The following helper accepts either a top-level list or a response envelope with a data field:

Rank #2

Sandisk 1TB Portable SSD, Up to 800MB/s Read Speeds, Black (Old Model)

• Solid state performance with up to 800MB/s read speeds in a portable drive. (Based on internal testing; performance may be lower depending on host device, interface, usage conditions and other factors. 1MB=1,000,000 bytes.)
• Back up your content and memories on a storage solution that fits seamlessly into your mobile lifestyle.
• Take it with you on your adventures—up to two-meter drop protection means this durable drive can take a beating. (Based on internal testing.)
• Secure it to your belt loop or backpack for extra peace of mind thanks to the tough rubber hook.
• From Sandisk, a brand professional photographers trust to take on assignments.

$188.90

Buy on Amazon

payload = response.json()

if isinstance(payload, dict):
    records = payload.get("data", [])
elif isinstance(payload, list):
    records = payload
else:
    raise ValueError("Unexpected API response type")

if not isinstance(records, list):
    raise ValueError("Expected records to be a list")

The extraction path is an example, not a universal rule. A real endpoint may use results, items, or a nested path. Validate required identifiers, uniqueness, timestamp formats, numeric values, nullable fields, and nested arrays deliberately. Pydantic can provide stronger model validation in a larger project, but it is not required for the introductory importer.

How should API fields map to SQL columns?

Transform each API record into a deliberate database row instead of creating a column for every JSON key automatically.

Use the API’s stable identifier as a primary key when the identifier is guaranteed to be unique for that source. Decide whether absent values become SQL NULL, whether timestamps are stored in UTC, and whether nested objects belong in child tables or a JSON column.

A useful hybrid design stores frequently queried fields in ordinary columns and preserves the original record in raw_json. The hybrid design helps debugging and future reprocessing, but it uses more storage and may retain sensitive information. Apply an explicit retention and privacy policy to raw payloads.

How do you create a SQL table for API records?

A SQLite table can combine relational columns with the original JSON payload and fetch timestamp.

CREATE TABLE IF NOT EXISTS items (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    price REAL,
    updated_at TEXT,
    raw_json TEXT NOT NULL,
    fetched_at TEXT NOT NULL
);

For SQLite, UTC ISO 8601 text is a practical timestamp representation. Use a database type appropriate to the target engine and application. PostgreSQL can use relational columns alongside JSONB:

CREATE TABLE IF NOT EXISTS items (
    id BIGINT PRIMARY KEY,
    name TEXT NOT NULL,
    price NUMERIC,
    updated_at TIMESTAMPTZ,
    raw_json JSONB NOT NULL,
    fetched_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

PostgreSQL JSONB is useful for semi-structured fields, but it should not replace relational columns that the application frequently filters, joins, constrains, or indexes. Repeated nested entities, such as orders with many line items, are usually better represented by related child tables.

How do you insert API records safely with parameterized SQL?

Build a list of values separately from the SQL statement, then use bound parameters and a batch operation.

import json
import sqlite3
from datetime import datetime, timezone

rows = [
    (
        item["id"],
        item.get("name"),
        item.get("price"),
        item.get("updated_at"),
        json.dumps(item),
        datetime.now(timezone.utc).isoformat(),
    )
    for item in records
]

with sqlite3.connect("items.db") as conn:
    conn.execute("""
        CREATE TABLE IF NOT EXISTS items (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            price REAL,
            updated_at TEXT,
            raw_json TEXT NOT NULL,
            fetched_at TEXT NOT NULL
        )
    """)

    conn.executemany("""
        INSERT INTO items
            (id, name, price, updated_at, raw_json, fetched_at)
        VALUES (?, ?, ?, ?, ?, ?)
    """, rows)

SQLite uses ? placeholders in this example. Python’s sqlite3 documentation describes placeholder-based parameter binding and executemany() for repeatedly executing parameterized data-modification statements.

The context manager commits a successful transaction and rolls back when an exception escapes the block. Avoid committing every row unless there is a specific reason: per-row commits are slower and can leave a partially loaded batch. Do not put millions of rows into one unbounded transaction either; process controlled batches.

Why are parameterized queries essential?

Parameterized queries keep SQL code separate from values, protecting against injection and avoiding errors caused by quotes, nulls, types, and encoding.

Unsafe SQL concatenation looks like this:

sql = f"""
    INSERT INTO items (id, name)
    VALUES ({item["id"]}, '{item["name"]}')
"""

The unsafe version can be exploited when a value contains SQL syntax, and ordinary names containing apostrophes can break the statement. The safe form passes values as a separate argument:

cur.execute(
    "INSERT INTO items (id, name) VALUES (%s, %s)",
    (item["id"], item["name"]),
)

Psycopg uses %s placeholders for values. The placeholders are not Python string-formatting instructions. Psycopg performs the necessary conversion and quoting when values are passed separately. See the Psycopg parameter-binding documentation.

Rank #3

SSK Portable SSD 500GB External Solid State Hard Drive USB C Up to 1050MB/s

• Capacity Display Variance: 500GB external ssd often appears as around 465GB on Windows. MacOS can show full 500 GB capacity. This is binary calculation difference and doesn’t affect SSD hard drive actual physical storage
• 1050 MB/s Speed: Instantly access to your files with blazing-fast 10Gbps external SSD read up to 1050MB/s and write up to 1000MB/s. LED Light indicates USB SSD instant activity
• Data Security: Solid state drives S.M.A.R.T. health diagnostics​ and adaptive TRIM optimizing data block management ensures consistent write speeds and extends the longevity of the portable SSD
• USB-C & USB-A Cable: Both cables featuring rapid USB 3.2 Gen2, this USB SSD effortlessly bridges devices, enabling seamless cross-platform file transfers and backup between computers, smartphones, tablets and iPhone
• Always Fast: No slowdowns for large file transfers. With SLC caching (25% of current available capacity allocated as high-speed cache), this external SSD delivers steady 10Gbps for transfers within the cache capacity

$89.99

Buy on Amazon

Bound parameters normally cannot substitute table names, column names, or SQL keywords. If identifiers must be dynamic, use the driver’s identifier-composition facilities or a strict allowlist. Never treat an uncontrolled table or column name as an ordinary value parameter.

How do you make repeated imports safe with upserts?

Use a stable unique key and an insert-or-update policy so a retry or scheduled rerun does not create duplicate current-state rows.

To ignore an existing SQLite record:

INSERT OR IGNORE INTO items
    (id, name, price, updated_at, raw_json, fetched_at)
VALUES (?, ?, ?, ?, ?, ?);

To update an existing SQLite record:

INSERT INTO items
    (id, name, price, updated_at, raw_json, fetched_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
    name = excluded.name,
    price = excluded.price,
    updated_at = excluded.updated_at,
    raw_json = excluded.raw_json,
    fetched_at = excluded.fetched_at;

The PostgreSQL equivalent uses EXCLUDED:

INSERT INTO items
    (id, name, price, updated_at, raw_json, fetched_at)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (id) DO UPDATE SET
    name = EXCLUDED.name,
    price = EXCLUDED.price,
    updated_at = EXCLUDED.updated_at,
    raw_json = EXCLUDED.raw_json,
    fetched_at = EXCLUDED.fetched_at;

If the API deletes records, an upsert alone will not remove the local rows. Choose a deletion policy such as hard deletion after a complete reconciliation, a soft-delete flag, or retaining historical records.

How do you fetch every API page?

Implement the pagination model documented by the provider; never assume the first response contains all records.

A page-number implementation might look like this:

def fetch_all(url, headers):
    records = []
    page = 1
    page_size = 100

    while True:
        response = requests.get(
            url,
            params={"page": page, "limit": page_size},
            headers=headers,
            timeout=30,
        )
        response.raise_for_status()
        payload = response.json()

        page_records = payload.get("data", [])
        if not page_records:
            break

        records.extend(page_records)

        if len(page_records) < page_size:
            break

        page += 1

    return records

The short-page stopping rule is valid only when the API guarantees that a page shorter than the requested size is the final page. Prefer an explicit next URL, cursor, or continuation token when the API provides one.

Common models include page number and size, offset and limit, cursor tokens, provider-returned next URLs, and time-window or ID-based ranges. Cursor pagination is often safer for changing datasets because offset pages can skip or repeat records while rows are added or removed. Shopify's REST Admin API documentation is one provider-specific example: its REST endpoints use cursor-based pagination and document a 429 Too Many Requests response when the rate limit is exceeded. That behavior is not universal.

Supabase's Python data API documentation also states a default maximum of 1,000 returned rows and documents range-based pagination. A hosted API's result limit is therefore another reason to inspect provider documentation rather than assuming one request is sufficient.

How should you handle retries and API rate limits?

Retry temporary failures such as timeouts, connection errors, 408, 429, and selected 5xx responses, but surface permanent errors such as invalid credentials or a missing endpoint.

import random
import time
import requests

RETRYABLE_STATUS_CODES = {408, 429, 500, 502, 503, 504}

def get_with_retries(url, *, headers=None, params=None, attempts=5):
    for attempt in range(attempts):
        try:
            response = requests.get(
                url,
                headers=headers,
                params=params,
                timeout=30,
            )
        except requests.RequestException:
            if attempt == attempts - 1:
                raise
            time.sleep(min(60, 2 ** attempt + random.random()))
            continue

        if response.status_code not in RETRYABLE_STATUS_CODES:
            response.raise_for_status()
            return response

        if response.status_code == 429 and response.headers.get("Retry-After"):
            delay = float(response.headers["Retry-After"])
        else:
            delay = min(60, 2 ** attempt + random.random())

        if attempt == attempts - 1:
            response.raise_for_status()

        time.sleep(delay)

    raise RuntimeError("Request failed")

Honor Retry-After when provided and add jitter to exponential backoff. Do not retry every error. A 401 may require new credentials, a 403 may indicate insufficient permission, and a 404 may indicate a wrong endpoint. Retrying mutating requests such as POST or PATCH can create duplicates unless the API supports idempotency keys or the operation is otherwise idempotent.

GitHub's REST API guidance recommends authenticated requests, conditional requests where appropriate, avoiding unnecessary concurrency, and honoring Retry-After rather than immediately retrying.

How should database transactions be managed?

Fetch and validate a batch first, write the complete valid batch in a transaction, commit only after success, and roll back when a required database operation fails.

1. Fetch one page or controlled batch.
2. Validate its JSON structure and required fields.
3. Transform records into database values.
4. Insert or upsert all rows in one transaction.
5. Commit after every required operation succeeds.
6. Save a pagination cursor or synchronization checkpoint only after the commit succeeds.

Psycopg starts a transaction for database operations by default. After a database error, the transaction can remain in a failed state until it is rolled back. The Psycopg transaction documentation explains this behavior and the available transaction controls.

try:
    with conn.transaction():
        cur.executemany(insert_sql, rows)
except Exception:
    logger.exception("Database batch failed")
    raise

The exact transaction context-manager behavior depends on the installed driver version, so verify it against the driver documentation. For very large imports, use bounded batches rather than one transaction that grows without limit.

Rank #4

Sale

Seagate 2TB Portable Hard Drive | USB 3.0 (STGX2000400)

• Easily store and access 2TB to content on the go with the Seagate Portable Drive, a USB external hard drive
• Designed to work with Windows or Mac computers, this external hard drive makes backup a snap just drag and drop
• To get set up, connect the portable hard drive to a computer for automatic recognition no software required
• This USB drive provides plug and play simplicity with the included 18 inch USB 3.0 cable
• The available storage capacity may vary.

$119.99

Buy on Amazon

What is the complete Python API-to-SQL example?

The following script fetches fictional page-based API data, validates a data array, stores the records in SQLite, and upserts repeated imports.

import json
import logging
import os
import sqlite3
import time
from datetime import datetime, timezone

import requests

logging.basicConfig(level=logging.INFO)

API_URL = "https://api.example.com/v1/items"
DB_PATH = "items.db"
API_TOKEN = os.environ.get("API_TOKEN")


def utc_now():
    return datetime.now(timezone.utc).isoformat()


def fetch_page(page: int, page_size: int = 100):
    headers = {"Accept": "application/json"}

    if API_TOKEN:
        headers["Authorization"] = f"Bearer {API_TOKEN}"

    response = requests.get(
        API_URL,
        params={"page": page, "limit": page_size},
        headers=headers,
        timeout=30,
    )

    if response.status_code == 429:
        retry_after = response.headers.get("Retry-After", "5")
        time.sleep(float(retry_after))
        return fetch_page(page, page_size)

    response.raise_for_status()
    payload = response.json()

    if not isinstance(payload, dict):
        raise ValueError("Expected a JSON object")

    records = payload.get("data")
    if not isinstance(records, list):
        raise ValueError("Expected payload['data'] to be a list")

    return records


def save_records(records):
    rows = []

    for item in records:
        if "id" not in item:
            logging.warning("Skipping record without id: %r", item)
            continue

        rows.append((
            item["id"],
            item.get("name"),
            item.get("price"),
            item.get("updated_at"),
            json.dumps(item),
            utc_now(),
        ))

    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS items (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                price REAL,
                updated_at TEXT,
                raw_json TEXT NOT NULL,
                fetched_at TEXT NOT NULL
            )
        """)

        conn.executemany("""
            INSERT INTO items
                (id, name, price, updated_at, raw_json, fetched_at)
            VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                price = excluded.price,
                updated_at = excluded.updated_at,
                raw_json = excluded.raw_json,
                fetched_at = excluded.fetched_at
        """, rows)

    logging.info("Saved %d records", len(rows))


def main():
    page = 1

    while True:
        records = fetch_page(page)

        if not records:
            break

        save_records(records)

        if len(records) < 100:
            break

        page += 1


if __name__ == "__main__":
    main()

Assumptions you must change

• https://api.example.com/v1/items is fictional.
• The response is assumed to contain a data array.
• Pagination is assumed to use page and limit.
• Each record is assumed to contain a stable id.
• The example assumes optional bearer authentication.
• The example assumes a short page means the final page.
• The example assumes repeated retrieval is safe and that the API's records represent current state.

Replace each assumption with the behavior documented by the real API. The recursive 429 branch in this compact example should also be replaced with a bounded retry loop in a production importer so repeated rate limiting cannot create unbounded recursion.

How do you adapt the importer to PostgreSQL?

Use Psycopg 3 when the database is a shared PostgreSQL service or the application needs PostgreSQL-specific features.

import json
from datetime import datetime, timezone
import psycopg

rows = [
    (
        item["id"],
        item.get("name"),
        item.get("price"),
        item.get("updated_at"),
        json.dumps(item),
        datetime.now(timezone.utc),
    )
    for item in records
]

with psycopg.connect(
    "dbname=app user=app_user password=secret host=localhost"
) as conn:
    with conn.cursor() as cur:
        cur.executemany("""
            INSERT INTO items
                (id, name, price, updated_at, raw_json, fetched_at)
            VALUES (%s, %s, %s, %s, %s, %s)
        """, rows)

Keep the connection string in an environment variable or secret manager rather than placing a real password in source code. Psycopg supports parameter binding, transactions, batch operations, connection pooling, asynchronous access, and PostgreSQL-specific features. Its basic usage documentation covers connections, cursors, execution, and fetching.

SQLite is not automatically unsuitable for production; suitability depends on concurrency, data volume, deployment, and recovery requirements. PostgreSQL is generally the stronger choice when several services write concurrently or when centralized backups, permissions, and operational tooling matter.

Should you use a direct database driver or SQLAlchemy?

Use a direct driver when the goal is a small importer or clear SQL instruction, and use SQLAlchemy when the application benefits from abstraction across database engines or from application-level database tooling.

SQLAlchemy 2.x includes PostgreSQL dialect support and an insertmanyvalues mechanism for qualifying multi-row inserts, but behavior depends on the statement, driver, and dialect. The SQLAlchemy PostgreSQL documentation should be consulted for the chosen version and insert pattern. SQLAlchemy does not automatically solve API pagination, response validation, duplicate policy, or schema design.

How do you query the stored API data?

Query the table with ordinary SQL after the import succeeds.

with sqlite3.connect("items.db") as conn:
    rows = conn.execute("""
        SELECT id, name, price, updated_at
        FROM items
        WHERE price IS NOT NULL
        ORDER BY updated_at DESC
    """).fetchall()

for row in rows:
    print(row)

Use fetchone() for one result, fetchmany() for bounded batches, or iterate over a cursor when the result set may be large. Avoid unbounded fetchall() for large tables because it loads every returned row into memory. Add indexes to columns frequently used in filters, joins, or ordering.

The same query can be executed through a Psycopg cursor:

with psycopg.connect(DATABASE_URL) as conn:
    with conn.cursor() as cur:
        cur.execute("""
            SELECT id, name, price, updated_at
            FROM items
            WHERE price IS NOT NULL
            ORDER BY updated_at DESC
        """)
        for row in cur:
            print(row)

How do you build incremental synchronization?

Use incremental synchronization when a full reload is too expensive or the API provides a reliable change filter, cursor, webhook, or event endpoint.

Possible synchronization markers include an updated_since timestamp, provider-issued cursor, highest known ID, webhook event, or time window. Store the marker in SQL:

CREATE TABLE IF NOT EXISTS sync_state (
    source_name TEXT PRIMARY KEY,
    cursor TEXT,
    last_successful_run TEXT,
    updated_at TEXT NOT NULL
);

Update the saved cursor only after the corresponding records have been committed. If a process crashes before the commit, rerunning the same cursor is safer than advancing the checkpoint prematurely.

Best Value

Sale

Samsung T7 Portable SSD 1TB Titan Gray, USB 3.2 Gen 2, Up to 1,050MB/s

• MADE FOR THE MAKERS: Create; Explore; Store; The T7 Portable SSD delivers fast speeds and durable features to back up any endeavor; Build your video editing empire, file your photographs or back up your blogs all in an instant
• SHARE IDEAS IN A FLASH: Don’t waste a second waiting and spend more time doing; The T7 is embedded with PCIe NVMe technology that brings fast read and write speeds up to 1,050/1,000 MB/s¹, making it almost twice as fast as the T5
• ALWAYS MAKE THE SAVE: Compact design with massive capacity; With capacities up to 4TB, save exactly what you need to your drive – from large working files to game data and everything in between
• ADAPTS TO EVERY NEED: Whether using a PC or mobile phone, count on the T7 for extensive compatibility²; It’s a true team player when it comes to heavy-duty application usage or file-saving
• HI RESOLUTION VIDEO RECORDING: Record Ultra High Resolution (4K 60fs) videos directly onto the T7 Portable SSD with your favorite camera or mobile devices; Supports iPhone 15 Pro Res 4K at 60fps video and more³

$229.99

Buy on Amazon

For a current-state table, use stable source IDs and upserts. For an append-only history, store every version or observation with its source timestamp and ingestion timestamp. For large or sensitive pipelines, a raw landing table followed by a validated merge can make recovery and reprocessing easier.

What failure modes should the importer handle?

Reliable ingestion separates transport failures, payload failures, database failures, and operational failures.

Also account for valid HTTP responses containing an error object, empty response bodies, duplicate IDs, dates in multiple formats, numeric strings, inconsistent pagination metadata, and API schema changes.

What should a production API-to-SQL checklist include?

• Use HTTPS and keep credentials in environment variables or a secret manager.
• Set connect and read timeouts appropriate to the API and job.
• Call raise_for_status() and validate the JSON shape.
• Record the API provider, endpoint, page or cursor, and request identifier in structured logs.
• Redact authorization headers, passwords, API keys, personal data, and confidential payload fields.
• Use stable unique keys, parameterized SQL, and an explicit insert, update, or upsert policy.
• Wrap controlled batches in transactions and advance synchronization checkpoints only after commit.
• Handle pagination according to the provider's contract.
• Retry only transient failures, honor Retry-After, and add jitter.
• Prevent overlapping importer jobs with a lock, scheduler constraint, or database advisory mechanism.
• Decide how source deletions, historical versions, raw JSON, and retention will be handled.
• Monitor fetched, inserted, updated, skipped, failed, and retried records.
• Use schema migrations rather than silently changing a production table when the API evolves.
• Back up the database and test restoring it.

Which database should you choose?

Choose SQLite for a local experiment or small standalone script, PostgreSQL for a shared application, a managed PostgreSQL service when backups and operations should be delegated, and an ETL platform when connector maintenance and monitoring across many sources justify the extra complexity.

Supabase combines hosted PostgreSQL with features such as authentication, storage, a dashboard, and generated data APIs; its Python data API documentation documents range-based pagination and a default maximum of 1,000 returned rows. Supabase adds platform and deployment complexity that a local SQLite importer does not need.

Managed PostgreSQL options include Amazon RDS for PostgreSQL, Google Cloud SQL for PostgreSQL, and Azure Database for PostgreSQL. These options are most useful when an organization already uses the corresponding cloud, needs centralized operations, or requires managed backups and networking. Their costs depend on region, compute, storage, backups, and data transfer, so current pricing must be checked on the provider's official pricing tools.

Neon is another hosted PostgreSQL option for development, preview environments, and applications that benefit from separated compute and storage. Review the current Neon pricing page and operational features before choosing it. Airbyte, Fivetran, and Meltano are relevant when scheduled connectors and multi-source synchronization matter more than writing a custom importer.

Frequently Asked Questions

Can I fetch API data directly into a SQL database?

Yes. A Python script can request API data, validate the response, transform the records, and write them to SQLite, PostgreSQL, MySQL, or SQL Server. SQL itself generally does not replace the API client, authentication, pagination, and retry logic required by the provider.

How do I save nested JSON from an API in SQL?

Store frequently queried nested fields in relational columns, move repeated child objects into related tables, and optionally preserve the original record in a JSON or JSONB column. The best design depends on whether the nested data must be filtered, joined, constrained, or retained for reprocessing.

How do I update existing SQL rows when importing API data?

Give each source record a stable unique key and use an upsert such as SQLite's ON CONFLICT or PostgreSQL's ON CONFLICT ... DO UPDATE. Use append-only storage instead when every historical version or observation must be retained.

What should I do when an API returns HTTP 429?

Treat HTTP 429 as rate limiting: honor the provider's Retry-After header when present, wait with bounded exponential backoff and jitter, reduce concurrency, and stop after a finite number of attempts. Rate-limit rules vary by provider and endpoint.

Is SQLite suitable for an API importer in production?

SQLite can be suitable for some small, single-machine production jobs. PostgreSQL is usually a better fit when several services write concurrently, users need shared access, or centralized backups, permissions, and operational controls are important.

The Bottom Line

The dependable API-to-SQL pattern is simple in outline but disciplined in implementation: request over HTTPS with authentication and timeouts, validate the actual JSON shape, map fields deliberately, use parameterized SQL, commit controlled batches, and make reruns idempotent with unique keys or upserts. Start with SQLite to learn the workflow, then move to PostgreSQL or a managed ingestion platform when concurrency, scale, monitoring, or operational requirements demand it.

Quick Recap

Bestseller No. 1

Sandisk 2TB Extreme Portable SSD, Up to 1050MB/s, USB-C, USB 3.2 Gen 2, IP65 Water and Dust Resistance, Updated Firmware, External Solid State Drive, SDSSDE61-2T00-G25

$289.99

Bestseller No. 2

Sandisk 1TB Portable SSD, Up to 800MB/s Read Speeds, Black (Old Model)

From Sandisk, a brand professional photographers trust to take on assignments.

$188.90

Bestseller No. 3

SSK Portable SSD 500GB External Solid State Hard Drive USB C Up to 1050MB/s

$89.99

SaleBestseller No. 4

Seagate 2TB Portable Hard Drive | USB 3.0 (STGX2000400)

This USB drive provides plug and play simplicity with the included 18 inch USB 3.0 cable; The available storage capacity may vary.

$119.99

SaleBestseller No. 5

Samsung T7 Portable SSD 1TB Titan Gray, USB 3.2 Gen 2, Up to 1,050MB/s

$229.99

Product prices and availability are accurate as of the date/time indicated and are subject to change. Any price and availability information displayed on Amazon at the time of purchase will apply.