← Back to Developer Hub

Python: Analyze Water Quality with ZipCheckup API

Use Python to fetch, analyze, and visualize water quality data for any U.S. ZIP code

Setup

Install the required libraries:

pip install requests pandas matplotlib

That's it. The ZipCheckup API is a standard REST endpoint — no SDK or API key required for the free tier.

Quick test

import requests

resp = requests.get("https://zipcheckup.com/api/v1/zip/90210")
data = resp.json()

print(f"ZIP: {data['zip']}")
print(f"Score: {data['score']}/100 (Grade: {data['grade']})")
print(f"Violations: {data['violations']}")
print(f"Lead level: {data['lead_level']}")

Output:

ZIP: 90210
Score: 72/100 (Grade: B)
Violations: 3
Lead level: low

Fetching Data for a Single ZIP Code

Here's a complete function that handles errors and returns a clean dictionary:

import requests

API_BASE = "https://zipcheckup.com/api/v1"

def get_zip_report(zip_code: str) -> dict | None:
    """Fetch the ZipCheckup report for a single ZIP code."""
    url = f"{API_BASE}/zip/{zip_code}"
    resp = requests.get(url, timeout=10)

    if resp.status_code == 404:
        print(f"ZIP {zip_code} not found")
        return None
    if resp.status_code == 429:
        print("Rate limit reached (100/day on free tier)")
        return None

    resp.raise_for_status()
    return resp.json()


# Usage
report = get_zip_report("60614")
if report:
    print(f"{report['zip']}: {report['grade']} ({report['score']})")
    for component, score in report['components'].items():
        print(f"  {component}: {score}")

Batch Fetching Multiple ZIP Codes

To compare data across ZIP codes, fetch them in a loop with a small delay to respect rate limits:

import time
import requests

API_BASE = "https://zipcheckup.com/api/v1"

def get_batch_reports(zip_codes: list[str], delay: float = 0.5) -> list[dict]:
    """Fetch reports for multiple ZIP codes with rate limiting."""
    results = []

    for i, zip_code in enumerate(zip_codes):
        try:
            resp = requests.get(f"{API_BASE}/zip/{zip_code}", timeout=10)
            if resp.status_code == 200:
                results.append(resp.json())
                print(f"[{i+1}/{len(zip_codes)}] {zip_code}: OK")
            elif resp.status_code == 429:
                print(f"Rate limit hit at ZIP #{i+1}. Collected {len(results)} results.")
                break
            else:
                print(f"[{i+1}/{len(zip_codes)}] {zip_code}: HTTP {resp.status_code}")
        except requests.RequestException as e:
            print(f"[{i+1}/{len(zip_codes)}] {zip_code}: Error - {e}")

        if i < len(zip_codes) - 1:
            time.sleep(delay)

    return results


# Example: Compare Chicago neighborhoods
chicago_zips = ["60601", "60614", "60622", "60637", "60657", "60660"]
reports = get_batch_reports(chicago_zips)

Building a pandas DataFrame

Convert API responses into a tabular format for analysis:

import pandas as pd

def reports_to_dataframe(reports: list[dict]) -> pd.DataFrame:
    """Convert a list of ZipCheckup reports to a pandas DataFrame."""
    rows = []
    for r in reports:
        row = {
            "zip": r["zip"],
            "score": r["score"],
            "grade": r["grade"],
            "violations": r["violations"],
            "lead_level": r["lead_level"],
            "radon_zone": r["radon_zone"],
            "flood_claims": r["flood_claims"],
        }
        # Flatten component scores
        for comp, val in r.get("components", {}).items():
            row[f"comp_{comp}"] = val
        rows.append(row)

    return pd.DataFrame(rows)


df = reports_to_dataframe(reports)
print(df.to_string(index=False))

