Ipums Microdata API

https://api.ipums.org/extracts/

# Register at https://www.ipums.org/
# API key from your account settings
export IPUMS_KEY="..."

# Request a data extract (IPUMS USA example)
curl -X POST "https://api.ipums.org/extracts/?collection=usa&version=2" \
  -H "Authorization: $IPUMS_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "description": "Income by education, 2020 ACS",
    "data_structure": {"rectangular": {"on": "P"}},
    "data_format": "csv",
    "samples": {"us2020a": {}},
    "variables": {
      "AGE": {},
      "SEX": {},
      "RACE": {},
      "EDUC": {},
      "INCTOT": {},
      "EMPSTAT": {}
    }
  }'

curl "https://api.ipums.org/extracts/42?collection=usa&version=2" \
  -H "Authorization: $IPUMS_KEY"

# When status is "completed"
curl -O "https://api.ipums.org/extracts/42/download?collection=usa&version=2" \
  -H "Authorization: $IPUMS_KEY"

# List available variables
curl "https://api.ipums.org/metadata/usa/variables?version=2" \
  -H "Authorization: $IPUMS_KEY"

# Get variable details
curl "https://api.ipums.org/metadata/usa/variables/EDUC?version=2" \
  -H "Authorization: $IPUMS_KEY"

# List available samples
curl "https://api.ipums.org/metadata/usa/samples?version=2" \
  -H "Authorization: $IPUMS_KEY"

import os
import time
import requests

BASE_URL = "https://api.ipums.org"
HEADERS = {"Authorization": os.environ.get("IPUMS_KEY", "")}


def create_extract(collection: str, samples: dict,
                   variables: list, description: str = "",
                   data_format: str = "csv") -> int:
    """Create an IPUMS data extract request."""
    var_dict = {v: {} for v in variables}
    body = {
        "description": description,
        "data_format": data_format,
        "data_structure": {"rectangular": {"on": "P"}},
        "samples": {s: {} for s in samples} if isinstance(samples, list)
                   else samples,
        "variables": var_dict,
    }

    resp = requests.post(
        f"{BASE_URL}/extracts/?collection={collection}&version=2",
        headers={**HEADERS, "Content-Type": "application/json"},
        json=body,
    )
    resp.raise_for_status()
    return resp.json()["number"]


def wait_for_extract(extract_id: int, collection: str,
                     poll_interval: int = 30) -> str:
    """Poll until extract is ready, return download URL."""
    while True:
        resp = requests.get(
            f"{BASE_URL}/extracts/{extract_id}"
            f"?collection={collection}&version=2",
            headers=HEADERS,
        )
        resp.raise_for_status()
        data = resp.json()
        status = data.get("status")

        if status == "completed":
            return data["download_links"]["data"]["url"]
        elif status == "failed":
            raise RuntimeError(f"Extract failed: {data}")
        print(f"Status: {status}, waiting {poll_interval}s...")
        time.sleep(poll_interval)


def get_variable_info(collection: str, variable: str) -> dict:
    """Get metadata about a variable."""
    resp = requests.get(
        f"{BASE_URL}/metadata/{collection}/variables/{variable}"
        f"?version=2",
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()


# Example: request 2020 ACS income data
extract_id = create_extract(
    collection="usa",
    samples=["us2020a"],
    variables=["AGE", "SEX", "RACE", "EDUC", "INCTOT", "EMPSTAT"],
    description="Education-income analysis 2020",
)
print(f"Extract #{extract_id} submitted. Waiting...")

download_url = wait_for_extract(extract_id, "usa")
print(f"Ready: {download_url}")