Blog / Data engineering

Web Data to a Warehouse: Loading API JSON into Postgres

19 Aug 2026Perceptlayer7 min read

To load API JSON into Postgres, store each response untouched in a jsonb column, expose the fields you need through a view, and upsert into a typed table keyed on the record's own ID. This tutorial does that with the AutoScout24 search endpoint: raw pages in, a flat listings table out, and re-runs that never create duplicates.

What does the pipeline look like?

The pipeline has three layers: a raw table that holds each API response exactly as received, a view that unpacks the JSON into columns, and a typed table that holds one current row per listing. A small Python script fetches pages, writes the raw rows, then runs one SQL statement to move new rows into the typed table. Each layer can be rebuilt from the one before it.

You need Python 3.10 or later, the oldest version the current psycopg 3 release supports (psycopg 3 installation docs, 2026), a Postgres database, and an API key from the free tier: 1,000 free credits every month, no credit card. Install the two libraries with pip install requests "psycopg[binary]". The key goes in the X-API-Key header, as described in the authentication docs.

Why land the raw JSON in a JSONB column first?

Landing the raw response first separates "what the API said" from "what I decided to keep". Postgres stores jsonb in a decomposed binary form that supports path operators and indexing (PostgreSQL docs: JSON types), so you can query the payload directly while you are still deciding on columns. If you later add co2 or consumption to the typed table, you re-run the view over rows you already hold.

The cost is storage: every page keeps its image URLs and fields you may never read. If you only ever need five fields and never replay, flatten in Python and skip the raw table. For anything that feeds reports, the raw copy is cheap insurance.

Step 1: How do you create the landing table?

Create one table for raw pages, one for current listings and one for price history. The raw table stores the request parameters next to the payload, so any row explains itself. A partial index on unprocessed rows keeps the "what is new" lookup fast however large the table grows. listing_id is text rather than uuid because the endpoint documents it only as a string.

-- schema.sql
CREATE TABLE raw_autoscout24_search (
  id           bigserial PRIMARY KEY,
  fetched_at   timestamptz NOT NULL DEFAULT now(),
  market       text NOT NULL,
  params       jsonb NOT NULL,
  payload      jsonb NOT NULL,
  processed_at timestamptz
);

CREATE INDEX raw_autoscout24_search_unprocessed
  ON raw_autoscout24_search (id) WHERE processed_at IS NULL;

CREATE TABLE listings (
  listing_id        text PRIMARY KEY,
  market            text NOT NULL,
  title             text,
  make              text,
  model             text,
  price_amount      integer,
  price_currency    text,
  mileage_km        integer,
  registration_year integer,
  power_kw          integer,
  fuel              text,
  transmission      text,
  city              text,
  seller_type       text,
  dealer_slug       text,
  first_seen_at     timestamptz NOT NULL,
  last_seen_at      timestamptz NOT NULL,
  last_raw_id       bigint NOT NULL
);

CREATE TABLE listing_prices (
  listing_id     text NOT NULL,
  raw_id         bigint NOT NULL REFERENCES raw_autoscout24_search (id),
  observed_at    timestamptz NOT NULL,
  price_amount   integer,
  price_currency text,
  PRIMARY KEY (listing_id, raw_id)
);

Step 2: How do you fetch pages and land them?

Loop over GET /v1/autoscout24/search, follow pagination.next_page until has_more is false, and commit each page as soon as it arrives. Committing per page matters because every successful page was already billed: if page 7 fails, pages 1 to 6 stay in your table. The loop below only stores responses that returned HTTP 200, and it logs credit headers, never bodies.

# load.py
import os
from pathlib import Path

import psycopg
import requests
from psycopg.types.json import Jsonb

BASE = "https://api.perceptlayer.com"
HEADERS = {"X-API-Key": os.environ["API_KEY"]}
PARAMS = {
    "market": "de", "make_id": "13", "model_id": "1639",
    "price_min": 5000, "price_max": 20000, "sort": "price", "limit": 20,
}


