Case Study 3 — Building a Business Dashboard

An FDE case study: pull data from multiple APIs, pipe it into a database, and ship a live analytics dashboard the leadership team actually opens every morning.

Table of Contents

• The Problem
• Discovery: What Decisions Need Data
• Solution Architecture
• Step 1 — Connecting the APIs
• Step 2 — Building the Data Pipeline
• Step 3 — Modeling the Database
• Step 4 — Analytics Queries
• Step 5 — The Dashboard
• Deployment and Monitoring
• Full Database Schema
• Second and Third Sources
• Backfilling Historical Data
• Data Quality Checks
• Project Timeline and Cost

The Problem

A SaaS company's leadership team assembles a weekly revenue report by hand: export Stripe, export the CRM, export Google Analytics, paste into a spreadsheet, fix broken formulas. It takes an analyst half a day, and the numbers are always a week stale.

Worse, each source uses its own definition of "customer", so the spreadsheets never quite agree.

The goal: one dashboard, refreshed automatically, showing revenue, signups, and churn — with numbers everyone trusts.

Discovery: What Decisions Need Data

Dashboards fail when they show everything and answer nothing. Start from the decisions, not the data:

1. What decisions does this dashboard support? e.g. "Should we increase marketing spend this month?"
2. Which 5–8 metrics drive those decisions? Revenue, MRR, new signups, churn rate, CAC. Nothing else in v1.
3. Who looks at it, and how often? A CEO wants 4 big numbers; a growth analyst wants filters and date ranges.
4. What does each source actually mean? Stripe "charges" ≠ revenue (refunds, failures). Define every metric in writing with the stakeholders.

Note: writing a one-page metric dictionary ("MRR = sum of active subscription amounts on the 1st, excludes trials") prevents 90% of future trust arguments.

Solution Architecture

Stripe API ─────┐
HubSpot API ────┤
Google Analytics┼──► Ingestion script ──► Database (Postgres) ──► SQL metrics ──► Dashboard
Custom app DB ──┘        (scheduled)           (raw + clean tables)   (views)

The pattern is the classic analytics stack: extract → load → transform → visualize. Keep raw API responses in a raw table so you can re-process without re-fetching.

Step 1 — Connecting the APIs

Each source has its own client library, but the pattern is identical: authenticate, paginate, normalize.

# pip install stripe
from datetime import datetime, timedelta, timezone
import stripe

stripe.api_key = "sk_live_..."

def days_ago_timestamp(days: int) -> int:
    """Unix timestamp (seconds) for N days ago — Stripe created-filters take seconds."""
    return int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp())

def fetch_all_charges(days_back: int = 30) -> list[dict]:
    """Fetch every Stripe charge with automatic pagination."""
    charges = stripe.Charge.list(limit=100, created={"gte": days_ago_timestamp(days_back)})
    all_charges = []
    for charge in charges.auto_paging_iter():
        all_charges.append({
            "id": charge.id,
            "amount": charge.amount / 100,          # Stripe uses cents
            "currency": charge.currency,
            "status": charge.status,                # succeeded / failed / refunded
            "customer": charge.customer,
            "created": charge.created,
        })
    return all_charges

Simple explanation: the Stripe client returns paginated results; auto_paging_iter() walks all pages for you. Notice the normalization — converting cents to dollars and timestamps to a consistent format happens at ingestion, not in the dashboard.

Note: every API you touch will have quirks (timezone bugs, cents vs dollars, deleted records). Read one full response object carefully before writing any pipeline code.

Step 2 — Building the Data Pipeline

The ingestion script runs on a schedule, pulls fresh data, and writes it to the database idempotently.

import psycopg2

def load_charges(charges: list[dict], conn):
    """Upsert charges — safe to run twice for the same data."""
    with conn.cursor() as cur:
        for c in charges:
            cur.execute("""
                INSERT INTO raw_stripe_charges (id, amount, currency, status, customer, created)
                VALUES (%s, %s, %s, %s, %s, to_timestamp(%s))
                ON CONFLICT (id) DO UPDATE
                SET amount = EXCLUDED.amount,
                    status = EXCLUDED.status;   -- only mutable fields; id/customer/created are immutable facts
            """, (c["id"], c["amount"], c["currency"], c["status"], c["customer"], c["created"]))
    conn.commit()

