Project: a price-watch alert bot

Step 2: state that survives a rerun

Jobs get rerun: a scheduler retries after a crash, or someone runs it by hand to check. A rerun must not email everyone again. The fix is to remember what was sent. Each alert gets a key, and the keys go in a JSON file:

import json
from pathlib import Path


def load_sent(path):
    p = Path(path)
    return set(json.loads(p.read_text())) if p.exists() else set()


def save_sent(path, sent):
    Path(path).write_text(json.dumps(sorted(sent), indent=2))


key = f"{member_id}:{sku}:{price_date}"     # e.g. "w1:TE-107:2025-06-30"

A set makes "already sent?" a quick key in sent check. JSON has no sets, so it's saved as a sorted list and turned back into a set when loaded. The key includes the price's date. If the price falls to the target again on a later day, that's news worth a new alert, so it gets a new key.

Save the state after the emails go out. If the job crashes halfway, the worst case is a resend of the alerts that didn't get recorded, never a silent loss. Keys that aren't recorded mean "try again".

import json
sent = {"w1:TE-107:2025-06-30", "w3:TE-107:2025-06-30"}
text = json.dumps(sorted(sent))
again = set(json.loads(text))
print(again == sent, "w1:TE-107:2025-06-30" in again, len(text) > 0)

Step 3: one digest per person

Ten alerts for one person should be one email, not ten. Collect the alerts in a dictionary keyed by member, then send each member one message with a line per alert:

from email.message import EmailMessage


def digest(member, lines):
    msg = EmailMessage()
    msg["From"] = "alerts@shop.example.com"
    msg["To"] = member["email"]
    msg["Subject"] = f"Price alerts: {len(lines)} update{'s' if len(lines) != 1 else ''}"
    first = member["name"].split()[0]
    msg.set_content(f"Hi {first},\n\n" + "\n".join(f"- {line}" for line in lines) + "\n\nThe shop team")
    return msg

Lines read like "TE-107 is now $356.25 (your target: $400.00)". A discontinued product gets its own line, "AP-127 is discontinued; try AP-138 instead", but only once: it goes into the state file like any alert, with "gone" in place of the date. All the digests then go out over one SMTP connection, as in the email lesson.

Step 4: wiring it together

The run function is the only place that knows about all three pieces. Its shape:

1. Load the sent keys, create the client, fetch the watchlists.
2. Loop over members and their watched products. Inside the loop, a try around client.latest with one except per exception keeps each product's failure local. An else block runs only when no exception happened, which is the natural home for the price rule.
3. Collect each member's new lines; make one digest for each member who has any.
4. Send all digests over one connection, but only if there's at least one, so a quiet evening never logs in to the mail server at all.
5. Save the keys and return a summary.

Returning a summary dictionary, rather than printing, makes the function easy to test and easy to log. The checks below call it twice in a row and confirm the second call does nothing new: the most important property of any scheduled job.

import requests
from collections import defaultdict

API = "https://api.gradient.test/v1"
members = requests.get(f"{API}/watchlists").json()["data"]

latest, problems = {}, {}
for sku in sorted({w["sku"] for m in members for w in m["watching"]}):
    reply = requests.get(f"{API}/prices/{sku}", params={"days": 1})
    if reply.ok:
        latest[sku] = reply.json()["prices"][-1]
    else:
        problems[sku] = (reply.status_code, reply.json()["error"].get("replaced_by"))

would_send = defaultdict(list)
for m in members:
    for w in m["watching"]:
        p = latest.get(w["sku"])
        if p and p["price"] <= w["target_price"]:
            would_send[m["id"]].append(f"{w['sku']} is now ${p['price']:.2f} (your target: ${w['target_price']:.2f})")

for member_id, lines in would_send.items():
    print(member_id, lines)
print("problems:", problems)
print(len(latest) + len(problems), "products fetched, once each")

Running it every evening

A bot like this runs on a schedule, on a computer that stays on. Common choices:

• cron on a Linux server: a line like 0 19 * * * /srv/bot/.venv/bin/python /srv/bot/run.py runs it at 19:00 every day. Write down which time zone the server uses, or the bot will run at the wrong hour.
• GitHub Actions with a schedule trigger, which needs no server of your own.
• A cloud scheduler, such as a cron job on the platform that hosts your app.

Wherever it runs, three things matter. The SMTP password and any API keys come from environment variables or the platform's secret settings. The state file lives somewhere that survives between runs: some platforms wipe local files on every deploy, so a small database is safer there. And every run writes a log line with its summary, so when a member says "I never got my alert", you can see what happened that evening.