Blog / Data engineering

Detect Schema Changes in API JSON Responses

20 Oct 2026Perceptlayer6 min read

To detect schema changes in API JSON responses, infer a schema from a known-good response, store it, and compare every new response against it: report paths that disappeared, paths that changed type, and paths that are new. A few dozen lines of Python do it with no dependencies. Run on our recorded Google Search response, the inference finds 71 paths, and the checks below catch a removed snippet or a position that turns into a string before your pipeline breaks on it.

What counts as a schema change in a web data response?

Three things, in order of how much they hurt. A retyped field, such as a number that arrives as a string, breaks parsing and arithmetic immediately. A removed field breaks any code that reads it, and fills your tables with nulls if the code tolerates it. A new field breaks nothing, but it often shows up just before a change to an old one, so it's worth logging.

Web data APIs change for reasons outside anyone's control: the website behind an endpoint adds a module, renames a section, or shows a field only for some results. That makes the second category tricky. A field that's missing from one response may simply not apply to it, so a useful check separates "absent this time" from "gone".

Which response do you start from?

A known-good response from the endpoint you depend on, saved to disk. This guide uses our recorded Google Search response. The request was:

curl "https://api.perceptlayer.com/v1/google-search/search?query=best+restaurants+in+Berlin&hl=en&safe=off" \
  -H "X-API-Key: $API_KEY"

The response mixes flat fields, arrays of objects and nested modules, which is what makes schema checks worth having. A trimmed excerpt:

{
  "search_information": { "query_displayed": "best restaurants in Berlin", "organic_results_count": 6 },
  "organic_results": [
    { "position": 1, "title": "The Best Restaurants in Berlin - Food Guide & Map",
      "link": "https://andershusa.com/best-restaurants-berlin-germany-guide/",
      "snippet": "Dec 3, 2025 — Tim Raue is the most internationally renowned restaurant in Berlin. …",
      "rank_group": 1, "rank_absolute": 4, … },
    …
  ],
  "local_results": {
    "places": [
      { "position": 1, "title": "BLESS Restaurant", "rating": 4.8, "reviews": 2400,
        "phone": "+49 30 20619795", "kgmid": "/g/11lkzg8dhv", … },
      …
    ],
    …
  },
  …
}

Save the full response as baseline.json. Get a key from the free signup if you want to record your own; the requests and responses docs describe the envelope every endpoint shares.

How do you infer a schema from a response?

Walk the JSON and record, for every path, which types appear there. Objects add .key to the path and arrays add [], so all items of an array share one path and their types merge. Map Python types to JSON types, with int and float both as number, because JSON doesn't distinguish them and data often switches between 5 and 4.81.

# schema.py
import json

KIND = {dict: "object", list: "array", str: "string", bool: "boolean",
        int: "number", float: "number", type(None): "null"}

def infer(value, path="$", schema=None):
    """Map every path in a JSON value to the set of types seen there."""
    schema = {} if schema is None else schema
    schema.setdefault(path, set()).add(KIND[type(value)])
    if isinstance(value, dict):
        for key, child in value.items():
            infer(child, f"{path}.{key}", schema)
    elif isinstance(value, list):
        for child in value:
            infer(child, f"{path}[]", schema)
    return schema

def diff(old, new):
    removed = sorted(p for p in old if p not in new)
    added = sorted(p for p in new if p not in old)
    retyped = sorted((p, sorted(old[p]), sorted(new[p]))
                     for p in old.keys() & new.keys()
                     if not new[p] <= old[p] | {"null"})
    return removed, added, retyped

def save(schema, path):
    json.dump({p: sorted(t) for p, t in schema.items()}, open(path, "w"), indent=1)

def load(path):
    return {p: set(t) for p, t in json.load(open(path)).items()}