Simple explanation: ON CONFLICT ... DO UPDATE makes the load idempotent — running the job twice (or re-running after a failure) never duplicates rows. This one habit saves more debugging time than any other in pipeline work.

Schedule it with cron, GitHub Actions, or Airflow:

# crontab: run every hour
0 * * * * /usr/bin/python3 /opt/pipeline/ingest.py >> /var/log/pipeline.log 2>&1

Cron alone reports nothing when the job dies — wrap main() in a try/except and call notify_failure("stripe", str(e)) (shown in Deployment and Monitoring) so a crashed run pages you instead of failing silently.

Step 3 — Modeling the Database

Keep two layers: raw (exactly what the APIs returned) and clean (typed, joined, business-ready).

-- Clean layer: one row per customer per month, revenue from succeeded charges only
CREATE VIEW monthly_revenue AS
SELECT
    date_trunc('month', created) AS month,
    customer,
    SUM(amount) FILTER (WHERE status = 'succeeded') AS revenue
FROM raw_stripe_charges
GROUP BY 1, 2;

-- Business metric: MRR trend for the dashboard
CREATE VIEW mrr_by_month AS
SELECT
    month,
    SUM(revenue) AS total_revenue,
    COUNT(DISTINCT customer) AS paying_customers
FROM monthly_revenue
GROUP BY month
ORDER BY month;

Simple explanation: raw tables are your safety net (you can always rebuild clean views), and views keep metric logic in one version-controlled place. When the CFO asks "is MRR including trials?", the answer is one SQL file, not a buried spreadsheet formula.

Step 4 — Analytics Queries

Go beyond "what happened" into "what changed" — deltas and cohort-style metrics are what make a dashboard useful.

-- Month-over-month growth
SELECT
    month,
    total_revenue,
    ROUND(
        100.0 * (total_revenue - LAG(total_revenue) OVER (ORDER BY month))
        / LAG(total_revenue) OVER (ORDER BY month)
    , 1) AS mom_growth_pct
FROM mrr_by_month;

-- Churn: paying customers last month who are absent this month
SELECT COUNT(*) AS churned_customers
FROM (
    SELECT DISTINCT customer FROM monthly_revenue WHERE month = date_trunc('month', CURRENT_DATE) - INTERVAL '1 month'
) prev
WHERE NOT EXISTS (
    SELECT 1 FROM monthly_revenue cur
    WHERE cur.customer = prev.customer
      AND cur.month = date_trunc('month', CURRENT_DATE)
);

Simple explanation: LAG() compares each month to the previous one without a self-join, and the anti-join pattern (NOT EXISTS) finds customers who disappeared — the honest way to count churn.

Note: the churn query is only valid for completed months — run it on the 1st for the previous month. Mid-month, "this month" is still in progress, so every active customer looks churned. (Also, LAG() returns NULL for the first month of data, which is expected.)

Step 5 — The Dashboard

Don't build a custom web app for v1. Use what already connects to Postgres: Metabase (open source), Grafana, or Google Looker Studio (free).

Layout that works for executives:

┌─────────────┬─────────────┬─────────────┬─────────────┐
│   MRR       │ New signups │   Churn %   │  MoM growth │   ← 4 big KPI cards
├─────────────┴─────────────┴─────────────┴─────────────┤
│              Revenue trend (12 months, line)          │   ← the "one chart"
├──────────────────────────┬────────────────────────────┤
│  Revenue by plan (bar)   │  Signups by source (bar)   │   ← supporting detail
└──────────────────────────┴────────────────────────────┘

Dashboard best practices:

• 4 KPI cards on top. If a decision can't be made in 10 seconds, the dashboard fails.
• One hero chart showing the trend the team cares about most.
• Filters for date range and segment — below the fold, not competing with the KPIs.
• Label freshness: show "last updated 07:00 today" so nobody wonders if the data is stale.
• Add the metric definitions as tooltips — this kills most trust arguments before they start.

Deployment and Monitoring

Run the pipeline and dashboard on a small VM or container service, and monitor the pipeline, not just the dashboard.

# end of ingest.py — a silent pipeline is your biggest risk
import requests, os

def notify_failure(source: str, error: str):
    requests.post(os.environ["SLACK_WEBHOOK"], json={
        "text": f":red_circle: Pipeline FAILED for {source}: {error}"
    })