def fetch_pages(params, max_pages=50):
    page = 1
    while page <= max_pages:
        resp = requests.get(
            f"{BASE}/v1/autoscout24/search",
            headers=HEADERS,
            params={**params, "page": page},
            timeout=30,
        )
        resp.raise_for_status()
        print(
            f"page {page}: used {resp.headers.get('X-Credits-Used')}, "
            f"remaining {resp.headers.get('X-Credits-Remaining')}"
        )
        body = resp.json()
        yield page, body
        pagination = body["data"]["pagination"]
        if not pagination["has_more"]:
            break
        page = pagination["next_page"]


def land(conn, params, page, body):
    conn.execute(
        "INSERT INTO raw_autoscout24_search (market, params, payload) "
        "VALUES (%s, %s, %s)",
        (params["market"], Jsonb({**params, "page": page}), Jsonb(body)),
    )
    conn.commit()

max_pages is a guard so a bad cursor can't loop forever. For pacing, retries and timeouts, see safe page loops in Python, retrying without paying twice and the error codes.

Step 3: How do you flatten the JSON into a view?

Use jsonb_array_elements in a lateral join to turn data.listings into one row per listing, and #>> to pull nested values out as text before casting. The paths below match the recorded response: price.amount, mileage.value_km, first_registration.year, power.kw, location.city and seller.type (PostgreSQL docs: JSON functions).