Output:

   zip  score grade  violations lead_level  radon_zone  flood_claims  comp_water_quality  comp_lead_risk  comp_radon_risk  comp_flood_risk  comp_air_quality
 60601     65     C           5        low           2           312                  62              80               58               45                68
 60614     72     B           3        low           2           148                  78              85               62               55                71
 60622     68     C           4        low           2           205                  70              82               60               50                69
 60637     41     D           9       high           1           487                  35              42               52               38                55
 60657     74     B           2        low           2           132                  80              88               64               58                72
 60660     70     B           3        low           2           178                  72              84               61               53                70

Useful DataFrame operations

# Sort by score (worst first)
df.sort_values("score", ascending=True)

# Filter to high-risk ZIPs
high_risk = df[df["score"] < 50]

# Average score across all ZIPs
print(f"Average score: {df['score'].mean():.1f}")

# Grade distribution
print(df["grade"].value_counts())

# Worst component across all ZIPs
comp_cols = [c for c in df.columns if c.startswith("comp_")]
worst = df[comp_cols].mean().idxmin()
print(f"Weakest component: {worst} ({df[comp_cols].mean().min():.1f})")

Visualization with matplotlib

Bar Chart: ZIP Code Comparison

import matplotlib.pyplot as plt

def plot_zip_scores(df: pd.DataFrame, title: str = "ZipCheckup Scores"):
    """Create a horizontal bar chart comparing ZIP code scores."""
    df_sorted = df.sort_values("score", ascending=True)

    colors = []
    for grade in df_sorted["grade"]:
        if grade == "A":
            colors.append("#2E7D32")
        elif grade == "B":
            colors.append("#558B2F")
        elif grade == "C":
            colors.append("#F9A825")
        elif grade == "D":
            colors.append("#EF6C00")
        else:
            colors.append("#C62828")

    fig, ax = plt.subplots(figsize=(10, max(4, len(df) * 0.6)))
    bars = ax.barh(df_sorted["zip"], df_sorted["score"], color=colors, height=0.6)

    # Add score labels
    for bar, score, grade in zip(bars, df_sorted["score"], df_sorted["grade"]):
        ax.text(bar.get_width() + 1, bar.get_y() + bar.get_height()/2,
                f"{score} ({grade})", va="center", fontsize=11)

    ax.set_xlabel("Home Safety Score (0-100)")
    ax.set_title(title)
    ax.set_xlim(0, 105)
    ax.spines["top"].set_visible(False)
    ax.spines["right"].set_visible(False)
    plt.tight_layout()
    plt.savefig("zip_scores.png", dpi=150)
    plt.show()


plot_zip_scores(df, "Chicago Neighborhood Water Quality Scores")

Radar Chart: Component Breakdown

import numpy as np

def plot_component_radar(report: dict):
    """Plot a radar chart of component scores for a single ZIP."""
    components = report["components"]
    labels = [k.replace("_", " ").title() for k in components.keys()]
    values = list(components.values())

    # Close the polygon
    angles = np.linspace(0, 2 * np.pi, len(labels), endpoint=False).tolist()
    values += values[:1]
    angles += angles[:1]

    fig, ax = plt.subplots(figsize=(6, 6), subplot_kw=dict(polar=True))
    ax.fill(angles, values, alpha=0.25, color="#1a6b8a")
    ax.plot(angles, values, color="#1a6b8a", linewidth=2)
    ax.set_xticks(angles[:-1])
    ax.set_xticklabels(labels, fontsize=11)
    ax.set_ylim(0, 100)
    ax.set_title(f"ZIP {report['zip']} — Component Scores", pad=20, fontsize=14)
    plt.tight_layout()
    plt.savefig(f"radar_{report['zip']}.png", dpi=150)
    plt.show()


# Plot radar for a single ZIP
plot_component_radar(reports[0])

Plotly Interactive Chart (Optional)

For web dashboards, use Plotly instead of matplotlib:

# pip install plotly
import plotly.express as px