Simple explanation: when the Stripe API changes or a token expires, the dashboard keeps showing yesterday's numbers with no error anywhere. Alert on pipeline failures (Slack/email) and on data freshness ("no new rows in 24h") so you hear about breakage before the CEO does.

A weekly 15-minute number-check against the old spreadsheet — for the first month only — is what earns the dashboard's trust and retires the manual report.

Full Database Schema

The exact DDL you'd ship for this project — raw tables, clean views, and a pipeline bookkeeping table.

-- ============ RAW LAYER (what the APIs returned, unmodified) ============

CREATE TABLE raw_stripe_charges (
    id        TEXT PRIMARY KEY,
    amount    NUMERIC,
    currency  TEXT,
    status    TEXT,
    customer  TEXT,
    created   TIMESTAMPTZ,
    _loaded_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE raw_hubspot_contacts (
    id          TEXT PRIMARY KEY,
    email       TEXT,
    lifecycle   TEXT,           -- lead / customer / evangelist ...
    signup_date TIMESTAMPTZ,
    source      TEXT,           -- organic / paid / referral
    _loaded_at  TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE raw_ga4_sessions (
    date       DATE,
    source     TEXT,
    sessions   INTEGER,
    signups    INTEGER,
    _loaded_at TIMESTAMPTZ DEFAULT now()
);

-- ============ PIPELINE BOOKKEEPING ============

CREATE TABLE pipeline_runs (
    id          SERIAL PRIMARY KEY,
    source      TEXT,
    started_at  TIMESTAMPTZ,
    finished_at TIMESTAMPTZ,
    rows_upserted INTEGER,
    status      TEXT             -- running / success / failed
);

-- ============ CLEAN LAYER (business logic lives here) ============

CREATE VIEW dim_customers AS
SELECT
    c.id,
    c.email,
    c.lifecycle,
    c.signup_date,
    c.source,
    MIN(ch.created) AS first_payment_date,
    SUM(ch.amount) FILTER (WHERE ch.status = 'succeeded') AS lifetime_value
FROM raw_hubspot_contacts c
LEFT JOIN raw_stripe_charges ch ON ch.customer = c.id
GROUP BY c.id, c.email, c.lifecycle, c.signup_date, c.source;

-- monthly_revenue + mrr_by_month views: exactly as defined in Step 3 — same DDL, don't re-create

And a small helper so every ingestion run records itself in pipeline_runs — this is the log_run used by the pipeline and the backfill:

def log_run(source: str, rows: int, status: str, started_at, conn):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO pipeline_runs (source, started_at, finished_at, rows_upserted, status)
            VALUES (%s, %s, now(), %s, %s)
        """, (source, started_at, rows, status))
    conn.commit()

Simple explanation: the pipeline_runs table is your pipeline's own diary — every run records when it started, how many rows it touched, and whether it succeeded. The dashboard gets a free "data last refreshed" widget from it, and debugging a bad night is a SELECT instead of log archaeology. Note how dim_customers joins two sources on the shared Stripe/HubSpot customer id — mapping those ids in discovery is what makes cross-source analytics possible at all.

Second and Third Sources

Stripe was the easy one. Real dashboards need the CRM and web analytics too — here's the pattern for each.

# HubSpot — paginated, rate-limited (100 req/10s)
import time, requests

# since = epoch MILLISECONDS (string) of the last successful sync watermark
def fetch_hubspot_contacts(since: str) -> list[dict]:
    url = "https://api.hubapi.com/crm/v3/objects/contacts/search"
    headers = {"Authorization": f"Bearer {HUBSPOT_TOKEN}"}
    contacts, after = [], None
    while True:
        body = {
            "filterGroups": [{"filters": [
                {"propertyName": "lastmodifieddate", "operator": "GTE", "value": since}
            ]}],
            "limit": 100, **({"after": after} if after else {})
        }
        resp = requests.post(url, headers=headers, json=body).json()
        for c in resp.get("results", []):
            props = c["properties"]
            contacts.append({
                "id": c["id"],
                "email": props.get("email"),
                "lifecycle": props.get("lifecyclestage"),
                "signup_date": props.get("createdate"),
                "source": props.get("hs_analytics_source") or "unknown",
            })
        after = resp.get("paging", {}).get("next", {}).get("after")
        if not after:
            return contacts
        time.sleep(0.2)      # stay under the rate limit

# GA4 — run a report via the Analytics Data API
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import RunReportRequest, DateRange, Dimension, Metric

def fetch_ga4_signups(days_back: int = 1) -> list[dict]:
    client = BetaAnalyticsDataClient()
    response = client.run_report(RunReportRequest(
        property_id="properties/123456",
        dimensions=[Dimension(name="date"), Dimension(name="sessionDefaultChannelGroup")],
        metrics=[Metric(name="sessions"), Metric(name="conversions")],
        date_ranges=[DateRange(start_date=f"{days_back}daysAgo", end_date="today")],  # no space — GA4 format
    ))
    return [
        {"date": row.dimension_values[0].value,
         "source": row.dimension_values[1].value,
         "sessions": int(row.metric_values[0].value),
         "signups": int(row.metric_values[1].value)}
        for row in response.rows
    ]

Simple explanation: three different APIs, three different auth schemes, three different pagination styles — but all three reduce to the same shape: fetch pages, normalize to dicts, upsert into a raw table. Once the pattern is built for Stripe, each new source is usually a half-day of work, not a new project.

Note: an incremental filter (like HubSpot's lastmodifieddate >= since) is what lets you run hourly syncs forever without hammering the APIs or paying for full re-fetches. Keep the watermark (last successful sync time) in pipeline_runs.

Backfilling Historical Data

A dashboard with 2 weeks of history is useless for trend analysis. Before launch, backfill 12+ months.

def backfill_stripe(months: int = 13):
    """Fetch in month-sized chunks: friendlier to rate limits and resumable."""
    now = datetime.now(timezone.utc)
    for m in range(months, 0, -1):
        start = now - timedelta(days=30 * m)
        end = now - timedelta(days=30 * (m - 1))
        # fetch_charges_between: same as fetch_all_charges, but with an explicit
        # created={"gte": start_ts, "lt": end_ts} filter instead of days_back
        charges = fetch_charges_between(start, end)
        load_charges(charges, conn)
        log_run("stripe_backfill", rows=len(charges), status="success",
                started_at=start, conn=conn)
        print(f"backfilled {start:%Y-%m}: {len(charges)} charges")

Simple explanation: month-sized chunks mean a failure on month 7 doesn't restart the whole job — you just resume from there. Expect backfill to surface surprises: customers who exist in Stripe but not the CRM, charges with no customer record, timezone edge cases on month boundaries. Every surprise is a data-quality conversation to have before the dashboard launches, not after the CFO asks.

Data Quality Checks

Trust is the product. Automated checks catch silent breakage before users do.

# dq_checks.py — run right after each ingestion
def run_dq_checks(conn) -> list[str]:
    failures = []
    with conn.cursor() as cur:
        # 1. Freshness: did yesterday's data arrive?
        cur.execute("""
            SELECT count(*) FROM raw_stripe_charges
            WHERE created > now() - interval '26 hours'
        """)
        if cur.fetchone()[0] == 0:
            failures.append("stripe: no charges in last 26h")

        # 2. Volume anomaly: today's row count way off the recent median?
        cur.execute("""
            SELECT count(*) FROM raw_stripe_charges
            WHERE created > now() - interval '1 day'
        """)
        today = cur.fetchone()[0]
        # expected_daily_median(conn): median daily row count over the last 30 days
        if today < 0.3 * expected_daily_median(conn):
            failures.append(f"stripe: only {today} rows today")

        # 3. Referential integrity: charges linked to unknown customers
        cur.execute("""
            SELECT count(*) FROM raw_stripe_charges ch
            LEFT JOIN raw_hubspot_contacts c ON c.id = ch.customer
            WHERE c.id IS NULL AND ch.created > now() - interval '7 days'
        """)
        orphans = cur.fetchone()[0]
        if orphans > 100:
            failures.append(f"integrity: {orphans} orphan charges (CRM sync broken?)")
    return failures

Simple explanation: these three checks catch the three ways pipelines actually die — stopped syncing (freshness), partial sync (volume anomaly), and broken joins (integrity). Any failure posts to Slack with the source name and the numbers, using the notify_failure helper from the monitoring section.

Project Timeline and Cost

A realistic 5-week timeline:

Estimated running cost:

Simple explanation: the ROI is one analyst's half-day per week recovered, plus decisions made on data that's hours old instead of a week old. The hidden cost is maintenance — APIs deprecate endpoints, so budget a few hours per quarter to keep the pipeline healthy.