Skip to main content

Python Examples

These examples use the requests library. Install it with:

pip install requests

Store your API key in an environment variable — never hardcode it in source files:

export PRAIRIECLOUD_API_KEY="pck_live_YOUR_KEY_HERE"

Setup​

import os
import requests

API_BASE = "https://api.prairiecloud.io"
API_KEY = os.environ["PRAIRIECLOUD_API_KEY"]

# Reusable session with auth header
session = requests.Session()
session.headers.update({"X-API-Key": API_KEY})

Single Variable Lookup​

response = session.get(
    f"{API_BASE}/v1/data",
    params={
        "variables": "pop_total",
        "geo": "state:48",
    }
)
response.raise_for_status()

result = response.json()
geo_data = result["data"]["state:48"]
population = geo_data["variables"]["pop_total"]["estimate"]

print(f"{geo_data['name']}: {population:,}")
# → Texas: 30,029,572

Multi-Variable Query​

response = session.get(
    f"{API_BASE}/v1/data",
    params={
        "variables": "pop_total,income_median_household",
        "geo": "state:48,state:06,state:36",
        "vintage": 2023,
    }
)
response.raise_for_status()

data = response.json()["data"]

for geo_key, geo_data in data.items():
    name = geo_data["name"]
    population = geo_data["variables"]["pop_total"]["estimate"]
    income = geo_data["variables"]["income_median_household"]["estimate"]
    print(f"{name}: pop={population:,}, median income=${income:,.0f}")

Output:

Texas: pop=30,029,572, median income=$67,321
California: pop=39,455,353, median income=$84,097
New York: pop=19,677,151, median income=$74,314

Browse the Variable Catalog​

def search_variables(keyword, limit=20):
    """Search the variable catalog by keyword."""
    response = session.get(
        f"{API_BASE}/v1/variables",
        params={"search": keyword, "limit": limit}
    )
    response.raise_for_status()
    return response.json()["data"]

variables = search_variables("income")
for var in variables:
    print(f"{var['api_name']} — {var['label']} ({var['unit']})")

Paginating the Variable Catalog​

def get_all_variables():
    """Fetch all variables by paginating through the catalog."""
    all_vars = []
    offset = 0
    limit = 200  # maximum per request

    while True:
        response = session.get(
            f"{API_BASE}/v1/variables",
            params={"limit": limit, "offset": offset}
        )
        response.raise_for_status()
        payload = response.json()

        batch = payload["data"]
        all_vars.extend(batch)

        total = payload["meta"]["total_count"]
        offset += len(batch)

        if offset >= total:
            break

    return all_vars

all_vars = get_all_variables()
print(f"Total variables in catalog: {len(all_vars)}")

Error Handling​

def safe_request(url, params):
    """Make a request with proper error handling."""
    try:
        response = session.get(url, params=params)

        if response.status_code == 401:
            raise ValueError("Invalid or missing API key")

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            raise RuntimeError(f"Rate limited. Retry after {retry_after} seconds.")

        if response.status_code == 404:
            problem = response.json()
            raise LookupError(f"Not found: {problem['detail']}")

        response.raise_for_status()
        return response.json()

    except requests.exceptions.ConnectionError:
        raise RuntimeError("Could not connect to the API. Check your network.")

Retry with Exponential Backoff​

import time
import random

def request_with_retry(url, params, max_retries=5):
    """Request with Retry-After-aware exponential backoff."""
    for attempt in range(max_retries):
        response = session.get(url, params=params)

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 1))
            wait = retry_after + random.uniform(0, 1)
            print(f"Rate limited. Waiting {wait:.1f}s (attempt {attempt + 1}/{max_retries})")
            time.sleep(wait)
            continue

        if response.status_code >= 500:
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"Server error {response.status_code}. Waiting {wait:.1f}s")
            time.sleep(wait)
            continue

        response.raise_for_status()
        return response.json()

    raise RuntimeError(f"Request failed after {max_retries} retries")

Check Your Usage​

response = session.get(f"{API_BASE}/v1/account/usage")
response.raise_for_status()
usage = response.json()["data"]

print(f"Plan: {usage['tier']}")
print(f"Usage: {usage['quota_used']:,} / {usage['quota_limit']:,}")
print(f"Remaining: {usage['quota_remaining']:,}")
print(f"Resets: {usage['period_end']}")

Next Steps​