A search-results table cannot tell you whether a query returned no organic results or whether the collector never received usable data. Both situations can leave the table with no rows. If you want to compare observations over time, the run itself needs a record.
This tutorial stores Google SERP captures in SQLite using a parent runs table and a child results table. It retains the original JSON, keeps the submitted request as the comparison key, and refuses to compare pending or failed observations as if they were ranking changes.
Scrapeless Google Search API supplies the upstream search data. The SQLite model is application code you own; it builds history from captures you save rather than retrieving an existing historical ranking database.
Define the Capture Contract
The importer expects a JSON capture envelope, not a bare API response. Each file contains a nonempty run_id, a request object, and the HTTP outcome. The request contains actor: scraper.google.search and an input object holding the exact submitted search settings.
For a received response, preserve http_status, received_at, and response. Keep requested_at too if the collector records it. A transport failure can have no HTTP status and no response; the run still exists and belongs in the collection-coverage report.
These envelope fields are defined by the application. The upstream request contract uses POST https://api.scrapeless.com/api/v1/scraper/request and authentication through x-api-token. HTTP 200 returns task data as the response body; HTTP 201 indicates a pending task. Do not assume that a successful network exchange always contains completed search results.
The importer below makes no network calls and needs no API key. Live collection is a separate prerequisite, and no authenticated search result is claimed in this tutorial. Local tests use explicitly synthetic records to exercise storage behavior.
Separate Runs From Organic Rows
runs stores one immutable capture, including its state and raw record. results stores the organic items from usable captures. The foreign key connects each result back to the search that produced it.
The child key is (run_id, ordinal). Ordinal preserves array order, while position retains the returned position if it is a positive integer. These fields are deliberately separate: a missing position should not be replaced with an invented ranking based on array order.
The raw_result field preserves optional fields and unusual values that the simple projection does not map. A nonstring title becomes null in the convenience column but remains visible in the raw item. A future parser can reconstruct a richer projection from the archive.
SQLite foreign-key enforcement must be enabled for the connection. The script does that before importing. Its parameterized statements also keep query values separate from SQL text, following the Python sqlite3 interface.
Use the Whole Request as the Context Key
A conservative comparison key serializes the submitted request with sorted object keys. It includes country, language, pagination offset, and any other settings your collector supplied. It excludes run identifiers and client timestamps because those live outside the request.
This is a key for exact submitted configurations, not a universal semantic canonicalizer. Explicitly providing a default and omitting it produce different keys. Equivalent-looking URLs can also remain different. Keeping them separate is safer than merging them without a documented rule.
The country and language parameter definitions explain why a query alone is insufficient. A different start also changes the collected slice, so it remains part of the key. The script compares page slices; combining several pages into a collection window would need a separate parent model.
Sorting keys makes the serialization stable for these JSON objects. The allow_nan=False setting rejects nonstandard numeric values during storage rather than silently creating an archive that is difficult to exchange with another JSON implementation.
Import Captures With One Transaction Per Run
Save this program as serp_sqlite.py. It uses only Python's standard library. Put capture files with the envelope described above in a directory such as snapshots, then run python3 serp_sqlite.py research.sqlite --input-dir snapshots.
The program creates its tables if needed. Run imports are transactional: the parent and its children are committed together. Earlier successfully imported files remain committed if a later file is invalid; the script does not promise an all-or-nothing transaction for the entire directory.
import argparse
import json
import sqlite3
from pathlib import Path
from urllib.parse import urlsplit
SCHEMA = """
CREATE TABLE IF NOT EXISTS runs (
run_id TEXT PRIMARY KEY,
context_key TEXT NOT NULL,
received_at TEXT,
state TEXT NOT NULL,
raw_record TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS results (
run_id TEXT NOT NULL REFERENCES runs(run_id),
ordinal INTEGER NOT NULL,
position INTEGER,
hostname TEXT,
title TEXT,
link TEXT,
raw_result TEXT NOT NULL,
PRIMARY KEY (run_id, ordinal)
);
CREATE INDEX IF NOT EXISTS runs_context ON runs(context_key, received_at);
CREATE INDEX IF NOT EXISTS results_host ON results(hostname);
"""
def canonical(value):
return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"), allow_nan=False)
def state_and_rows(record):
payload = record.get("response")
rows = payload.get("organic_results") if isinstance(payload, dict) else None
status = record.get("http_status")
if status == 201:
return "pending", []
if status is None:
return "transport_error", []
if status != 200:
return "http_error", []
if not isinstance(rows, list) or any(not isinstance(x, dict) for x in rows):
return "unmapped", []
return ("observed" if rows else "empty"), rows
def hostname(link):
if not isinstance(link, str):
return None
try:
parsed = urlsplit(link)
return parsed.hostname.lower().rstrip(".") if parsed.scheme in ("http", "https") and parsed.hostname else None
except ValueError:
return None
def import_record(db, record):
run_id, request = record.get("run_id"), record.get("request")
if not isinstance(run_id, str) or not run_id or not isinstance(request, dict):
raise ValueError("A nonempty run_id and request object are required")
if not isinstance(request.get("input"), dict) or request.get("actor") != "scraper.google.search":
raise ValueError("Expected the Google Search request envelope")
raw = canonical(record)
existing = db.execute("SELECT raw_record FROM runs WHERE run_id=?", (run_id,)).fetchone()
if existing:
if existing[0] != raw:
raise ValueError("Conflicting content for existing run_id: " + run_id)
return False
state, rows = state_and_rows(record)
with db:
db.execute("INSERT INTO runs VALUES (?, ?, ?, ?, ?)",
(run_id, canonical(request), record.get("received_at"), state, raw))
for ordinal, row in enumerate(rows):
position = row.get("position")
if type(position) is not int or position < 1:
position = None
title = row.get("title") if isinstance(row.get("title"), str) else None
link = row.get("link") if isinstance(row.get("link"), str) else None
db.execute("INSERT INTO results VALUES (?, ?, ?, ?, ?, ?, ?)",
(run_id, ordinal, position, hostname(link), title, link, canonical(row)))
return True
def compare(db, before, after):
records = [db.execute("SELECT context_key, state FROM runs WHERE run_id=?", (x,)).fetchone()
for x in (before, after)]
if any(x is None for x in records):
raise ValueError("Both runs must exist")
if records[0][0] != records[1][0]:
raise ValueError("Request contexts differ")
if any(x[1] not in ("observed", "empty") for x in records):
raise ValueError("Only usable observations can be compared")
query = "SELECT DISTINCT hostname FROM results WHERE run_id=? AND hostname IS NOT NULL"
old = {x[0] for x in db.execute(query, (before,))}
new = {x[0] for x in db.execute(query, (after,))}
return {"newly_observed": sorted(new - old), "not_observed_later": sorted(old - new),
"present_in_both": sorted(old & new)}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("database")
parser.add_argument("--input-dir")
parser.add_argument("--compare", nargs=2, metavar=("BEFORE", "AFTER"))
args = parser.parse_args()
db = sqlite3.connect(args.database)
try:
db.execute("PRAGMA foreign_keys = ON")
db.executescript(SCHEMA)
if args.input_dir:
paths = sorted(Path(args.input_dir).glob("*.json"))
if not paths:
raise ValueError("No JSON capture files found")
added = sum(import_record(db, json.loads(p.read_text(encoding="utf-8"))) for p in paths)
print(f"Imported {added} new runs from {len(paths)} files")
if args.compare:
print(json.dumps(compare(db, *args.compare), ensure_ascii=False, indent=2))
print("Runs:", db.execute("SELECT COUNT(*) FROM runs").fetchone()[0])
finally:
db.close()
if __name__ == "__main__":
main()
A repeated import of identical content with the same run identifier is a no-op. Different content under an existing identifier raises an error. That prevents a later capture from silently replacing the evidence used in an earlier report.
For a pending task that later completes, preserve the original pending capture and write the completed retrieval as a new observation with its own identifier. A larger system can add a task identifier to relate those records. Do not change the stored raw record under the same run identifier merely to make a dashboard look complete.
Keep Empty and Unavailable Outcomes Visible
The script derives state from the recorded HTTP status and response shape. It does not trust an arbitrary incoming state label as proof that the data is usable.
A nonempty organic array containing objects becomes observed. A present empty array becomes empty. A missing array, a non-array field, or a malformed item makes the run unmapped, with no projected result rows. The raw record remains available for inspection.
HTTP 201 becomes pending, another non-200 HTTP status becomes http_error, and a missing status becomes transport_error under this capture contract. These are application classifications. They are not extra API response codes.
You can now distinguish “no child rows because the response was empty” from “no child rows because the task was pending.” That distinction matters when preparing any absence-based metric. Keep unresolved states in the coverage report and exclude them from result comparisons.
Compare Domain Sets Within One Context
To compare saved runs, execute python3 serp_sqlite.py research.sqlite --compare BEFORE_RUN_ID AFTER_RUN_ID, replacing the identifiers with actual values from your database.
The comparison first checks that both runs exist, have identical request keys, and are usable. It then selects distinct non-null hostnames and reports domains newly observed, not observed in the later slice, and present in both.
This is hostname comparison. It does not merge subdomains into a corporate entity or infer a registrable domain. If that is required, introduce a reviewed grouping policy and retain the original hostname. The URL parsing reference also makes clear that parsing is not full input validation; the helper never fetches the URLs it reads.
“Not observed later” is intentionally scoped to the captured slice. It is not a claim that the domain disappeared from Google. An empty usable array can produce an empty set, but the report still needs its state and collection context attached.
Query Coverage Before Building a Trend
Inspect counts grouped by state before interpreting a series. A planned collection may include pending, failed, or unmapped runs that do not appear in the child table. Looking only at results would hide them.
Keep the definition of a comparison stable. If you change parser rules, domain grouping, or query coverage, record the version and consider rebuilding derived rows from archived JSON. The raw captures should remain immutable.
SQLite is a useful local starting point because the records and joins are easy to inspect. This example does not implement scheduling, remote storage, concurrent ingestion coordination, or schema migrations. Add those only when the actual collection workflow requires them, with explicit rules for preserving existing evidence.
Conclusion
Store the run even when it has no organic rows. Preserve the raw capture, compare exact request contexts, and make repeated imports safe to inspect. Those choices let the database distinguish collection health from observed domain movement.
FAQ
Why not use (query, date) as the primary key?
It can collapse different markets, languages, page offsets, or several observations on the same day. A run identifier preserves each capture; the request key defines comparable context.
Does this program obtain past rankings?
No. It imports capture files you already have. Your history starts with saved observations.
Can pending tasks count as empty results?
No. Pending tasks are stored under their own state and rejected by the comparison function until a separate completed observation is available.
Is the hostname list a competitor traffic report?
No. It describes domain presence in the selected organic slices. It does not contain visits, market share, or revenue.
Top comments (0)