MIL-003 e-mail alert (Step 2) #34

Merged
Tirsvad merged 2 commits from mil-003-email-alert into main 2026-10-28 07:44:28 +01:00
17 changed files with 545 additions and 12 deletions
+12 -2
View File
@@ -20,7 +20,7 @@ The assignment is built in four lecture steps. Each step was delivered on its ow
| Step | What it adds | Status | | Step | What it adds | Status |
| --- | --- | --- | | --- | --- | --- |
| 1 | Scrape the price from the practice page with BeautifulSoup | done | | 1 | Scrape the price from the practice page with BeautifulSoup | done |
| 2 | E-mail alert below the target price, secrets in `.env` | planned (MIL-003) | | 2 | E-mail alert below the target price, secrets in `.env` | done |
| 3 | Browser-like request headers | planned (MIL-004) | | 3 | Browser-like request headers | planned (MIL-004) |
| 4 | The live Amazon page | planned (MIL-005) | | 4 | The live Amazon page | planned (MIL-005) |
@@ -32,6 +32,16 @@ The assignment is built in four lecture steps. Each step was delivered on its ow
![Output of the bot: title and price as a float](docs/images/step-1-run.png) ![Output of the bot: title and price as a float](docs/images/step-1-run.png)
### Step 2: e-mail alert below the target
`check_price` in `src/amazon_price_tracker/application/alert.py` compares the price with `TARGET_PRICE` (100) from `constants.py`. When it is lower, `SmtpNotifier` (`infrastructure/mailer.py`) connects to your SMTP server on port 587, starts TLS, logs in and sends a message with the title, the price and the link to buy. The SMTP address, e-mail address and password come from your own `.env` through `python-dotenv`, so they never appear in the code. The practice page costs 99.99, so a target of 100 triggers the alert.
The screenshots below come from a run where the SMTP connection is replaced by a recording stub, so no mail is sent; the second image is rendered from the message the code builds.
![Run with a stubbed SMTP connection](docs/images/step-2-run.png)
![The alert message](docs/images/step-2-email.png)
Repository layout: Repository layout:
| Path | What it is for | | Path | What it is for |
@@ -96,7 +106,7 @@ Gmail needs 2-Step Verification and an App Password; use the App Password as `EM
## ▶️ Run ## ▶️ Run
The runnable steps are added milestone by milestone (see the table above). After step 1 the bot starts with: Fill in your `.env` first (see Setup), then start the bot. It prints the title and the price and sends the alert when the price is below the target:
```bash ```bash
python -m amazon_price_tracker python -m amazon_price_tracker
+1 -1
View File
@@ -15,7 +15,7 @@ document of a type. `Primary File` may contain a glob (e.g.
| SA | Stakeholder Analysis | docs/stakeholder-analysis.md | 002 | | SA | Stakeholder Analysis | docs/stakeholder-analysis.md | 002 |
| PP | Project Plan | docs/project-plan.md | 002 | | PP | Project Plan | docs/project-plan.md | 002 |
| MIL | Milestone / Gateway | docs/milestones/*.md | 007 | | MIL | Milestone / Gateway | docs/milestones/*.md | 007 |
| RC | SQA Review Record | docs/sqa/reviews/rc-*.md | 012 | | RC | SQA Review Record | docs/sqa/reviews/rc-*.md | 013 |
## Languages ## Languages
Binary file not shown.

After

Width:  |  Height:  |  Size: 14 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 16 KiB

+62
View File
@@ -0,0 +1,62 @@
# RC-012 Code review of MIL-003
## Metadata
| Key | Value |
| --- | --- |
| ID | RC-012 |
| CrossReference | [MIL-003], [QC-PY-001] |
## Version History
| Date | Status | Author | Reviewer | Change | Commit |
| --- | --- | --- | --- | --- | --- |
| 2026-10-28 | Proposed | Jens Tirsvad Nielsen | AI as stand-in for S01 | Initial version | [07b46f0] |
---
## Artifact Under Review
- Instance reviewed: [MIL-003]
- Checklist used: [QC-PY-001]
- Scope: full review
- Language and domain: n/a (technical)
- Language reviewer: none
- Mode: autonomous, not independent (AI as stand-in for S01); the author and the stand-in are both the AI working for S01
## Checklist Results
| # | Criterion | Status | Evidence/Notes |
| --- | --- | --- | --- |
| 1 | Packages, modules, functions, variables, classes and constants follow PEP 8 casing (`snake_case`, `PascalCase`, `UPPER_SNAKE`) | Pass | Modules and functions are snake_case, classes PascalCase, constants UPPER_SNAKE |
| 2 | Names state purpose in the domain's language; no unexplained abbreviations, no single-letter names outside tiny scopes | Pass | Names state purpose (price, target, headers); no abbreviations left unexplained |
| 3 | Code is produced by the project's formatter and passes its linter with no unexplained suppressions | Pass | ruff format and ruff check pass: 33 pytest tests with fake SMTP and fakes of the ports, ruff 0 findings, mypy strict 0 errors (26 files) |
| 4 | Every function and method signature is type-annotated, including `-> None` | Pass | All signatures annotated; mypy --strict passes: 33 pytest tests with fake SMTP and fakes of the ports, ruff 0 findings, mypy strict 0 errors (26 files) |
| 5 | No bare `except:`, no swallowed exceptions; specific exceptions are raised and the cause is kept (`raise ... from`) | Pass | No bare except; errors are specific and chained with `raise ... from` where a cause exists |
| 6 | No mutable default arguments and no shadowed builtins | Pass | No mutable defaults, no shadowed builtins |
| 7 | Files, locks and connections are managed with context managers | Pass | The SMTP connection is used as a context manager in SmtpNotifier.send |
| 8 | Public modules, classes and functions have docstrings that say what, not how | Pass | Public modules and functions have docstrings with Doxygen tags |
| 9 | Logging uses `logging`, not `print`; no secrets or personal data in log output | Pass | No logging of secrets: the password is excluded from the settings repr and from error texts; print is used only for the assignment output and the missing-setting message |
| 10 | Classes and operations trace to the Design Class Diagram they implement; deviations are recorded | N-A | No Design Class Diagram exists; the milestone's technical tasks need none (Needs use case: No) |
| 11 | Tests exist for new behaviour, are named for the behaviour, and do not depend on order or the network | Pass | Tests exist, are named for behaviour and use no network or real mail server: 33 pytest tests with fake SMTP and fakes of the ports, ruff 0 findings, mypy strict 0 errors (26 files) |
| 12 | Type checker runs in strict mode without errors; `Any` is justified in a comment | Pass | mypy --strict passes; no Any |
| 13 | Dependencies are declared and pinned in the project's dependency file, none unused | Pass | Dependencies are in pyproject.toml with lower bounds; none unused |
| 14 | Each piece of knowledge (a business rule, constant, format, validation or query) is defined in one place; code is merged only where it expresses the same knowledge, not where it merely looks alike | Pass | Constants live only in constants.py |
| 15 | Imports point inward: business-rule modules import no framework, database, UI or delivery-mechanism code, and there are no import cycles | Pass | Imports point inward; no cycles |
| 16 | SOLID holds: a class or module has one reason to change; new behaviour is added by extension, not by editing a type test in several places; subclasses honour the contract of their base; protocols are narrow; high-level code depends on protocols or abstract types and receives its implementations | Pass | Notifier, PageFetcher and ProductParser are narrow Protocols; AlertMessage lives in domain so ports and use case share it without a cycle |
| 17 | Source is laid out in the layers and folders of the clean architecture folder structure (`domain`, `application`, `infrastructure`, `presentation` packages under `src/`, and `unit`, `integration` and `e2e` under `tests/`), in `snake_case`, and imports between the layers point inward | Pass | domain, application and infrastructure packages; main.py is the composition root; tests mirror them under tests/unit/ |
| 18 | File and folder names are English, contain no space and no character outside letters, digits, `-`, `_` and `.`, and follow the casing of the language for files | Pass | File names are English snake_case |
## Overall Verdict
Go — all Mandatory criteria pass or are not applicable.
## Action Items
| Action | Owner | Due |
| --- | --- | --- |
| None: no Fail and no Pending criterion | S01 | 2026-10-28 |
---
[MIL-003]: ../../milestones/mil-003-email-alert.md
[QC-PY-001]: ../../../framework/qc/code/qc-programming-python.md
[07b46f0]: https://git.tirsystem.com/Tirsvad-Udemy-100-days-of-code/047-amazon-price-tracker/commit/07b46f0180266dc7880841031737ef8a5151d5e6
@@ -0,0 +1,68 @@
"""Use case: alert when the price is below the target.
@file alert.py
@brief Builds the alert message and sends it when the price is low enough.
"""
from dataclasses import dataclass
from amazon_price_tracker.application.ports import (
Notifier,
PageFetcher,
ProductParser,
)
from amazon_price_tracker.application.read_product import read_product
from amazon_price_tracker.domain.alert_message import AlertMessage
from amazon_price_tracker.domain.pricing import is_below_target
from amazon_price_tracker.domain.product import Product
## @brief Subject of every price alert.
ALERT_SUBJECT = "Amazon Price Alert!"
@dataclass(frozen=True, slots=True)
class PriceCheck:
"""Result of a price check."""
## @brief The product read from the page.
product: Product
## @brief True when an alert was sent.
alert_sent: bool
def build_message(title: str, price: float, url: str) -> AlertMessage:
"""Return the alert for a product.
@param title Product title.
@param price Current price.
@param url Link to buy the product.
@return The message with title, price and link in the body.
"""
return AlertMessage(
subject=ALERT_SUBJECT,
body=f"{title} is now ${price:.2f}\n{url}",
)
def check_price(
fetcher: PageFetcher,
parser: ProductParser,
notifier: Notifier,
url: str,
target: float,
) -> PriceCheck:
"""Send an alert when the product at url costs less than target.
@param fetcher Port that downloads the page.
@param parser Port that reads the product from the HTML.
@param notifier Port that sends the alert.
@param url Address of the product page.
@param target Price under which an alert is sent.
@return The product read and whether an alert was sent.
@throws ProductParseError When the page holds no readable product.
"""
product = read_product(fetcher, parser, url)
if not is_below_target(product.price, target):
return PriceCheck(product, alert_sent=False)
notifier.send(build_message(product.title, product.price, url))
return PriceCheck(product, alert_sent=True)
+10 -1
View File
@@ -1,11 +1,12 @@
"""Ports the use cases need from the outside. """Ports the use cases need from the outside.
@file ports.py @file ports.py
@brief Protocols for fetching a page and parsing a product from it. @brief Protocols for fetching a page, parsing a product and sending an alert.
""" """
from typing import Protocol from typing import Protocol
from amazon_price_tracker.domain.alert_message import AlertMessage
from amazon_price_tracker.domain.product import Product from amazon_price_tracker.domain.product import Product
@@ -23,3 +24,11 @@ class ProductParser(Protocol):
def parse(self, html: str) -> Product: def parse(self, html: str) -> Product:
"""Return the product described by html.""" """Return the product described by html."""
... ...
class Notifier(Protocol):
"""Sends an alert message to the user."""
def send(self, message: AlertMessage) -> None:
"""Deliver message to the user."""
...
@@ -0,0 +1,17 @@
"""The alert message.
@file alert_message.py
@brief Defines the AlertMessage value object shared by use case and port.
"""
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class AlertMessage:
"""A message to send to the user."""
## @brief Subject line.
subject: str
## @brief Plain text body.
body: str
+5 -1
View File
@@ -1,7 +1,7 @@
"""Errors of the domain. """Errors of the domain.
@file errors.py @file errors.py
@brief Exceptions raised when a product cannot be read from a page. @brief Exceptions raised when a product cannot be read or a setting is missing.
""" """
@@ -15,3 +15,7 @@ class PriceNotFoundError(ProductParseError):
class TitleNotFoundError(ProductParseError): class TitleNotFoundError(ProductParseError):
"""The page has no product title.""" """The page has no product title."""
class MissingSettingError(Exception):
"""A required setting is missing or empty."""
@@ -0,0 +1,15 @@
"""Price rules.
@file pricing.py
@brief Decides whether a price is low enough to alert on.
"""
def is_below_target(price: float, target: float) -> bool:
"""Return True when price is strictly below target.
@param price Current price.
@param target Price under which an alert is wanted.
@return True when price < target; an equal price is not below.
"""
return price < target
@@ -0,0 +1,54 @@
"""SMTP adapter for the Notifier port.
@file mailer.py
@brief Sends the alert to your own address with smtplib and STARTTLS.
"""
import smtplib
from collections.abc import Callable
from email.message import EmailMessage
from amazon_price_tracker.constants import SMTP_PORT
from amazon_price_tracker.domain.alert_message import AlertMessage
from amazon_price_tracker.infrastructure.settings import MailSettings
class SmtpNotifier:
"""Notifier port implemented with smtplib."""
def __init__(
self,
settings: MailSettings,
smtp_factory: Callable[..., smtplib.SMTP] = smtplib.SMTP,
) -> None:
"""Keep the settings and the SMTP connection factory.
@param settings Server, address and password.
@param smtp_factory Creates the connection; replaced in tests.
"""
self._settings = settings
self._smtp_factory = smtp_factory
def _build(self, message: AlertMessage) -> bytes:
email = EmailMessage()
email["Subject"] = message.subject
email["From"] = self._settings.email_address
email["To"] = self._settings.email_address
email.set_content(message.body, charset="utf-8")
return email.as_bytes()
def send(self, message: AlertMessage) -> None:
"""Send message from the sender address to the same address.
@param message Subject and body to send.
@throws smtplib.SMTPException When the server refuses the login or message.
"""
settings = self._settings
with self._smtp_factory(settings.smtp_address, port=SMTP_PORT) as connection:
connection.starttls()
connection.login(settings.email_address, settings.email_password)
connection.sendmail(
from_addr=settings.email_address,
to_addrs=settings.email_address,
msg=self._build(message),
)
@@ -0,0 +1,46 @@
"""Mail settings read from the environment.
@file settings.py
@brief Builds MailSettings from a mapping of environment variables.
"""
from collections.abc import Mapping
from dataclasses import dataclass, field
from amazon_price_tracker.constants import (
ENV_EMAIL_ADDRESS,
ENV_EMAIL_PASSWORD,
ENV_SMTP_ADDRESS,
)
from amazon_price_tracker.domain.errors import MissingSettingError
@dataclass(frozen=True, slots=True)
class MailSettings:
"""Where and how to send the alert e-mail."""
smtp_address: str
email_address: str
# Kept out of repr so the secret never reaches logs or tracebacks.
email_password: str = field(repr=False)
def _require(environ: Mapping[str, str], name: str) -> str:
value = environ.get(name, "")
if not value.strip():
raise MissingSettingError(f"Missing setting: {name}")
return value
def load_mail_settings(environ: Mapping[str, str]) -> MailSettings:
"""Return the mail settings found in environ.
@param environ Environment variables, for example os.environ.
@return The settings.
@throws MissingSettingError Naming the first missing or empty variable.
"""
return MailSettings(
smtp_address=_require(environ, ENV_SMTP_ADDRESS),
email_address=_require(environ, ENV_EMAIL_ADDRESS),
email_password=_require(environ, ENV_EMAIL_PASSWORD),
)
+25 -7
View File
@@ -1,17 +1,35 @@
"""Composition root. """Composition root.
@file main.py @file main.py
@brief Builds the adapters and prints the title and price of the product. @brief Builds the adapters, prints the product and sends a price alert.
""" """
from amazon_price_tracker.application.read_product import read_product import os
from amazon_price_tracker.constants import PRACTICE_URL import sys
from dotenv import find_dotenv, load_dotenv
from amazon_price_tracker.application.alert import check_price
from amazon_price_tracker.constants import PRACTICE_URL, TARGET_PRICE
from amazon_price_tracker.domain.errors import MissingSettingError
from amazon_price_tracker.infrastructure.http_client import RequestsPageFetcher from amazon_price_tracker.infrastructure.http_client import RequestsPageFetcher
from amazon_price_tracker.infrastructure.mailer import SmtpNotifier
from amazon_price_tracker.infrastructure.scraper import SoupProductParser from amazon_price_tracker.infrastructure.scraper import SoupProductParser
from amazon_price_tracker.infrastructure.settings import load_mail_settings
def main() -> None: def main() -> None:
"""Print the title and the price of the practice product.""" """Print the product and send an e-mail when its price is below target."""
product = read_product(RequestsPageFetcher(), SoupProductParser(), PRACTICE_URL) load_dotenv(find_dotenv(usecwd=True))
print(product.title) try:
print(product.price) settings = load_mail_settings(os.environ)
except MissingSettingError as error:
print(f"{error}. Copy .env.example to .env and fill it in.", file=sys.stderr)
return
notifier = SmtpNotifier(settings)
result = check_price(
RequestsPageFetcher(), SoupProductParser(), notifier, PRACTICE_URL, TARGET_PRICE
)
print(result.product.title)
print(result.product.price)
print("Alert sent" if result.alert_sent else "Price is not below the target")
+78
View File
@@ -0,0 +1,78 @@
from amazon_price_tracker.application.alert import (
PriceCheck,
build_message,
check_price,
)
from amazon_price_tracker.domain.alert_message import AlertMessage
from amazon_price_tracker.domain.product import Product
URL = "https://example.test/p"
class FakeFetcher:
def __init__(self) -> None:
self.urls: list[str] = []
def fetch(self, url: str) -> str:
self.urls.append(url)
return "<html></html>"
class FakeParser:
def __init__(self, product: Product) -> None:
self.product = product
def parse(self, html: str) -> Product:
return self.product
class FakeNotifier:
def __init__(self) -> None:
self.sent: list[AlertMessage] = []
def send(self, message: AlertMessage) -> None:
self.sent.append(message)
def _run(price: float) -> tuple[PriceCheck, FakeNotifier, FakeFetcher]:
notifier = FakeNotifier()
fetcher = FakeFetcher()
result = check_price(
fetcher, FakeParser(Product("Pot", price)), notifier, URL, 100.0
)
return result, notifier, fetcher
def test_build_message_contains_title_price_and_link() -> None:
message = build_message("Instant Pot", 79.5, URL)
assert message.subject == "Amazon Price Alert!"
assert "Instant Pot" in message.body
assert "$79.50" in message.body
assert URL in message.body
def test_check_price_sends_once_when_below_target() -> None:
result, notifier, fetcher = _run(99.99)
assert result.alert_sent
assert result.product == Product("Pot", 99.99)
assert fetcher.urls == [URL]
assert len(notifier.sent) == 1
assert "$99.99" in notifier.sent[0].body
def test_check_price_does_not_send_when_equal_to_target() -> None:
result, notifier, fetcher = _run(100.0)
assert not result.alert_sent
assert fetcher.urls == [URL]
assert notifier.sent == []
def test_check_price_does_not_send_when_above_target() -> None:
result, notifier, fetcher = _run(150.0)
assert not result.alert_sent
assert fetcher.urls == [URL]
assert notifier.sent == []
+13
View File
@@ -0,0 +1,13 @@
from amazon_price_tracker.domain.pricing import is_below_target
def test_is_below_target_true_when_price_lower() -> None:
assert is_below_target(99.99, 100.0)
def test_is_below_target_false_when_price_equal() -> None:
assert not is_below_target(100.0, 100.0)
def test_is_below_target_false_when_price_higher() -> None:
assert not is_below_target(100.01, 100.0)
+85
View File
@@ -0,0 +1,85 @@
from email import message_from_bytes, policy
from types import TracebackType
from typing import Any, cast
from amazon_price_tracker.constants import SMTP_PORT
from amazon_price_tracker.domain.alert_message import AlertMessage
from amazon_price_tracker.infrastructure.mailer import SmtpNotifier
from amazon_price_tracker.infrastructure.settings import MailSettings
class FakeSmtp:
def __init__(self, host: str, port: int) -> None:
self.host = host
self.port = port
self.calls: list[str] = []
self.login_args: tuple[str, str] | None = None
self.mail: dict[str, Any] = {}
self.closed = False
def __enter__(self) -> "FakeSmtp":
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
self.closed = True
def starttls(self) -> None:
self.calls.append("starttls")
def login(self, user: str, password: str) -> None:
self.calls.append("login")
self.login_args = (user, password)
def sendmail(self, from_addr: str, to_addrs: str, msg: bytes) -> None:
self.calls.append("sendmail")
self.mail = {"from": from_addr, "to": to_addrs, "msg": msg}
def _send(message: AlertMessage) -> FakeSmtp:
created: list[FakeSmtp] = []
def factory(host: str, port: int) -> FakeSmtp:
created.append(FakeSmtp(host, port))
return created[0]
settings = MailSettings("smtp.example.test", "me@example.test", "pw")
SmtpNotifier(settings, smtp_factory=cast(Any, factory)).send(message)
return created[0]
def test_send_connects_to_host_on_starttls_port() -> None:
smtp = _send(AlertMessage("Subject", "Body"))
assert smtp.host == "smtp.example.test"
assert smtp.port == SMTP_PORT == 587
assert smtp.closed
def test_send_runs_starttls_login_sendmail_in_order() -> None:
smtp = _send(AlertMessage("Subject", "Body"))
assert smtp.calls == ["starttls", "login", "sendmail"]
assert smtp.login_args == ("me@example.test", "pw")
def test_send_mails_to_own_address_with_utf8_content() -> None:
body = "Pot æøå is now $9.50\nhttps://example.test/p"
smtp = _send(AlertMessage("Amazon Price Alert!", body))
assert smtp.mail["from"] == "me@example.test"
assert smtp.mail["to"] == "me@example.test"
raw = smtp.mail["msg"]
assert isinstance(raw, bytes)
parsed = message_from_bytes(raw, policy=policy.default)
assert parsed["Subject"] == "Amazon Price Alert!"
assert parsed["From"] == parsed["To"] == "me@example.test"
content = parsed.get_content()
assert "Pot æøå" in content
assert "$9.50" in content
assert "https://example.test/p" in content
assert parsed.get_content_charset() == "utf-8"
@@ -0,0 +1,54 @@
import pytest
from amazon_price_tracker.constants import (
ENV_EMAIL_ADDRESS,
ENV_EMAIL_PASSWORD,
ENV_SMTP_ADDRESS,
)
from amazon_price_tracker.domain.errors import MissingSettingError
from amazon_price_tracker.infrastructure.settings import load_mail_settings
FULL = {
ENV_SMTP_ADDRESS: "smtp.example.test",
ENV_EMAIL_ADDRESS: "me@example.test",
ENV_EMAIL_PASSWORD: "s3cret-pw",
}
def test_load_mail_settings_reads_all_values() -> None:
settings = load_mail_settings(FULL)
assert settings.smtp_address == "smtp.example.test"
assert settings.email_address == "me@example.test"
assert settings.email_password == "s3cret-pw"
@pytest.mark.parametrize(
"name", [ENV_SMTP_ADDRESS, ENV_EMAIL_ADDRESS, ENV_EMAIL_PASSWORD]
)
def test_load_mail_settings_raises_naming_missing_variable(name: str) -> None:
environ = {key: value for key, value in FULL.items() if key != name}
with pytest.raises(MissingSettingError, match=name):
load_mail_settings(environ)
def test_load_mail_settings_raises_on_empty_value() -> None:
with pytest.raises(MissingSettingError, match=ENV_EMAIL_ADDRESS):
load_mail_settings({**FULL, ENV_EMAIL_ADDRESS: ""})
def test_load_mail_settings_names_first_missing_variable() -> None:
with pytest.raises(MissingSettingError, match=ENV_SMTP_ADDRESS):
load_mail_settings({})
def test_mail_settings_repr_hides_password() -> None:
assert "s3cret-pw" not in repr(load_mail_settings(FULL))
def test_missing_setting_error_does_not_contain_password() -> None:
with pytest.raises(MissingSettingError) as info:
load_mail_settings({**FULL, ENV_SMTP_ADDRESS: ""})
assert "s3cret-pw" not in str(info.value)