Preparing your learning space...
100% through Real-World FDE Case Studies tutorials
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.
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.
Dashboards fail when they show everything and answer nothing. Start from the decisions, not the data:
Note: writing a one-page metric dictionary ("MRR = sum of active subscription amounts on the 1st, excludes trials") prevents 90% of future trust arguments.
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.
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.
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.
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.
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.)
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:
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.
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.
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.
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.
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.
A realistic 5-week timeline:
| Week | Work | Output |
|---|---|---|
| 1 | Metric dictionary, source mapping, API access (this often takes the longest — start day 1) | Signed-off scope, all credentials working |
| 2 | Stripe ingestion + backfill 13 months, raw + clean layers | Revenue trend queryable in SQL |
| 3 | HubSpot + GA4 sources, dim_customers join, data quality checks | Full dataset, first automated checks green |
| 4 | Metabase installed, dashboard built, reviewed with CEO against old spreadsheet | Dashboard matching (and explaining) the manual numbers |
| 5 | Cron scheduling, Slack alerts, runbook, handover | Weekly manual report retired |
Estimated running cost:
| Item | Math | Monthly |
|---|---|---|
| Small VM (pipeline + Postgres + Metabase) | 2 vCPU / 4GB | ~$20–40 |
| GA4 / Stripe / HubSpot APIs | included in existing plans | $0 |
| Alerts + logging | existing Slack | $0 |
| Total | ~$40/month |
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.
Save your progress and earn XP for completing tutorials.
4 questions · Pass with 70%+
1Why is idempotency important in data pipelines?
2What is the purpose of keeping raw API responses in a raw table?
3Why should dashboards start with 4-8 key metrics instead of showing everything?
4 What is the hidden cost of maintaining a dashboard?
Technology
Forward Deployed Engineer
Lesson group
Real-World FDE Case Studies
Progress
100% complete