Run save(infer(json.load(open("baseline.json"))), "schema.json") on the recorded response and you get 71 paths. The first few are $.search_information.query_displayed (string), $.search_information.organic_results_count (number), $.organic_results (array) and $.organic_results[].position (number). The diff function tolerates a new null on a known path, so a field that's sometimes empty doesn't raise a type alert; drop the | {"null"} if your pipeline can't take nulls.

Why can't you alert on the first missing field?

Because many fields are optional by nature. In the recorded response, phone is present on 2 of the 3 places in local_results; the third simply has no phone field. The path $.local_results.places[].phone exists in this response because two items carry it, but a future response where none of the places has a phone would make it look removed.

The same goes for whole modules. A search with no local businesses has no local_results at all, so every path under it would vanish in one go. The fix is to count: alert when a known path has been missing from several consecutive responses, not from one. Three is a sensible starting point for an endpoint you call many times a day; for a daily job, use a higher number or check per query.

How do you check each new response?

Diff the new response's schema against the stored one, alert on retyped paths at once, count missing paths across runs, and fold new paths into the stored schema after logging them. Keep the counts in a small state file so they survive between runs.

# monitor.py
import json
import os

from schema import diff, infer, load, save

STATE = "schema_state.json"
MISSING_LIMIT = 3

def check(response, schema_path="schema.json"):
    known = load(schema_path)
    current = infer(response)
    removed, added, retyped = diff(known, current)
    state = json.load(open(STATE)) if os.path.exists(STATE) else {}
    alerts = [f"RETYPED {p}: {old} -> {new}" for p, old, new in retyped]
    for p in known:
        state[p] = state.get(p, 0) + 1 if p in removed else 0
        if state[p] == MISSING_LIMIT:
            alerts.append(f"MISSING {p} in {MISSING_LIMIT} responses in a row")
    for p in added:
        known[p] = current[p]
        alerts.append(f"NEW {p}: {sorted(current[p])}")
    for p, old, new in retyped:
        known[p] |= current[p]
    save(known, schema_path)
    json.dump(state, open(STATE, "w"))
    return alerts

Each alert fires once: a retyped path's new type joins the stored schema, and a missing count only matches the limit on the run that reaches it. Call check(r.json()) after every successful API call, and send the alerts wherever your team looks.

What does it report on a real change?

We tested it by editing a copy of the recorded response: removing snippet from every organic result and changing the first result's position from 1 to the string "1". Checking that edited response three times printed:

1 ["RETYPED $.organic_results[].position: ['number'] -> ['number', 'string']"]
2 []
3 ['MISSING $.organic_results[].snippet in 3 responses in a row']

The type change was caught on the first response, as it should be, since parsing position as an integer would fail there. The missing snippet waited for the third, which is the price of not alerting on optional fields. Lower MISSING_LIMIT for paths you know are always present.

Should you use JSON Schema instead?

Use JSON Schema when other tools need to read the schema, such as validators in other languages or API documentation. The path map above is easier to diff and explain, but it isn't a standard. GenSON, a Python library, builds a JSON Schema from any number of JSON objects and can merge schemas, so you can grow one schema from many responses (GenSON, 2026). Validate new responses against it with any JSON Schema validator.

Either way, decide which paths you rely on and give them their own stricter rule. A check that treats all 71 paths equally will be noisy; one that watches the dozen fields your code reads, such as position, title, link and snippet, will be quiet until it matters. The web data quality checklist adds null rates, duplicates and outliers on top of schema checks.

What does it cost to monitor schemas?

Nothing extra in credits. The checks run on responses your pipeline already fetched, so they add CPU time, not calls. Each successful Google search is 1 credit, and a failed call costs nothing and has no response body to check. Check its error code instead; the errors docs list them.

If you'd rather run a separate canary, one fixed query a day for each endpoint you depend on, that's 1 credit per endpoint per day: 10 endpoints × 30 days = 300 credits a month, inside the 1,000 free monthly credits. The credits docs explain the balance and pricing has what packs cost. To record a baseline for your own query, run it in the playground on the Google Search API page and save the JSON.

jsonschema driftdata pipelinespythonmonitoring