-- view.sql
CREATE VIEW autoscout24_search_listings AS
SELECT
  r.id                                          AS raw_id,
  r.fetched_at,
  r.market,
  l ->> 'listing_id'                            AS listing_id,
  l ->> 'title'                                 AS title,
  l ->> 'make'                                  AS make,
  l ->> 'model'                                 AS model,
  (l #>> '{price,amount}')::integer             AS price_amount,
  l #>> '{price,currency}'                      AS price_currency,
  (l #>> '{mileage,value_km}')::integer         AS mileage_km,
  (l #>> '{first_registration,year}')::integer  AS registration_year,
  (l #>> '{power,kw}')::integer                 AS power_kw,
  l ->> 'fuel'                                  AS fuel,
  l ->> 'transmission'                          AS transmission,
  l #>> '{location,city}'                       AS city,
  l #>> '{seller,type}'                         AS seller_type,
  l #>> '{seller,dealer_slug}'                  AS dealer_slug
FROM raw_autoscout24_search AS r
CROSS JOIN LATERAL jsonb_array_elements(r.payload #> '{data,listings}') AS l;

Applied to the two listings in the recorded response, the view gives:

The private seller has no dealer_slug, because its seller object holds only id and type. Path operators return NULL for missing keys instead of raising, which is why this works for optional fields. Text values such as Schaltgetriebe come back as found, so map them yourself if you mix markets.

Step 4: How do you upsert on listing ID?

Key the typed table on listing_id and use INSERT … ON CONFLICT DO UPDATE (PostgreSQL docs: INSERT). Two details make it safe. First, DISTINCT ON (listing_id) keeps one row per listing per batch, because Postgres raises an error if one statement tries to update the same row twice. Second, the WHERE guard stops an older batch from overwriting newer values.

-- upsert.sql
WITH batch AS (
  SELECT v.*
  FROM autoscout24_search_listings AS v
  JOIN raw_autoscout24_search AS r ON r.id = v.raw_id
  WHERE r.processed_at IS NULL AND r.id <= %(max_id)s
),
prices AS (
  INSERT INTO listing_prices
    (listing_id, raw_id, observed_at, price_amount, price_currency)
  SELECT listing_id, raw_id, fetched_at, price_amount, price_currency
  FROM batch
  ON CONFLICT (listing_id, raw_id) DO NOTHING
),
latest AS (
  SELECT DISTINCT ON (listing_id)
         batch.*,
         min(fetched_at) OVER (PARTITION BY listing_id) AS first_fetched
  FROM batch
  ORDER BY listing_id, fetched_at DESC, raw_id DESC
)
INSERT INTO listings (
  listing_id, market, title, make, model, price_amount, price_currency,
  mileage_km, registration_year, power_kw, fuel, transmission, city,
  seller_type, dealer_slug, first_seen_at, last_seen_at, last_raw_id
)
SELECT
  listing_id, market, title, make, model, price_amount, price_currency,
  mileage_km, registration_year, power_kw, fuel, transmission, city,
  seller_type, dealer_slug, first_fetched, fetched_at, raw_id
FROM latest
ON CONFLICT (listing_id) DO UPDATE SET
  title             = EXCLUDED.title,
  price_amount      = EXCLUDED.price_amount,
  price_currency    = EXCLUDED.price_currency,
  mileage_km        = EXCLUDED.mileage_km,
  registration_year = EXCLUDED.registration_year,
  power_kw          = EXCLUDED.power_kw,
  fuel              = EXCLUDED.fuel,
  transmission      = EXCLUDED.transmission,
  city              = EXCLUDED.city,
  seller_type       = EXCLUDED.seller_type,
  dealer_slug       = EXCLUDED.dealer_slug,
  last_seen_at      = EXCLUDED.last_seen_at,
  last_raw_id       = EXCLUDED.last_raw_id
WHERE listings.last_raw_id <= EXCLUDED.last_raw_id;

The <= lets you replay a raw row after fixing the view, while still refusing older data. first_seen_at is left out of the update list, so it keeps the value from the first insert. The script below runs this statement and marks the rows processed in the same transaction. Add it to load.py:

UPSERT_SQL = Path("upsert.sql").read_text()


def transform(conn):
    max_id = conn.execute(
        "SELECT max(id) FROM raw_autoscout24_search WHERE processed_at IS NULL"
    ).fetchone()[0]
    if max_id is None:
        return
    conn.execute(UPSERT_SQL, {"max_id": max_id})
    conn.execute(
        "UPDATE raw_autoscout24_search SET processed_at = now() "
        "WHERE processed_at IS NULL AND id <= %(max_id)s",
        {"max_id": max_id},
    )
    conn.commit()


def main():
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        for page, body in fetch_pages(PARAMS):
            land(conn, PARAMS, page, body)
        transform(conn)


if __name__ == "__main__":
    main()

Capturing max_id first means pages that land while the transform runs wait for the next run instead of being marked processed unseen.

Step 5: How do you run it and check the result?

Create the objects once, then run the script as often as you like:

psql "$DATABASE_URL" -f schema.sql -f view.sql
API_KEY=sk_live_... DATABASE_URL=postgresql://localhost/cars python load.py

Run it twice and compare SELECT count(*) FROM listings. The second run adds raw rows and price observations, but listings only grows by listings that are new. Then query it like any table:

SELECT registration_year,
       count(*) AS listings,
       percentile_cont(0.5) WITHIN GROUP (ORDER BY price_amount) AS median_price
FROM listings
WHERE price_currency = 'EUR'
GROUP BY registration_year
ORDER BY registration_year;

To see price movement, compare each observation to the previous one for the same listing:

SELECT *
FROM (
  SELECT listing_id, observed_at, price_amount,
         price_amount - lag(price_amount) OVER (
           PARTITION BY listing_id ORDER BY observed_at
         ) AS change
  FROM listing_prices
) AS p
WHERE change <> 0;

For which fields to model next, see used car listing data for price analysis.

What does a refresh cost in credits?

A successful call costs 1 credit, so a refresh costs one credit per page. The recorded response reports total: 193, limit: 20 and total_pages: 10, so that search needs 10 pages: 10 pages × 1 credit = 10 credits. A daily refresh for 30 days is 30 × 10 = 300 credits, inside the 1,000 free credits each month. Failed calls cost nothing and return X-Credits-Used: 0.

The total changes as listings come and go, so read pagination.total_pages from your own run. Because the raw table keeps every page, reprocessing never spends credits. See credits for how billing works, rate limits for pacing, and pricing for what credits cost.

Where does this design break?

It breaks in three places. Pages are fetched one at a time, so if listings change between requests, one can appear on two pages or be missed. The upsert absorbs duplicates, but a missed listing only shows up in the next run. A listing that disappears is never deleted, so use last_seen_at to find the ones that stopped appearing. Finally, fetching a page and storing it are not one transaction. A timed-out request may have succeeded and been billed, which the retry guide covers.

Raw rows also grow without limit. Once the typed tables are trusted, archive or drop old payload values. If your warehouse is BigQuery or Snowflake, the same layers apply with their own JSON types and MERGE in place of ON CONFLICT.

Next step: get a free API key, try the call in the playground on the AutoScout24 Search page, then run load.py. For the endpoint's fields and filters, read AutoScout24 API: Search European Car Listings as JSON, and find more pipeline guides in the data engineering category.

postgresjsonbdata pipelineupsertautoscout24etl