Research restaurants, cafés and canteens

Save as query.mjs, then run node query.mjs. Download the JavaScript example.

// Node.js 20+. No packages required. Optional key: DATASCOOP_API_KEY.
const endpoint = "https://datascoop.io/v1/datasets/fsa/query";
const dataset = "fsa";
const filters = {
  "localAuthorityExact": "Leeds",
  "businessTypeExact": "Restaurant/Cafe/Canteen"
};
const PAGE_SIZE = 25, MAX_ROWS = 2500, MAX_PAGES = 100;
const headers = { Accept: "application/json" };
const key = (process.env.DATASCOOP_API_KEY || "").trim();
if (key) headers.Authorization = "Bearer " + key;

async function retrieve() {
  const rows = [];
  let total = null, importedAt = null, first = null;
  for (let page = 0; page < MAX_PAGES; page++) {
    const url = new URL(endpoint);
    url.search = new URLSearchParams({ ...filters, strict_filters: "1",
      limit: String(PAGE_SIZE), offset: String(rows.length) });
    const response = await fetch(url, { headers, signal: AbortSignal.timeout(30000) });
    if (!response.ok) throw new Error("Query failed: HTTP " + response.status +
      ". Check filters and access; retry later for rate limits or server errors.");
    const result = await response.json();
    if (result.dataset !== dataset || !Array.isArray(result.rows) ||
        result.rows.some(row => !row || typeof row !== "object" || Array.isArray(row)) ||
        !Array.isArray(result.ignored) || result.ignored.length ||
        !Number.isSafeInteger(result.total) || result.total < 0 ||
        result.offset !== rows.length || result.returned !== result.rows.length ||
        result.returned > PAGE_SIZE || typeof result.truncated !== "boolean") {
      throw new Error("Unexpected query response; no partial result was printed.");
    }
    if (result.total > MAX_ROWS) throw new Error("More than " + MAX_ROWS +
      " rows match. Narrow the filters or review the bound before increasing it.");
    for (const [name, value] of Object.entries(filters)) {
      if (String(result.applied_filters?.[name]) !== String(value))
        throw new Error("A requested filter was not applied: " + name);
    }
    const imported = result.provenance?.imported_at;
    if (typeof imported !== "string" || !Number.isFinite(Date.parse(imported)))
      throw new Error("Import date is missing; no partial result was printed.");
    if (first === null) { first = result; total = result.total; importedAt = imported; }
    if (result.total !== total || imported !== importedAt)
      throw new Error("The import or match count changed. Run the request again.");
    const next = rows.length + result.returned;
    if (next > total || result.truncated !== (next < total) ||
        (result.returned === 0 && next < total))
      throw new Error("Pagination stopped making sense; no partial result was printed.");
    rows.push(...result.rows);
    if (rows.length === total) return { dataset, filters, total, rows,
      provenance: first.provenance, licence: first.licence, attribution: first.attribution };
  }
  throw new Error("Page limit reached. Narrow the filters; no partial result was printed.");
}

try {
  const result = await retrieve();
  if (result.total === 0) console.error("No matching records. Check the selected area and filters.");
  // Keep provenance and attribution with cached or republished data.
  console.log(JSON.stringify(result, null, 2));
} catch (error) {
  console.error(error instanceof Error ? error.message : "Request failed.");
  process.exitCode = 1;
}

Save as query.py, then run python3 query.py. Download the Python example.

# Python 3. No packages required. Optional key: DATASCOOP_API_KEY.
import json
import os
import sys
from datetime import datetime
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen

ENDPOINT = "https://datascoop.io/v1/datasets/fsa/query"
DATASET = "fsa"
FILTERS = json.loads("{\"localAuthorityExact\":\"Leeds\",\"businessTypeExact\":\"Restaurant/Cafe/Canteen\"}")
PAGE_SIZE, MAX_ROWS, MAX_PAGES = 25, 2500, 100
HEADERS = {"Accept": "application/json"}
key = os.environ.get("DATASCOOP_API_KEY", "").strip()
if key:
    HEADERS["Authorization"] = "Bearer " + key

def text_value(value):
    return "true" if value is True else "false" if value is False else str(value)

def retrieve():
    rows, total, imported_at, first = [], None, None, None
    for _ in range(MAX_PAGES):
        params = dict(FILTERS, strict_filters="1", limit=PAGE_SIZE, offset=len(rows))
        request = Request(ENDPOINT + "?" + urlencode(params), headers=HEADERS)
        with urlopen(request, timeout=30) as response:
            result = json.load(response)
        if (not isinstance(result, dict) or result.get("dataset") != DATASET or
            not isinstance(result.get("rows"), list) or
            any(not isinstance(row, dict) for row in result["rows"]) or
            not isinstance(result.get("ignored"), list) or result["ignored"] or
            type(result.get("total")) is not int or result["total"] < 0 or
            type(result.get("offset")) is not int or result["offset"] != len(rows) or
            type(result.get("returned")) is not int or
            result.get("returned") != len(result["rows"]) or
            result["returned"] > PAGE_SIZE or type(result.get("truncated")) is not bool):
            raise ValueError("Unexpected query response; no partial result was printed.")
        if result["total"] > MAX_ROWS:
            raise ValueError("More than 2500 rows match. Narrow filters or review the bound.")
        applied = result.get("applied_filters")
        if not isinstance(applied, dict):
            raise ValueError("Applied filters are missing.")
        for name, value in FILTERS.items():
            if name not in applied or text_value(applied[name]) != text_value(value):
                raise ValueError("A requested filter was not applied: " + name)
        provenance = result.get("provenance")
        imported = provenance.get("imported_at") if isinstance(provenance, dict) else None
        if not isinstance(imported, str):
            raise ValueError("Import date is missing; no partial result was printed.")
        datetime.fromisoformat(imported.replace("Z", "+00:00"))
        if first is None:
            first, total, imported_at = result, result["total"], imported
        if result["total"] != total or imported != imported_at:
            raise ValueError("The import or match count changed. Run the request again.")
        next_offset = len(rows) + result["returned"]
        if (next_offset > total or result["truncated"] != (next_offset < total) or
            (result["returned"] == 0 and next_offset < total)):
            raise ValueError("Pagination stopped making sense; no partial result was printed.")
        rows.extend(result["rows"])
        if len(rows) == total:
            return dict(dataset=DATASET, filters=FILTERS, total=total, rows=rows,
                        provenance=first["provenance"], licence=first.get("licence"),
                        attribution=first.get("attribution"))
    raise ValueError("Page limit reached. Narrow filters; no partial result was printed.")

try:
    result = retrieve()
    if result["total"] == 0:
        print("No matching records. Check the selected area and filters.", file=sys.stderr)
    # Keep provenance and attribution with cached or republished data.
    print(json.dumps(result, indent=2))
except HTTPError as error:
    print("Query failed: HTTP " + str(error.code) + ". Check filters and access; retry later for rate limits or server errors.", file=sys.stderr)
    sys.exit(1)
except (URLError, TimeoutError, ValueError, TypeError, KeyError) as error:
    print(str(error), file=sys.stderr)
    sys.exit(1)