The Brief
3 minBuild daily_report.py: a single command that fetches fresh data (API or scrape), transforms it, generates a report (Excel/PDF/JSON), emails it, and notifies a channel — designed to be run by a scheduler every morning, with logging and failure alerts. This is the capstone of the "deliver value while you sleep" arc.
- Fetch: pull today's data resiliently (Lessons 24-25).
- Transform: clean, aggregate, summarise (Lessons 15-17).
- Report: build the deliverable (Lesson 22).
- Deliver: email it + notify; schedule it; alert on failure.
The Pipeline, Drawn
5 min08:00 (cron/scheduler fires) │ ▼ [FETCH] resilient API call / scrape ──► raw data │ [TRANSFORM] clean + aggregate ──► summary │ [REPORT] xlsx + pdf + json ──► report-YYYYMMDD/ │ [DELIVER] email with attachments + Slack ping │ ▼ (any stage fails → log + alert, exit non-zero)
This is the load → transform → output pipeline from Lesson 22, extended at both ends: a resilient fetch at the front and delivery + scheduling + alerting at the back. Each stage is a function; main() wires them; failures are caught at the boundary and turned into alerts. Nothing here is new — it's the integration that's the lesson.
Build It · Fetch & Transform
14 minProject layout
daily_report/ ├── daily_report.py # the orchestrator (main) ├── fetch.py # resilient data fetch ├── transform.py # clean + aggregate ├── report.py # build xlsx/pdf/json (Lesson 22) ├── mailer.py # email (Lessons 29-30) ├── notify.py # Slack/alerts (Lessons 31-34) ├── .env # secrets (git-ignored) └── logs/ # rotating logs (Lesson 14)
Splitting into modules keeps each concern testable in isolation — and lets you reuse the mailer/notifier you already built.
Fetch (resilient)
# fetch.py import requests, logging from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry log = logging.getLogger("fetch") def _session(): retry = Retry(total=5, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], respect_retry_after_header=True) s = requests.Session() s.mount("https://", HTTPAdapter(max_retries=retry)) return s def fetch_today(url: str) -> list[dict]: s = _session() rows, page = [], 1 while True: r = s.get(url, params={"page": page, "per_page": 100}, timeout=15) r.raise_for_status() batch = r.json().get("data", []) if not batch: break rows += batch page += 1 log.info("fetched %d records", len(rows)) return rows
Transform
# transform.py from collections import defaultdict def summarise(rows: list[dict]) -> dict: by_cat = defaultdict(float) total = 0.0 for r in rows: amt = float(r.get("amount", 0) or 0) by_cat[r.get("category", "other")] += amt total += amt return {"rows": len(rows), "total": round(total, 2), "by_category": {k: round(v, 2) for k, v in by_cat.items()}}
Each module is small and pure — easy to test with sample data, no surprises at 8 a.m.
Build It · Orchestrate, Deliver, Alert
12 minThe orchestrator wires the stages, handles failure at the boundary, and returns an exit code the scheduler understands.
# daily_report.py import os, sys, logging from pathlib import Path from logging.handlers import RotatingFileHandler from datetime import datetime from dotenv import load_dotenv HERE = Path(__file__).resolve().parent # scheduler-safe paths (Lesson 36) load_dotenv(HERE / ".env") (HERE / "logs").mkdir(exist_ok=True) log = logging.getLogger("daily") log.setLevel(logging.INFO) _h = RotatingFileHandler(HERE / "logs" / "daily.log", maxBytes=1_000_000, backupCount=7) _h.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) log.addHandler(_h) log.addHandler(logging.StreamHandler()) from fetch import fetch_today from transform import summarise from report import build_report # Lesson 22, returns output folder from mailer import send_report # Lessons 29-30 from notify import alert # Lessons 31-34 def run() -> int: started = datetime.now() log.info("=== daily report run %s ===", started.isoformat()) try: rows = fetch_today(os.environ["DATA_URL"]) if not rows: log.warning("no data today — sending empty-report notice") summary = summarise(rows) out = build_report(rows, summary) # report-YYYYMMDD/ send_report(to=os.environ["REPORT_TO"].split(","), subject=f"Daily report — {started:%Y-%m-%d}", summary=summary, attachments=[str(p) for p in out.glob("*")]) alert(f"Daily report sent: {summary['rows']} rows, " f"total {summary['total']:,.2f}", "good") log.info("done in %.1fs", (datetime.now() - started).total_seconds()) return 0 except Exception as e: log.exception("daily report FAILED") alert(f"Daily report FAILED: {e}", "critical") # pages on-call return 1 if __name__ == "__main__": sys.exit(run())
2026-05-28 08:00:01 INFO === daily report run 2026-05-28T08:00:01 === 2026-05-28 08:00:04 INFO fetched 412 records 2026-05-28 08:00:05 INFO wrote summary.xlsx 2026-05-28 08:00:06 INFO report sent to 2 recipient(s), 3 attachment(s) 2026-05-28 08:00:06 INFO done in 5.2s # Slack: ✅ Daily report sent: 412 rows, total 84,210.50
# crontab — runs it every morning at 8am, scheduler-safe: 0 8 * * * /usr/bin/python3 /home/me/daily_report/daily_report.py >> /home/me/daily_report/cron.log 2>&1
Read the result
Every Level 7 skill converges here: resilient fetch (24), transform (15-17), report (22), email (29-30), notify (31-34), scheduler-safe paths + file logging (14, 36), and — crucially — a single boundary try/except that turns any failure into a critical alert and a non-zero exit. That last part is what makes it trustworthy: if the API is down at 8 a.m., you get paged instead of silently receiving no report. Schedule it once and it delivers value, or tells you why it couldn't, every single day.
Build It Yourself
13 minUse a free practice API (jsonplaceholder, a public data API) or a CSV you refresh by hand. Use the email debug server and a test Slack channel.
Get fetch → transform → report → email working end to end when run manually. Confirm the report arrives with attachments.
Make the fetch fail (point at a bad URL) and confirm: it logs the exception, fires a critical alert, and exits non-zero — while never sending a broken report.
Schedule the bot for a couple of minutes from now via cron/Task Scheduler (scheduler-safe paths!). Confirm it runs unattended, logs to the file, and delivers. Then set a real cadence.
Stretch · Make It Bulletproof
8 minAdd production hardening: (1) a heartbeat — even on a no-data day, send a "ran successfully, nothing to report" so silence always means failure; (2) a dedupe guard so two runs the same day don't double-send; (3) a --dry-run flag that does everything except actually emailing. These turn a working bot into a dependable one.
Show the key additions
import json from pathlib import Path from datetime import date MARKER = Path("logs/last_run.json") def already_ran_today() -> bool: if not MARKER.exists(): return False return json.loads(MARKER.read_text()).get("date") == date.today().isoformat() def mark_done(): MARKER.write_text(json.dumps({"date": date.today().isoformat()})) # in run(): if already_ran_today() and not args.force: log.info("already ran today — skipping"); return 0 # …on success… mark_done() # heartbeat: if no rows, still email a short "nothing today" note # --dry-run: skip send_report(), just log "would send to ..."
Non-negotiables: heartbeat on empty days, same-day dedupe (with --force override), working --dry-run.
Recap
3 minThe daily report bot integrates the whole level: resilient fetch, transform, report generation, email/notify delivery, scheduler-safe paths, file logging, and a boundary try/except that converts any failure into a critical alert + non-zero exit. Split into small modules so each stage is testable, wire them in main(), and schedule the one-shot with cron. The mindset that makes it dependable: silence must mean success — so add a heartbeat, dedupe same-day runs, and alert loudly on failure. This is "deliver value while you sleep," fully realised — and the template for the capstone in Lesson 47.
Vocabulary Card
- orchestrator
- The
mainthat wires pipeline stages and owns error handling. - heartbeat
- A success signal even on empty runs, so silence means failure.
- idempotent run
- Running twice has the same effect as once (dedupe guard).
- dry run
- Executing everything except the irreversible action, for safe testing.
Homework
4 minBuild a complete daily report bot for a data source you care about (a public API, a folder of CSVs, a site you may scrape). It must fetch, transform, generate a report, deliver (email/Slack to test endpoints), schedule, log to a file, and alert on failure. Add at least one stretch hardening (heartbeat, dedupe, or dry-run). Write a short README: what it reports, the schedule, and how failures are surfaced.
Sample · bot README
Daily Crypto Summary Bot
What: fetches yesterday's prices for 5 coins from a public API,
computes % change, builds an xlsx + pdf summary, emails it,
and pings #reports on Slack.
When: cron '0 8 * * *' (08:00 local), scheduler-safe absolute paths.
Logs: logs/daily.log (rotating, 7 days).
Failure: any stage error → logged + Slack 'critical' alert + exit 1.
Heartbeat: on a no-data day it still emails "no data today",
so a silent morning always means the bot itself failed.
Run manually: python daily_report.py [--dry-run] [--force]Non-negotiables: full fetch→deliver pipeline, scheduled, file logging, failure alerts, one hardening feature, a clear README.