Python

pip install requests

import os
import requests

API_KEY = os.environ["SCHEMACHECK_API_KEY"]
BASE_URL = "https://schemacheck.dev/api/v1"


def validate_url(url: str) -> dict:
    """Validate all JSON-LD schemas on a public URL."""
    response = requests.get(
        f"{BASE_URL}/validate",
        params={"url": url, "access_key": API_KEY},
        timeout=30,
    )
    data = response.json()

    if not response.ok or not data.get("success"):
        error = data.get("error", {})
        raise ValueError(f"[{error.get('code')}] {error.get('message')}")

    return data


# Example usage
result = validate_url("https://stripe.com")

print(f"Score: {result['summary']['score']}/100")
print(f"Schemas found: {result['schemas_found']}")

for schema in result["schemas"]:
    print(f"\n[{schema['type']}] valid={schema['valid']} rich_result={schema['rich_result_eligible']}")

    for error in schema["errors"]:
        print(f"  ✗ {error['property']}: {error['message']}")
        print(f"    Fix: {error['fix']}")

    for warning in schema["warnings"]:
        print(f"  ⚠ {warning['property']}: {warning['message']}")

import os
import requests

API_KEY = os.environ["SCHEMACHECK_API_KEY"]


def validate_jsonld(jsonld: dict) -> dict:
    """Validate a raw JSON-LD object. Useful before publishing."""
    response = requests.post(
        "https://schemacheck.dev/api/v1/validate",
        headers={
            "Content-Type": "application/json",
            "x-api-key": API_KEY,
        },
        json={"jsonld": jsonld},
        timeout=30,
    )
    data = response.json()

    if not response.ok or not data.get("success"):
        error = data.get("error", {})
        raise ValueError(f"[{error.get('code')}] {error.get('message')}")

    return data


# Example: validate Article schema before publishing
result = validate_jsonld({
    "@context": "https://schema.org",
    "@type": "Article",
    "headline": "How to validate Schema.org structured data",
    "author": {"@type": "Person", "name": "Jane Doe"},
    "datePublished": "2026-03-18",
    "image": "https://example.com/photo.jpg",
})

schema = result["schemas"][0]
if not schema["valid"]:
    print("Schema has errors:")
    for e in schema["errors"]:
        print(f"  {e['property']}: {e['fix']}")
    raise SystemExit(1)

print("Valid! Rich result eligible:", schema["rich_result_eligible"])

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests

API_KEY = os.environ["SCHEMACHECK_API_KEY"]


def validate_url(url: str) -> tuple[str, dict | Exception]:
    try:
        response = requests.get(
            "https://schemacheck.dev/api/v1/validate",
            params={"url": url, "access_key": API_KEY},
            timeout=30,
        )
        data = response.json()
        if not response.ok:
            raise ValueError(data.get("error", {}).get("message", "Unknown error"))
        return url, data
    except Exception as e:
        return url, e


def validate_batch(urls: list[str], max_workers: int = 5) -> dict:
    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(validate_url, url): url for url in urls}
        for future in as_completed(futures):
            url, result = future.result()
            results[url] = result
    return results


urls = [
    "https://stripe.com",
    "https://github.com",
    "https://shopify.com",
]

results = validate_batch(urls)

for url, result in results.items():
    if isinstance(result, Exception):
        print(f"{url} → ERROR: {result}")
    else:
        print(f"{url} → score {result['summary']['score']}/100")

import os
import time
import requests

API_KEY = os.environ["SCHEMACHECK_API_KEY"]


def validate_with_retry(url: str, max_retries: int = 3) -> dict:
    for attempt in range(1, max_retries + 1):
        response = requests.get(
            "https://schemacheck.dev/api/v1/validate",
            params={"url": url, "access_key": API_KEY},
            timeout=30,
        )
        data = response.json()

        if response.ok:
            return data

        code = data.get("error", {}).get("code", "")

        # Never retry client errors (except rate limits)
        if 400 <= response.status_code < 500 and code != "rate_limit_exceeded":
            raise ValueError(f"[{code}] {data['error']['message']}")

        if code == "rate_limit_exceeded":
            retry_after = int(response.headers.get("Retry-After", 60))
            print(f"Rate limited. Waiting {retry_after}s…")
            time.sleep(retry_after)
            continue

        # Exponential backoff for 5xx
        if attempt < max_retries:
            delay = 2 ** (attempt - 1)
            print(f"Attempt {attempt} failed. Retrying in {delay}s…")
            time.sleep(delay)

    raise RuntimeError("Max retries exceeded")