fig = px.bar(
    df.sort_values("score"),
    x="score",
    y="zip",
    orientation="h",
    color="grade",
    color_discrete_map={"A": "#2E7D32", "B": "#558B2F", "C": "#F9A825", "D": "#EF6C00", "F": "#C62828"},
    title="ZipCheckup Scores by ZIP Code",
    labels={"score": "Home Safety Score", "zip": "ZIP Code"},
)
fig.update_layout(yaxis={"categoryorder": "total ascending"})
fig.write_html("scores_interactive.html")
fig.show()

Saving Results to CSV

def save_reports_csv(df: pd.DataFrame, filename: str = "zipcheckup_results.csv"):
    """Save the DataFrame to a CSV file."""
    df.to_csv(filename, index=False)
    print(f"Saved {len(df)} records to {filename}")


save_reports_csv(df, "chicago_water_quality.csv")

The CSV output:

zip,score,grade,violations,lead_level,radon_zone,flood_claims,comp_water_quality,comp_lead_risk,comp_radon_risk,comp_flood_risk,comp_air_quality
60601,65,C,5,low,2,312,62,80,58,45,68
60614,72,B,3,low,2,148,78,85,62,55,71
...

Example: Which California ZIPs Have the Worst Lead Levels?

Here's a complete script that fetches data for California ZIP codes and identifies the highest lead risk areas:

import time
import requests
import pandas as pd

API_BASE = "https://zipcheckup.com/api/v1"

# Sample of California ZIP codes (major cities)
ca_zips = [
    "90001", "90012", "90210", "90401",  # Los Angeles area
    "92101", "92126",                     # San Diego
    "94102", "94110", "94158",            # San Francisco
    "95814",                               # Sacramento
    "93701",                               # Fresno
    "92374",                               # Redlands
    "93305",                               # Bakersfield
    "95202",                               # Stockton
]


def fetch_california_data(zips: list[str]) -> pd.DataFrame:
    """Fetch and analyze California ZIP codes."""
    results = []

    for i, z in enumerate(zips):
        try:
            resp = requests.get(f"{API_BASE}/zip/{z}", timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                results.append({
                    "zip": data["zip"],
                    "score": data["score"],
                    "grade": data["grade"],
                    "lead_level": data["lead_level"],
                    "lead_risk_score": data["components"]["lead_risk"],
                    "violations": data["violations"],
                    "water_quality": data["components"]["water_quality"],
                })
                print(f"[{i+1}/{len(zips)}] {z}: lead_risk={data['components']['lead_risk']}")
            if resp.status_code == 429:
                print("Rate limit — stopping")
                break
        except Exception as e:
            print(f"Error for {z}: {e}")

        time.sleep(0.5)

    return pd.DataFrame(results)


# Fetch data
df_ca = fetch_california_data(ca_zips)

# Sort by lead risk (worst first)
df_ca_sorted = df_ca.sort_values("lead_risk_score", ascending=True)

print("\n=== California ZIPs by Lead Risk (worst first) ===")
print(df_ca_sorted[["zip", "lead_level", "lead_risk_score", "violations"]].to_string(index=False))

# Summary stats
print(f"\nTotal ZIPs analyzed: {len(df_ca)}")
print(f"High lead level: {len(df_ca[df_ca['lead_level'] == 'high'])}")
print(f"Average lead risk score: {df_ca['lead_risk_score'].mean():.1f}")
print(f"Worst ZIP: {df_ca_sorted.iloc[0]['zip']} (lead risk: {df_ca_sorted.iloc[0]['lead_risk_score']})")

# Save to CSV
df_ca.to_csv("california_lead_analysis.csv", index=False)
print("\nSaved to california_lead_analysis.csv")

Next Steps

  • Full API reference: See all available endpoints, query parameters, and response fields at /api/docs/
  • State-level data: Use /api/v1/state/{code} to get aggregate statistics for an entire state
  • Rankings: Use /api/v1/rankings/{metric} to get national or state-level rankings
  • Embed widgets: Don't want to code? Embed a score card on your website with a single line of HTML
  • Automate: Connect to Zapier/Make for no-code data pipelines

Rate limits: The free tier allows 100 requests per day. For larger batch analysis, contact [email protected] about Pro tier access (10,000 requests/day).