---
title: "Web Data to a Warehouse: Loading API JSON into Postgres"
description: "Land AutoScout24 search JSON in a Postgres JSONB table, flatten it with a view, then upsert on listing_id. Code and credit cost per refresh."
url: https://perceptlayer.com/blog/load-web-data-json-into-postgres
---

[Blog](https://perceptlayer.com/blog) / [Data engineering](https://perceptlayer.com/blog/category/data-engineering)

# Web Data to a Warehouse: Loading API JSON into Postgres

19 Aug 2026 Perceptlayer 7 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.

## Key takeaways

-   Land every page as one `jsonb` row first. A parsing bug or a new field then costs a replay from your own table, not another API call.
-   A view over `jsonb_array_elements(payload #> '{data,listings}')` turns one response into one row per listing.
-   `INSERT … ON CONFLICT (listing_id) DO UPDATE` makes the load idempotent; `SELECT DISTINCT ON` stops one batch from touching a row twice.
-   In the recorded search, 193 results at 20 per page is 10 pages, so a full refresh is 10 credits at 1 credit per call.

## 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.

| Layer | Object | Key | Purpose |
| --- | --- | --- | --- |
| Landing | `raw_autoscout24_search` | `id` (one per page fetched) | Immutable record of what the API returned |
| Flattening | `autoscout24_search_listings` (view) | none | One row per listing per page, typed columns |
| Current state | `listings` | `listing_id` | Latest values for each listing |
| History | `listing_prices` | `listing_id`, `raw_id` | Price at each fetch |

You need Python 3.10 or later, the oldest version the current psycopg 3 release supports ( [psycopg 3 installation docs](https://www.psycopg.org/psycopg3/docs/basic/install.html), 2026), a Postgres database, and an API key from [the free tier](https://perceptlayer.com/signup): 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](https://docs.perceptlayer.com/authentication).

## 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](https://www.postgresql.org/docs/current/datatype-json.html)), 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.

```sql
-- 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.

```python
# 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](https://perceptlayer.com/blog/paginate-web-data-api-python), [retrying without paying twice](https://perceptlayer.com/blog/retry-web-api-calls-without-paying-twice) and the [error codes](https://docs.perceptlayer.com/errors).

## 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](https://www.postgresql.org/docs/current/functions-json.html)).

```sql
-- 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:

| listing\_id | price\_amount | mileage\_km | registration\_year | power\_kw | city | seller\_type | dealer\_slug |
| --- | --- | --- | --- | --- | --- | --- | --- |
| `f81b29b2-…` | 5700 | 215000 | 2013 | 85 | Mainz-Kastel | Dealer | `rheinkastelautomobile` |
| `2a852a90-…` | 5800 | 177000 | 2012 | 85 | Berlin | PrivateSeller | NULL |

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](https://www.postgresql.org/docs/current/sql-insert.html)). 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.

```sql
-- 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`:

```python
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:

```bash
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:

```sql
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:

```sql
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](https://perceptlayer.com/blog/used-car-listing-data-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](https://docs.perceptlayer.com/credits) for how billing works, [rate limits](https://docs.perceptlayer.com/rate-limits) for pacing, and [pricing](https://perceptlayer.com/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](https://perceptlayer.com/blog/retry-web-api-calls-without-paying-twice) 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](https://perceptlayer.com/signup), try the call in the playground on the [AutoScout24 Search page](https://perceptlayer.com/apis/autoscout24/search), then run `load.py`. For the endpoint's fields and filters, read [AutoScout24 API: Search European Car Listings as JSON](https://perceptlayer.com/blog/autoscout24-api-car-listings-json), and find more pipeline guides in the [data engineering category](https://perceptlayer.com/blog/category/data-engineering).

## APIs in this post

-   [AutoScout24 **1 credit per call** · 6 endpoints](https://perceptlayer.com/apis/autoscout24)

## Frequently asked questions

### Should I store the raw JSON or flatten it before inserting?

Store the raw JSON when the data feeds anything you will need to rebuild. A parsing bug or a new column then costs a replay from your own table, not new API calls. Flatten before inserting only when you need a handful of fields and will never replay. A view over `jsonb` costs nothing to change later.

### How do I avoid duplicate rows when I re-run the load?

Key the typed table on the record's own ID, here `listing_id`, and write with `INSERT … ON CONFLICT DO UPDATE`. Re-running then updates existing rows instead of adding new ones. Deduplicate inside each batch with `DISTINCT ON`, because one statement can't update the same row twice. The history table's `(listing_id, raw_id)` key keeps replays from duplicating observations.

### What happens if a page request fails halfway through a run?

Pages already committed stay in the raw table, and failed calls cost nothing. Run the script again: it fetches from page 1 and lands a second copy of the earlier pages, which the upsert handles without duplicates. If you want to avoid paying twice for those pages, store the last completed page and resume from it. The [error handling guide](https://perceptlayer.com/blog/handle-web-data-api-errors-python) covers which errors are worth retrying.

### Can I use the same pattern for other endpoints?

Yes. Any endpoint returns JSON, so the layers stay the same: one raw table per endpoint family, a view with that response's paths, and an upsert keyed on the ID the response provides. Check the [API reference for the endpoint](https://perceptlayer.com/apis) for the field names, and take the response shape only from a real call, not from guesswork.

### Is the data in my table current?

It is as found at the time of each call. The API doesn't promise freshness beyond that, so `fetched_at` and `last_seen_at` in your tables are the record of when you saw each listing. If you refresh often, [caching responses](https://perceptlayer.com/blog/cache-web-data-api-responses) can avoid repeating identical calls.

postgres jsonb data pipeline upsert autoscout24 etl

## More from the blog

1.  [Data engineering](https://perceptlayer.com/blog/category/data-engineering) 20 Oct 2026 Perceptlayer 6 min read
    
    ## [Detect Schema Changes in API JSON Responses](https://perceptlayer.com/blog/web-data-api-json-schema-changes)
    
    Infer a path-and-type schema from a recorded Google Search response (71 paths), diff each new response against it and alert on removed or retyped fields.
    
2.  [Data engineering](https://perceptlayer.com/blog/category/data-engineering) 19 Oct 2026 Perceptlayer 7 min read
    
    ## [Concurrent API Requests in Python: Asyncio Under Rate Limits](https://perceptlayer.com/blog/concurrent-api-requests-python-rate-limit)
    
    A Python asyncio fetcher that paces calls to your per-second limit, honours Retry-After and prices a 20,000-call batch at 20,000 credits.
    
3.  [Data engineering](https://perceptlayer.com/blog/category/data-engineering) 12 Oct 2026 Perceptlayer 7 min read
    
    ## [Incremental API Data Pulls: Fetch Only New Records](https://perceptlayer.com/blog/incremental-api-data-pulls)
    
    Pull only new listings: sort newest first, keep a high-water mark or a seen-ID set per query, and stop paging early. Python tested on Craigslist and eBay data.