From bcf2e2cdac9aad9d8b4747172e052d67e2fba45b Mon Sep 17 00:00:00 2001 From: Matthew McCann Date: Wed, 26 Aug 2026 18:11:47 -0700 Subject: [PATCH 1/2] feat: realistic domains for the six thinnest pythonic examples (#14) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rewrites pythonic.py (and tests) where the toy domain didn't answer "what does this look like in a real system": - facade: place_order() over inventory/payment/shipping/notification, with the payment-failure rollback tested - dependency_injection: overdue-invoice reminders with clock, source, and mail seams — deterministic tests, no monkeypatching - bridge: alert/digest notifiers bridged over email/Slack/SMS transports - prototype: report-job templates via partial + dataclasses.replace - mediator: checkout form with cascading country/shipping/payment rules - abstract_factory: one report rendered through HTML/Markdown document families naive.py toys kept as-is (the diff against them is the teaching device). READMEs aligned. 236 tests; ruff/mypy --strict clean. Co-authored-by: Claude Fable 5 --- patterns/behavioral/mediator/README.md | 6 +- patterns/behavioral/mediator/pythonic.py | 82 ++++++++++---- .../mediator/tests/test_mediator.py | 41 +++++-- .../creational/abstract_factory/README.md | 7 +- .../creational/abstract_factory/pythonic.py | 65 ++++++++--- .../tests/test_abstract_factory.py | 34 ++++-- patterns/creational/prototype/README.md | 10 +- patterns/creational/prototype/pythonic.py | 64 ++++++++--- .../prototype/tests/test_prototype.py | 23 ++-- .../modern/dependency_injection/README.md | 8 +- .../modern/dependency_injection/pythonic.py | 91 ++++++++++++---- .../tests/test_dependency_injection.py | 55 +++++++--- patterns/structural/bridge/README.md | 7 +- patterns/structural/bridge/pythonic.py | 74 +++++++++---- .../structural/bridge/tests/test_bridge.py | 39 +++++-- patterns/structural/facade/README.md | 7 +- patterns/structural/facade/pythonic.py | 103 +++++++++++++++--- .../structural/facade/tests/test_facade.py | 73 ++++++++++++- 18 files changed, 598 insertions(+), 191 deletions(-) diff --git a/patterns/behavioral/mediator/README.md b/patterns/behavioral/mediator/README.md index e8716b2..27a99d4 100644 --- a/patterns/behavioral/mediator/README.md +++ b/patterns/behavioral/mediator/README.md @@ -29,8 +29,10 @@ and *only* the mediator decides who reacts. ## Pythonic solution The mediator doesn't need a Colleague base class — widgets accept a -`notify` callable, and the mediator is a small coordinator holding the -interaction rules in one readable place. +`notify` callable and hold zero rules. `pythonic.py` scales the idea to a +checkout form whose rules genuinely tangle (country restricts shipping, +shipping gates payment and changes the total): one `_recheck` method holds +every rule, and a country change cascades through the dependent fields. ## In the wild diff --git a/patterns/behavioral/mediator/pythonic.py b/patterns/behavioral/mediator/pythonic.py index 614e800..44f5c26 100644 --- a/patterns/behavioral/mediator/pythonic.py +++ b/patterns/behavioral/mediator/pythonic.py @@ -1,43 +1,81 @@ """The mediator without a Colleague hierarchy. -Widgets take a ``notify`` callable; the coordinator holds every interaction -rule in one place and the widgets hold none. +A checkout form with enough interdependent rules to *justify* a mediator: +country restricts shipping methods, shipping method gates payment options +and recomputes the total, and submit is enabled only when the whole set is +coherent. Widgets know none of it -- every rule lives in one method. """ from __future__ import annotations from collections.abc import Callable +from dataclasses import dataclass, field +SHIPPING_BY_COUNTRY = { + "CA": {"standard": 900, "express": 2400}, + "US": {"standard": 700, "express": 1900}, + "DE": {"standard": 1100}, # no express lane +} +#: cash-on-delivery is only offered on express shipments +PAYMENTS_BY_SHIPPING: dict[str, tuple[str, ...]] = { + "standard": ("card",), + "express": ("card", "cod"), +} -class TextField: - def __init__(self, notify: Callable[[], None]) -> None: - self.text = "" - self._notify = notify - def type_text(self, text: str) -> None: - self.text = text - self._notify() +@dataclass +class Field: + """A dumb widget: holds a value, reports changes. No rules.""" + notify: Callable[[], None] + value: str = "" -class SignupForm: - """The mediator: rules in one readable method.""" + def set(self, value: str) -> None: + self.value = value + self.notify() - def __init__(self) -> None: - self.username = TextField(self._recheck) - self.password = TextField(self._recheck) - self.submit_enabled = False + +@dataclass +class CheckoutForm: + """The mediator: every cross-field rule, in one readable place.""" + + cart_cents: int + country: Field = field(init=False) + shipping: Field = field(init=False) + payment: Field = field(init=False) + shipping_options: tuple[str, ...] = () + payment_options: tuple[str, ...] = () + total_cents: int = 0 + submit_enabled: bool = False + + def __post_init__(self) -> None: + self.country = Field(self._recheck) + self.shipping = Field(self._recheck) + self.payment = Field(self._recheck) + self._recheck() def _recheck(self) -> None: - self.submit_enabled = bool(self.username.text) and len(self.password.text) >= 8 + lanes = SHIPPING_BY_COUNTRY.get(self.country.value, {}) + self.shipping_options = tuple(lanes) + if self.shipping.value not in lanes: + self.shipping.value = "" # country change invalidated the lane + self.payment_options = PAYMENTS_BY_SHIPPING.get(self.shipping.value, ()) + if self.payment.value not in self.payment_options: + self.payment.value = "" + self.total_cents = self.cart_cents + lanes.get(self.shipping.value, 0) + self.submit_enabled = bool( + self.country.value and self.shipping.value and self.payment.value + ) def main() -> None: - form = SignupForm() - form.username.type_text("ada") - form.password.type_text("short") - print(f"weak password: {form.submit_enabled}") - form.password.type_text("correcthorse") - print(f"valid form: {form.submit_enabled}") + form = CheckoutForm(cart_cents=5000) + form.country.set("CA") + form.shipping.set("express") + form.payment.set("cod") + print(f"total {form.total_cents}, submit={form.submit_enabled}") + form.country.set("DE") # express vanishes; dependent fields reset + print(f"after DE: shipping={form.shipping.value!r}, submit={form.submit_enabled}") if __name__ == "__main__": diff --git a/patterns/behavioral/mediator/tests/test_mediator.py b/patterns/behavioral/mediator/tests/test_mediator.py index 460bd5d..009520f 100644 --- a/patterns/behavioral/mediator/tests/test_mediator.py +++ b/patterns/behavioral/mediator/tests/test_mediator.py @@ -19,18 +19,39 @@ def test_weak_password_keeps_submit_disabled(self) -> None: class TestPythonic: - def test_form_coordination(self) -> None: - form = pythonic.SignupForm() - form.username.type_text("ada") - form.password.type_text("correcthorse") + def test_happy_path_enables_submit_and_totals(self) -> None: + form = pythonic.CheckoutForm(cart_cents=5000) + form.country.set("CA") + form.shipping.set("express") + form.payment.set("cod") assert form.submit_enabled - - def test_widgets_know_no_rules(self) -> None: - # A TextField is reusable with any notify callable -- no form coupling. + assert form.total_cents == 5000 + 2400 + + def test_country_change_cascades_through_dependent_fields(self) -> None: + form = pythonic.CheckoutForm(cart_cents=5000) + form.country.set("US") + form.shipping.set("express") + form.payment.set("cod") + form.country.set("DE") # DE has no express -> shipping and payment reset + assert form.shipping.value == "" and form.payment.value == "" + assert not form.submit_enabled + assert form.shipping_options == ("standard",) + + def test_payment_options_follow_shipping_method(self) -> None: + form = pythonic.CheckoutForm(cart_cents=1000) + form.country.set("CA") + form.shipping.set("standard") + standard_options: tuple[str, ...] = form.payment_options + assert standard_options == ("card",) + form.shipping.set("express") + express_options: tuple[str, ...] = form.payment_options + assert express_options == ("card", "cod") + + def test_widgets_hold_no_rules(self) -> None: pings: list[str] = [] - field = pythonic.TextField(lambda: pings.append("changed")) - field.type_text("x") - assert pings == ["changed"] + widget = pythonic.Field(notify=lambda: pings.append("changed")) + widget.set("anything") + assert pings == ["changed"] # reusable with any coordinator class TestRealWorld: diff --git a/patterns/creational/abstract_factory/README.md b/patterns/creational/abstract_factory/README.md index bf635fe..1c02510 100644 --- a/patterns/creational/abstract_factory/README.md +++ b/patterns/creational/abstract_factory/README.md @@ -29,9 +29,10 @@ factory per family, and client code programmed against the interface. ## Pythonic solution Classes and functions are first-class, so the guide's advice is: accept -*callables*. `pythonic.py` passes `Decimal` itself as the number factory; the -"complete" factory bundling several builders is just a small dataclass of -callables — no abstract base required. +*callables*. `pythonic.py` renders one sales report through interchangeable +document families (HTML for the web app, Markdown for the CLI) — each family +a dataclass of builder callables that belong together, no abstract base +required. ## In the wild diff --git a/patterns/creational/abstract_factory/pythonic.py b/patterns/creational/abstract_factory/pythonic.py index 85e34bc..d289909 100644 --- a/patterns/creational/abstract_factory/pythonic.py +++ b/patterns/creational/abstract_factory/pythonic.py @@ -1,41 +1,70 @@ """What to write instead: factories are callables, families are dataclasses. -``Decimal`` itself is already a factory -- pass it. When several builders -travel together, bundle them in a plain dataclass; swapping the family is -constructing a different bundle. +The real shape: a report renderer that must emit HTML for the web app and +Markdown for the CLI -- three builders that must stay consistent with each +other (heading, table, callout). Each family is a dataclass of callables; +the renderer never names a concrete format. """ from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass -from decimal import Decimal -def parse_numbers(texts: list[str], build: Callable[[str], object] = float) -> list[object]: - """A factory is just an argument with a sensible default.""" - return [build(t) for t in texts] +@dataclass(frozen=True) +class DocumentFamily: + """The 'complete' abstract factory: builders that belong together.""" + heading: Callable[[str], str] + table: Callable[[list[str], list[list[str]]], str] + callout: Callable[[str], str] -@dataclass(frozen=True) -class Family: - """The 'complete' abstract factory: a bundle of callables.""" - number: Callable[[str], object] - sequence: Callable[[list[object]], object] +def _html_table(headers: list[str], rows: list[list[str]]) -> str: + head = "".join(f"{h}" for h in headers) + body = "".join("" + "".join(f"{c}" for c in row) + "" for row in rows) + return f"{head}{body}
" + + +def _md_table(headers: list[str], rows: list[list[str]]) -> str: + lines = [ + "| " + " | ".join(headers) + " |", + "|" + "---|" * len(headers), + *("| " + " | ".join(row) + " |" for row in rows), + ] + return "\n".join(lines) + +HTML = DocumentFamily( + heading=lambda text: f"

{text}

", + table=_html_table, + callout=lambda text: f'
{text}
', +) -PYTHON_FAMILY = Family(number=float, sequence=list) -EXACT_FAMILY = Family(number=Decimal, sequence=tuple) +MARKDOWN = DocumentFamily( + heading=lambda text: f"## {text}", + table=_md_table, + callout=lambda text: f"> {text}", +) -def parse(texts: list[str], family: Family = PYTHON_FAMILY) -> object: - return family.sequence([family.number(t) for t in texts]) +def render_sales_report(family: DocumentFamily, rows: list[list[str]]) -> str: + """The client: builds a whole document without naming a format.""" + return "\n".join( + [ + family.heading("Sales by region"), + family.table(["region", "revenue"], rows), + family.callout("Figures exclude refunds."), + ] + ) def main() -> None: - print(parse_numbers(["1.1"], Decimal)) - print(parse(["1.1", "2.2"], EXACT_FAMILY)) + rows = [["west", "$12k"], ["east", "$9k"]] + print(render_sales_report(MARKDOWN, rows)) + print() + print(render_sales_report(HTML, rows)) if __name__ == "__main__": diff --git a/patterns/creational/abstract_factory/tests/test_abstract_factory.py b/patterns/creational/abstract_factory/tests/test_abstract_factory.py index e3e869e..f9e6751 100644 --- a/patterns/creational/abstract_factory/tests/test_abstract_factory.py +++ b/patterns/creational/abstract_factory/tests/test_abstract_factory.py @@ -1,6 +1,7 @@ """Behavioral tests for all three abstract-factory variants.""" from decimal import Decimal +from typing import ClassVar from patterns.creational.abstract_factory import naive, pythonic, real_world @@ -14,16 +15,29 @@ def test_client_builds_through_the_interface(self) -> None: class TestPythonic: - def test_callable_is_the_factory(self) -> None: - assert pythonic.parse_numbers(["2.5"], Decimal) == [Decimal("2.5")] - - def test_default_factory(self) -> None: - assert pythonic.parse_numbers(["2.5"]) == [2.5] - - def test_family_bundle_swaps_every_member(self) -> None: - result = pythonic.parse(["1.1"], pythonic.EXACT_FAMILY) - assert result == (Decimal("1.1"),) - assert pythonic.parse(["1.1"]) == [1.1] + ROWS: ClassVar[list[list[str]]] = [["west", "$12k"], ["east", "$9k"]] + + def test_markdown_family_renders_consistently(self) -> None: + doc = pythonic.render_sales_report(pythonic.MARKDOWN, self.ROWS) + assert doc.startswith("## Sales by region") + assert "| west | $12k |" in doc + assert doc.endswith("> Figures exclude refunds.") + + def test_html_family_renders_consistently(self) -> None: + doc = pythonic.render_sales_report(pythonic.HTML, self.ROWS) + assert "

Sales by region

" in doc + assert "west" in doc + assert '
' in doc + + def test_client_is_format_blind(self) -> None: + # A brand-new family works without touching the renderer. + plain = pythonic.DocumentFamily( + heading=str.upper, + table=lambda headers, rows: "; ".join(",".join(r) for r in rows), + callout=lambda text: f"NB: {text}", + ) + doc = pythonic.render_sales_report(plain, self.ROWS) + assert doc.splitlines()[0] == "SALES BY REGION" class TestRealWorld: diff --git a/patterns/creational/prototype/README.md b/patterns/creational/prototype/README.md index 21ad735..304e9c8 100644 --- a/patterns/creational/prototype/README.md +++ b/patterns/creational/prototype/README.md @@ -27,11 +27,11 @@ and a registry mapping names to exemplars that get cloned on demand. ## Pythonic solution -Python doesn't need the interface, because *callables* are the interface. A -registry can hold classes, `functools.partial` objects pre-loading the -arguments, or bound methods — anything you can call to get a fresh instance. -`pythonic.py` shows the guide's recommendation: a registry of zero-argument -factories. +Python doesn't need the interface, because *callables* are the interface. +`pythonic.py` shows the guide's recommendation on a real shape — a scheduler +stamping out report jobs from `functools.partial` templates, with per-run +tweaks via `dataclasses.replace` on the frozen product. No `clone()` +anywhere. ## In the wild diff --git a/patterns/creational/prototype/pythonic.py b/patterns/creational/prototype/pythonic.py index 2280640..e1b7b10 100644 --- a/patterns/creational/prototype/pythonic.py +++ b/patterns/creational/prototype/pythonic.py @@ -1,39 +1,67 @@ """What to write instead: a registry of callables. -Classes are first-class values in Python, and ``functools.partial`` turns -"this class plus these arguments" into a zero-argument factory. The registry -stores factories; asking for a fresh instance is just calling one. +The real shape: a scheduler stamping out report jobs from preconfigured +templates. ``functools.partial`` freezes each template's settings into a +zero-argument factory; per-run tweaks come from ``dataclasses.replace`` on +the frozen product -- no clone() protocol anywhere. """ from __future__ import annotations from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import dataclass, field, replace from functools import partial -@dataclass -class Circle: - radius: int - color: str +@dataclass(frozen=True) +class ReportJob: + name: str + query: str + recipients: tuple[str, ...] + fmt: str = "pdf" + filters: tuple[str, ...] = () -#: The whole pattern: names mapped to zero-argument factories. -MENU: dict[str, Callable[[], Circle]] = { - "small-red": partial(Circle, radius=1, color="red"), - "big-blue": partial(Circle, radius=10, color="blue"), +TEMPLATES: dict[str, Callable[[], ReportJob]] = { + "nightly-sales": partial( + ReportJob, + name="nightly-sales", + query="SELECT * FROM sales WHERE day = today()", + recipients=("sales-leads@example.com",), + filters=("exclude-test-accounts",), + ), + "weekly-audit": partial( + ReportJob, + name="weekly-audit", + query="SELECT * FROM ledger WHERE week = this_week()", + recipients=("finance@example.com", "cfo@example.com"), + fmt="xlsx", + ), } -def create(name: str) -> Circle: - return MENU[name]() +def schedule(template: str, **overrides: object) -> ReportJob: + """A fresh, independently-owned job; overrides customize this run only.""" + job = TEMPLATES[template]() + return replace(job, **overrides) if overrides else job # type: ignore[arg-type] + + +@dataclass +class Scheduler: + queue: list[ReportJob] = field(default_factory=list) + + def enqueue(self, template: str, **overrides: object) -> ReportJob: + job = schedule(template, **overrides) + self.queue.append(job) + return job def main() -> None: - a = create("small-red") - b = create("small-red") - print(f"fresh instances: {a is not b}, equal config: {a == b}") - print(create("big-blue")) + scheduler = Scheduler() + scheduler.enqueue("nightly-sales") + rush = scheduler.enqueue("weekly-audit", fmt="csv") + print(f"queued: {[j.name for j in scheduler.queue]}") + print(f"per-run override, template untouched: {rush.fmt} vs {schedule('weekly-audit').fmt}") if __name__ == "__main__": diff --git a/patterns/creational/prototype/tests/test_prototype.py b/patterns/creational/prototype/tests/test_prototype.py index 2156411..9c5c9af 100644 --- a/patterns/creational/prototype/tests/test_prototype.py +++ b/patterns/creational/prototype/tests/test_prototype.py @@ -23,13 +23,22 @@ def test_mutating_a_clone_leaves_the_exemplar_alone(self) -> None: class TestPythonic: - def test_factory_menu_produces_fresh_equal_instances(self) -> None: - a, b = pythonic.create("small-red"), pythonic.create("small-red") - assert a is not b - assert a == b == pythonic.Circle(radius=1, color="red") - - def test_distinct_entries_differ(self) -> None: - assert pythonic.create("big-blue") == pythonic.Circle(radius=10, color="blue") + def test_templates_stamp_out_fresh_equal_jobs(self) -> None: + a, b = pythonic.schedule("nightly-sales"), pythonic.schedule("nightly-sales") + assert a is not b and a == b + assert a.filters == ("exclude-test-accounts",) + + def test_per_run_overrides_leave_the_template_untouched(self) -> None: + rush = pythonic.schedule("weekly-audit", fmt="csv") + assert rush.fmt == "csv" + assert pythonic.schedule("weekly-audit").fmt == "xlsx" + + def test_scheduler_queues_customized_jobs(self) -> None: + scheduler = pythonic.Scheduler() + scheduler.enqueue("nightly-sales") + scheduler.enqueue("weekly-audit", recipients=("audit@x.com",)) + assert [j.name for j in scheduler.queue] == ["nightly-sales", "weekly-audit"] + assert scheduler.queue[1].recipients == ("audit@x.com",) class TestRealWorld: diff --git a/patterns/modern/dependency_injection/README.md b/patterns/modern/dependency_injection/README.md index 9bcbb05..4932b4e 100644 --- a/patterns/modern/dependency_injection/README.md +++ b/patterns/modern/dependency_injection/README.md @@ -28,9 +28,11 @@ clock. ## Pythonic solution -Pass the collaborators in. A `Protocol` types the seam, a keyword argument -carries the production default, and a test hands in a fake. No container, no -framework, no decorators. +Pass the collaborators in. `pythonic.py` is an overdue-invoice reminder +service with three seams — the clock, the invoice source, the mail transport — +each a `Protocol` or callable with a production default. Tests hand in a +frozen date and a capturing mailbox and become fully deterministic. No +container, no framework, no decorators. ## In the wild diff --git a/patterns/modern/dependency_injection/pythonic.py b/patterns/modern/dependency_injection/pythonic.py index 065edcf..893d18d 100644 --- a/patterns/modern/dependency_injection/pythonic.py +++ b/patterns/modern/dependency_injection/pythonic.py @@ -1,44 +1,95 @@ """Constructor injection with Protocol seams and production defaults. -The test hands in a frozen clock and a fake store; production changes -nothing and passes nothing. +A real service shape: overdue-invoice reminders. Three collaborators that +must be swappable in tests -- the clock, the invoice source, the mail +transport -- each behind a seam. Production passes nothing; tests pass +fakes and get deterministic behavior. """ from __future__ import annotations from collections.abc import Callable -from datetime import datetime +from dataclasses import dataclass +from datetime import date from typing import Protocol -class Store(Protocol): - def append(self, message: str) -> None: ... +@dataclass(frozen=True) +class Invoice: + number: str + customer_email: str + amount_cents: int + due: date -def wall_clock_hour() -> int: - return datetime.now().hour +class InvoiceSource(Protocol): + def unpaid(self) -> list[Invoice]: ... -class GreetingService: +class MailTransport(Protocol): + def send(self, to: str, subject: str, body: str) -> None: ... + + +class InMemoryInvoices: + """Production would wrap a database; the seam doesn't care.""" + + def __init__(self, invoices: list[Invoice] | None = None) -> None: + self._invoices = invoices or [] + + def unpaid(self) -> list[Invoice]: + return list(self._invoices) + + +class ConsoleMail: + """The production default transport (stand-in for SMTP).""" + + def send(self, to: str, subject: str, body: str) -> None: + print(f"MAIL to={to} subject={subject!r}") + + +class ReminderService: def __init__( self, - store: Store | None = None, - hour_now: Callable[[], int] = wall_clock_hour, + invoices: InvoiceSource, + mail: MailTransport | None = None, + today: Callable[[], date] = date.today, ) -> None: - self.store: Store = store if store is not None else [] - self.hour_now = hour_now + self.invoices = invoices + self.mail: MailTransport = mail if mail is not None else ConsoleMail() + self.today = today - def greet(self, name: str) -> str: - prefix = "good morning" if self.hour_now() < 12 else "good day" - message = f"{prefix}, {name}" - self.store.append(message) - return message + def send_reminders(self, grace_days: int = 3) -> list[str]: + """Remind every invoice more than grace_days overdue; return numbers.""" + reminded: list[str] = [] + for invoice in self.invoices.unpaid(): + overdue = (self.today() - invoice.due).days + if overdue > grace_days: + self.mail.send( + to=invoice.customer_email, + subject=f"Invoice {invoice.number} is {overdue} days overdue", + body=f"Please pay {invoice.amount_cents / 100:.2f}.", + ) + reminded.append(invoice.number) + return reminded def main() -> None: - print(GreetingService().greet("ada")) # production wiring: defaults - frozen = GreetingService(hour_now=lambda: 9) # test wiring: injected - print(frozen.greet("grace")) + source = InMemoryInvoices( + [ + Invoice("INV-1", "ada@example.com", 120_00, date(2026, 8, 1)), + Invoice("INV-2", "grace@example.com", 80_00, date(2026, 8, 25)), + ] + ) + # Test wiring: frozen clock, captured mail -- fully deterministic. + outbox: list[str] = [] + + class CapturingMail: + def send(self, to: str, subject: str, body: str) -> None: + outbox.append(f"{to}: {subject}") + + service = ReminderService(source, mail=CapturingMail(), today=lambda: date(2026, 8, 26)) + print(f"reminded: {service.send_reminders()}") + print(f"outbox: {outbox}") if __name__ == "__main__": diff --git a/patterns/modern/dependency_injection/tests/test_dependency_injection.py b/patterns/modern/dependency_injection/tests/test_dependency_injection.py index 453f818..34da291 100644 --- a/patterns/modern/dependency_injection/tests/test_dependency_injection.py +++ b/patterns/modern/dependency_injection/tests/test_dependency_injection.py @@ -1,30 +1,57 @@ """Behavioral tests for all three dependency-injection variants.""" +from datetime import date + from patterns.modern.dependency_injection import naive, pythonic, real_world class TestNaive: def test_works_but_depends_on_the_real_clock(self) -> None: message = naive.GreetingService().greet("ada") - # The strongest assertion possible without controlling the clock: assert message.endswith(", ada") assert message.startswith(("good morning", "good day")) +class CapturingMail: + def __init__(self) -> None: + self.outbox: list[tuple[str, str]] = [] + + def send(self, to: str, subject: str, body: str) -> None: + self.outbox.append((to, subject)) + + +def _service(mail: CapturingMail, today: date) -> pythonic.ReminderService: + source = pythonic.InMemoryInvoices( + [ + pythonic.Invoice("INV-1", "ada@example.com", 120_00, date(2026, 8, 1)), + pythonic.Invoice("INV-2", "grace@example.com", 80_00, date(2026, 8, 25)), + ] + ) + return pythonic.ReminderService(source, mail=mail, today=lambda: today) + + class TestPythonic: - def test_injected_clock_makes_behavior_deterministic(self) -> None: - morning = pythonic.GreetingService(hour_now=lambda: 9) - evening = pythonic.GreetingService(hour_now=lambda: 20) - assert morning.greet("ada") == "good morning, ada" - assert evening.greet("ada") == "good day, ada" - - def test_injected_store_observes_writes(self) -> None: - store: list[str] = [] - pythonic.GreetingService(store=store, hour_now=lambda: 9).greet("ada") - assert store == ["good morning, ada"] - - def test_production_defaults_still_work(self) -> None: - assert pythonic.GreetingService().greet("ada").endswith(", ada") + def test_frozen_clock_makes_reminders_deterministic(self) -> None: + mail = CapturingMail() + reminded = _service(mail, date(2026, 8, 26)).send_reminders(grace_days=3) + assert reminded == ["INV-1"] # 25 days overdue; INV-2 inside grace + assert mail.outbox == [("ada@example.com", "Invoice INV-1 is 25 days overdue")] + + def test_grace_period_is_respected(self) -> None: + mail = CapturingMail() + reminded = _service(mail, date(2026, 8, 26)).send_reminders(grace_days=30) + assert reminded == [] and mail.outbox == [] + + def test_every_seam_is_swappable(self) -> None: + # A different source, transport, and clock -- no monkeypatching anywhere. + source = pythonic.InMemoryInvoices([]) + mail = CapturingMail() + service = pythonic.ReminderService(source, mail=mail, today=lambda: date(2026, 1, 1)) + assert service.send_reminders() == [] + + def test_production_defaults_exist(self) -> None: + service = pythonic.ReminderService(pythonic.InMemoryInvoices([])) + assert isinstance(service.mail, pythonic.ConsoleMail) class TestRealWorld: diff --git a/patterns/structural/bridge/README.md b/patterns/structural/bridge/README.md index ed16c42..b26d87f 100644 --- a/patterns/structural/bridge/README.md +++ b/patterns/structural/bridge/README.md @@ -30,9 +30,10 @@ touching the other. ## Pythonic solution Strip the ceremony and the Bridge is *composition with an injected -dependency* — which is why the verdict points there. `pythonic.py` keeps the -two axes but needs no abstract bases: the renderer is a `Protocol`, shapes -are dataclasses holding one. +dependency* — which is why the verdict points there. `pythonic.py` bridges +notifiers (alerts, digests) over delivery transports (email, Slack, SMS): +the transport is a `Protocol`, notifiers are dataclasses holding one, and +"outage alert to Slack" is a constructor call, not a class. ## In the wild diff --git a/patterns/structural/bridge/pythonic.py b/patterns/structural/bridge/pythonic.py index 7310f7a..2d27762 100644 --- a/patterns/structural/bridge/pythonic.py +++ b/patterns/structural/bridge/pythonic.py @@ -1,41 +1,77 @@ """The Bridge without ceremony: composition plus an injected dependency. -A Protocol types the implementor side; shapes are dataclasses holding one. -Nothing here is special -- and that is the point. +The two real axes: what to say (alert severities, digest summaries) and how +to deliver it (email, Slack, SMS). M notifiers + N transports cover M x N +combinations, and "send the outage alert to Slack" is a constructor call. """ from __future__ import annotations -from dataclasses import dataclass -from typing import Protocol +from dataclasses import dataclass, field +from typing import ClassVar, Protocol -class Renderer(Protocol): - def circle(self, radius: float) -> str: ... +class Transport(Protocol): + """The implementor side of the bridge.""" + def deliver(self, recipient: str, text: str) -> None: ... -class Vector: - def circle(self, radius: float) -> str: - return f"" +@dataclass +class EmailTransport: + outbox: list[str] = field(default_factory=list) -class Raster: - def circle(self, radius: float) -> str: - return f"pixels for a circle of radius {radius}" + def deliver(self, recipient: str, text: str) -> None: + self.outbox.append(f"email to {recipient}: {text}") + + +@dataclass +class SlackTransport: + posts: list[str] = field(default_factory=list) + + def deliver(self, recipient: str, text: str) -> None: + self.posts.append(f"slack {recipient}: {text}") + + +@dataclass +class SmsTransport: + MAX_LEN: ClassVar[int] = 80 + messages: list[str] = field(default_factory=list) + + def deliver(self, recipient: str, text: str) -> None: + self.messages.append(f"sms {recipient}: {text[: self.MAX_LEN]}") @dataclass(frozen=True) -class Circle: - radius: float - renderer: Renderer +class AlertNotifier: + """One abstraction; the transport is the bridged-out detail.""" + + transport: Transport + recipient: str + + def alert(self, severity: str, message: str) -> None: + self.transport.deliver(self.recipient, f"[{severity.upper()}] {message}") + + +@dataclass(frozen=True) +class DigestNotifier: + """A second abstraction on the same bridge -- no transport changes needed.""" + + transport: Transport + recipient: str - def draw(self) -> str: - return self.renderer.circle(self.radius) + def digest(self, items: list[str]) -> None: + summary = f"{len(items)} updates: " + "; ".join(items) + self.transport.deliver(self.recipient, summary) def main() -> None: - print(Circle(2.0, Vector()).draw()) - print(Circle(2.0, Raster()).draw()) + slack = SlackTransport() + email = EmailTransport() + AlertNotifier(slack, "#ops").alert("critical", "db connection pool exhausted") + DigestNotifier(email, "team@example.com").digest(["3 deploys", "1 rollback"]) + print(slack.posts) + print(email.outbox) if __name__ == "__main__": diff --git a/patterns/structural/bridge/tests/test_bridge.py b/patterns/structural/bridge/tests/test_bridge.py index fb41c1d..82c37df 100644 --- a/patterns/structural/bridge/tests/test_bridge.py +++ b/patterns/structural/bridge/tests/test_bridge.py @@ -10,16 +10,35 @@ def test_same_abstraction_different_implementations(self) -> None: class TestPythonic: - def test_injected_renderer_decides_output(self) -> None: - assert pythonic.Circle(2.0, pythonic.Vector()).draw() == "" - assert "pixels" in pythonic.Circle(2.0, pythonic.Raster()).draw() - - def test_any_duck_typed_implementor_works(self) -> None: - class Ascii: - def circle(self, radius: float) -> str: - return "o" * int(radius) - - assert pythonic.Circle(3.0, Ascii()).draw() == "ooo" + def test_one_abstraction_over_two_transports(self) -> None: + slack, email = pythonic.SlackTransport(), pythonic.EmailTransport() + pythonic.AlertNotifier(slack, "#ops").alert("critical", "disk full") + pythonic.AlertNotifier(email, "ops@x.com").alert("critical", "disk full") + assert slack.posts == ["slack #ops: [CRITICAL] disk full"] + assert email.outbox == ["email to ops@x.com: [CRITICAL] disk full"] + + def test_two_abstractions_over_one_transport(self) -> None: + slack = pythonic.SlackTransport() + pythonic.AlertNotifier(slack, "#ops").alert("warn", "slow queries") + pythonic.DigestNotifier(slack, "#ops").digest(["a", "b"]) + assert len(slack.posts) == 2 # both sides vary independently + + def test_transport_specific_behavior_stays_in_the_transport(self) -> None: + sms = pythonic.SmsTransport() + pythonic.AlertNotifier(sms, "+1555").alert("info", "x" * 200) + assert len(sms.messages[0]) <= len("sms +1555: ") + sms.MAX_LEN + + def test_any_duck_typed_transport_works(self) -> None: + class Collector: + def __init__(self) -> None: + self.seen: list[str] = [] + + def deliver(self, recipient: str, text: str) -> None: + self.seen.append(text) + + collector = Collector() + pythonic.AlertNotifier(collector, "anyone").alert("info", "hello") + assert collector.seen == ["[INFO] hello"] class TestRealWorld: diff --git a/patterns/structural/facade/README.md b/patterns/structural/facade/README.md index e9663c8..a54d183 100644 --- a/patterns/structural/facade/README.md +++ b/patterns/structural/facade/README.md @@ -28,9 +28,10 @@ resource, and the copy-paste bill comes due. ## Pythonic solution Modules are namespaces and functions are entry points, so the natural Python -facade is a *function*: `pythonic.py` wraps a fiddly multi-step text -pipeline behind one call with sensible defaults — full controls still -importable beside it. +facade is a *function*: `pythonic.py` puts `place_order()` in front of an +order-fulfillment subsystem (inventory, payment, shipping, notification) — +including the payment-failure rollback every call site used to forget. The +subsystem stays public for callers needing the full controls. ## In the wild diff --git a/patterns/structural/facade/pythonic.py b/patterns/structural/facade/pythonic.py index e713a9b..ac99167 100644 --- a/patterns/structural/facade/pythonic.py +++ b/patterns/structural/facade/pythonic.py @@ -1,39 +1,106 @@ """The pythonic facade: a module-level function with good defaults. -The subsystem (tokenize / count / rank) stays public for callers who need -the controls; ``top_words`` is the one-call common case. +The subsystem is a small order-fulfillment flow -- inventory, payment, +shipping, notification -- four calls every checkout caller used to +copy-paste, in the right order, with the right rollback. ``place_order`` +is the one-call common case; the subsystem stays public for callers who +need the full controls (partial shipments, invoice-only, etc.). """ from __future__ import annotations -import re -from collections import Counter +from dataclasses import dataclass, field -STOPWORDS = frozenset({"the", "a", "an", "and", "of", "to", "in"}) +@dataclass +class Warehouse: + stock: dict[str, int] = field(default_factory=dict) -def tokenize(text: str) -> list[str]: - return re.findall(r"[a-z']+", text.lower()) + def reserve(self, sku: str, quantity: int) -> None: + if self.stock.get(sku, 0) < quantity: + raise LookupError(f"insufficient stock for {sku}") + self.stock[sku] -= quantity + def release(self, sku: str, quantity: int) -> None: + self.stock[sku] = self.stock.get(sku, 0) + quantity -def count(words: list[str], *, drop_stopwords: bool = True) -> Counter[str]: - kept = [w for w in words if not (drop_stopwords and w in STOPWORDS)] - return Counter(kept) +@dataclass +class PaymentGateway: + charges: list[tuple[str, int]] = field(default_factory=list) + declined_cards: set[str] = field(default_factory=set) -def rank(counts: Counter[str], n: int) -> list[tuple[str, int]]: - return counts.most_common(n) + def charge(self, card: str, amount_cents: int) -> str: + if card in self.declined_cards: + raise PermissionError(f"card {card} declined") + self.charges.append((card, amount_cents)) + return f"txn-{len(self.charges)}" -def top_words(text: str, n: int = 3) -> list[tuple[str, int]]: - """The facade: the whole pipeline, one call, sensible defaults.""" - return rank(count(tokenize(text)), n) +@dataclass +class Shipping: + labels: list[str] = field(default_factory=list) + + def create_label(self, sku: str, address: str) -> str: + label = f"label-{len(self.labels) + 1}:{sku}->{address}" + self.labels.append(label) + return label + + +@dataclass +class Notifier: + sent: list[str] = field(default_factory=list) + + def confirm(self, address: str, txn: str, label: str) -> None: + self.sent.append(f"to {address}: paid {txn}, ships as {label}") + + +@dataclass(frozen=True) +class OrderResult: + transaction_id: str + shipping_label: str + + +def place_order( + warehouse: Warehouse, + gateway: PaymentGateway, + shipping: Shipping, + notifier: Notifier, + *, + sku: str, + quantity: int, + price_cents: int, + card: str, + address: str, +) -> OrderResult: + """The facade: the whole checkout dance, in the right order, with the + rollback nobody remembers to write at the call site.""" + warehouse.reserve(sku, quantity) + try: + txn = gateway.charge(card, price_cents * quantity) + except PermissionError: + warehouse.release(sku, quantity) # the step copy-paste always forgets + raise + label = shipping.create_label(sku, address) + notifier.confirm(address, txn, label) + return OrderResult(transaction_id=txn, shipping_label=label) def main() -> None: - text = "the cat and the hat and the cat in the hat" - print(f"facade: {top_words(text, 2)}") - print(f"full controls: {rank(count(tokenize(text), drop_stopwords=False), 1)}") + warehouse = Warehouse(stock={"mug": 10}) + result = place_order( + warehouse, + PaymentGateway(), + Shipping(), + Notifier(), + sku="mug", + quantity=2, + price_cents=1200, + card="4242", + address="12 Grace Ave", + ) + print(result) + print(f"stock after: {warehouse.stock}") if __name__ == "__main__": diff --git a/patterns/structural/facade/tests/test_facade.py b/patterns/structural/facade/tests/test_facade.py index b158189..c57dfc3 100644 --- a/patterns/structural/facade/tests/test_facade.py +++ b/patterns/structural/facade/tests/test_facade.py @@ -4,6 +4,8 @@ import zipfile from pathlib import Path +import pytest + from patterns.structural.facade import naive, pythonic, real_world @@ -14,13 +16,72 @@ def test_one_call_runs_the_whole_sequence(self) -> None: class TestPythonic: - def test_facade_covers_the_common_case(self) -> None: - text = "the cat and the hat and the cat in the hat" - assert pythonic.top_words(text, 2) == [("cat", 2), ("hat", 2)] + def _subsystem( + self, + ) -> tuple[pythonic.Warehouse, pythonic.PaymentGateway, pythonic.Shipping, pythonic.Notifier]: + return ( + pythonic.Warehouse(stock={"mug": 10}), + pythonic.PaymentGateway(), + pythonic.Shipping(), + pythonic.Notifier(), + ) + + def test_facade_runs_every_step_in_order(self) -> None: + warehouse, gateway, shipping, notifier = self._subsystem() + result = pythonic.place_order( + warehouse, + gateway, + shipping, + notifier, + sku="mug", + quantity=2, + price_cents=1200, + card="4242", + address="12 Grace Ave", + ) + assert warehouse.stock["mug"] == 8 + assert gateway.charges == [("4242", 2400)] + assert result.shipping_label in shipping.labels + assert notifier.sent and result.transaction_id in notifier.sent[0] + + def test_declined_payment_rolls_back_the_reservation(self) -> None: + warehouse, gateway, shipping, notifier = self._subsystem() + gateway.declined_cards.add("0000") + with pytest.raises(PermissionError): + pythonic.place_order( + warehouse, + gateway, + shipping, + notifier, + sku="mug", + quantity=3, + price_cents=1200, + card="0000", + address="x", + ) + assert warehouse.stock["mug"] == 10 # released, not leaked + assert shipping.labels == [] and notifier.sent == [] + + def test_insufficient_stock_charges_nothing(self) -> None: + warehouse, gateway, shipping, notifier = self._subsystem() + with pytest.raises(LookupError): + pythonic.place_order( + warehouse, + gateway, + shipping, + notifier, + sku="mug", + quantity=99, + price_cents=1200, + card="4242", + address="x", + ) + assert gateway.charges == [] - def test_subsystem_stays_available_for_full_control(self) -> None: - counts = pythonic.count(["the", "cat"], drop_stopwords=False) - assert counts["the"] == 1 + def test_subsystem_stays_public_for_full_control(self) -> None: + # Invoice-only flow: callers can still drive the parts directly. + gateway = pythonic.PaymentGateway() + assert gateway.charge("4242", 500) == "txn-1" class TestRealWorld: From fd800cd3517858aca189bbfb44545212f4af2d76 Mon Sep 17 00:00:00 2001 From: Matthew McCann Date: Wed, 26 Aug 2026 18:17:43 -0700 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20review-pass=20findings=20=E2=80=94?= =?UTF-8?q?=20sandbox,=20safe=5Feval,=20pickle=20warning,=20lazy=20server?= =?UTF-8?q?=20(#15)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From a full review pass (bandit + manual, report in PR): HIGH - interpreter/real_world.py: safe_eval now rejects bool constants (bool subclasses int; True + 1 no longer computes) and depth-limits nesting so hostile input gets ValueError, not RecursionError - sandbox.py: assert path.is_file() replaced with a real pre-flight FileNotFoundError check (asserts vanish under python -O); timeout path now preserves captured stderr - memento/real_world.py: explicit CWE-502 warning — pickle.loads only on snapshots this process produced; README caveat added MEDIUM - server.py: catalog/index now lazy via lru_cache accessors — imports do no disk I/O, practicing what patterns/python/global_object teaches - facade/singleton/flyweight pythonic variants: honest inline notes on the saga boundary, unguarded lazy init, and unbounded pool docs/code-review.md: reviewer standards — tool layer, written standards, severity checklist, MCP reference sources. 238 tests (2 new hostile-input tests); ruff/mypy --strict clean. Co-authored-by: Claude Fable 5 --- docs/code-review.md | 61 +++++++++++++++++++ docs/index.md | 1 + patterns/behavioral/interpreter/real_world.py | 27 +++++--- .../interpreter/tests/test_interpreter.py | 10 +++ patterns/behavioral/memento/README.md | 1 + patterns/behavioral/memento/real_world.py | 6 ++ patterns/creational/singleton/pythonic.py | 7 ++- patterns/structural/facade/pythonic.py | 3 + patterns/structural/flyweight/pythonic.py | 6 +- src/design_patterns_mcp/sandbox.py | 11 +++- src/design_patterns_mcp/server.py | 37 +++++++---- 11 files changed, 146 insertions(+), 24 deletions(-) create mode 100644 docs/code-review.md diff --git a/docs/code-review.md b/docs/code-review.md new file mode 100644 index 0000000..3d301aa --- /dev/null +++ b/docs/code-review.md @@ -0,0 +1,61 @@ +# Code review standards + +The reviewer's contract for this repo — and a reusable checklist for any +Python team. + +## Layer 0: machines argue about style, humans argue about design + +These run in CI; a human review comment about anything they cover is wasted: + +| Tool | Standard it enforces | +|---|---| +| `ruff check` + `ruff format` | PEP 8, import order, bugbear/simplify/pyupgrade rule packs — each rule documented at [docs.astral.sh/ruff/rules](https://docs.astral.sh/ruff/rules/) | +| `mypy --strict` | PEP 484 typing, no untyped defs, no implicit Any | +| `pytest` + coverage | behavior, not just "it imports" | +| `python -m design_patterns.readme_table --check` | docs can't drift from code | + +Worth adding for security-sensitive work: `bandit` (SAST) and `pip-audit` +(dependency CVEs). Note: bandit flags every `assert` (B101) — in pytest +tests that's idiomatic, not a finding. + +## Layer 1: the written standards behind the tools + +- **PEP 8** (style) · **PEP 257** (docstrings) · **PEP 20** (design sensibility) +- **Google Python Style Guide** — the most common team-level extension +- This repo's own bar: [CLAUDE.md](../CLAUDE.md) (unit template, frontmatter + schema) and [verdicts.md](verdicts.md) + +## Layer 2: what human reviewers actually check + +Severity-ordered — block on CRITICAL/HIGH, note MEDIUM: + +**CRITICAL** +- Injection: user input reaching `eval`/`exec`, SQL strings, `subprocess` with `shell=True` +- Unsafe deserialization: `pickle.loads`/`yaml.load` on data crossing a trust boundary +- Secrets in code + +**HIGH** +- `assert` as a runtime guard (vanishes under `python -O`) +- Mutable default arguments; shared mutable module state +- Swallowed exceptions (`except: pass`), or `except Exception` hiding real errors +- Resources without context managers; missing cleanup on the error path +- Thread-safety claims the code doesn't earn (unguarded lazy init, shared caches) +- Unbounded recursion/loops on user-controlled input + +**MEDIUM** +- Work at import time (I/O, big computation) — see `patterns/python/global_object` +- `isinstance` traps (`bool` passes `int` checks), `is` vs `==` on sentinels +- API honesty: docstrings/comments that promise more than the code delivers +- A design pattern where a language feature suffices — check the catalog's verdict first + +## Reference sources for reviewers (MCP) + +- **This repo's own MCP server** — `claude mcp add design-patterns -- uv run --directory python-design-patterns-mcp`. `recommend_pattern` answers "should this be a Singleton?" with python-patterns.guide's verdicts and caveats; `get_pattern` serves the reference implementation to compare against. +- **Context7 MCP** — current library/framework docs, for "is this the right API usage?" questions. +- **python-patterns.guide** — the prose authority behind this catalog's verdicts. + +## Review etiquette + +- Cite the rule or the file, not taste ("B008: mutable default" beats "I don't like this"). +- One approval pass = one severity sweep top-down; don't drip-feed. +- The author of a change never approves it. diff --git a/docs/index.md b/docs/index.md index 8cfb22f..208cff9 100644 --- a/docs/index.md +++ b/docs/index.md @@ -3,4 +3,5 @@ - [How to read this repo](how-to-read-this-repo.md) — the unit anatomy and where to start - [Verdicts](verdicts.md) — what ✅ / ⚠️ / 🔄 mean, and who decides - [MCP server](mcp.md) — connect agents to the catalog +- [Code review standards](code-review.md) — the reviewer's contract and severity checklist - [Contributing](contributing.md) — adding or improving a pattern unit diff --git a/patterns/behavioral/interpreter/real_world.py b/patterns/behavioral/interpreter/real_world.py index f288305..21b9490 100644 --- a/patterns/behavioral/interpreter/real_world.py +++ b/patterns/behavioral/interpreter/real_world.py @@ -18,18 +18,31 @@ } -def safe_eval(formula: str) -> float: - """Evaluate arithmetic like '2 * (3 + 4)'; reject everything else.""" - return _walk(ast.parse(formula, mode="eval").body) +#: Deeper than any human formula; shallower than the recursion limit, so a +#: hostile input gets a clean ValueError instead of a RecursionError crash. +MAX_DEPTH = 50 -def _walk(node: ast.expr) -> float: - if isinstance(node, ast.Constant) and isinstance(node.value, int | float): +def safe_eval(formula: str) -> float: + """Evaluate arithmetic like '2 * (3 + 4)'; reject everything else.""" + return _walk(ast.parse(formula, mode="eval").body, depth=0) + + +def _walk(node: ast.expr, depth: int) -> float: + if depth > MAX_DEPTH: + raise ValueError("expression too deeply nested") + if ( + isinstance(node, ast.Constant) + and isinstance(node.value, int | float) + and not isinstance(node.value, bool) + # bool subclasses int, and a *safe* evaluator should not quietly + # compute True + 1 -- so it is excluded explicitly. + ): return float(node.value) if isinstance(node, ast.BinOp) and type(node.op) in _BINOPS: - return _BINOPS[type(node.op)](_walk(node.left), _walk(node.right)) + return _BINOPS[type(node.op)](_walk(node.left, depth + 1), _walk(node.right, depth + 1)) if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub): - return -_walk(node.operand) + return -_walk(node.operand, depth + 1) raise ValueError(f"disallowed syntax: {ast.dump(node)[:40]}") diff --git a/patterns/behavioral/interpreter/tests/test_interpreter.py b/patterns/behavioral/interpreter/tests/test_interpreter.py index deb7f77..762f951 100644 --- a/patterns/behavioral/interpreter/tests/test_interpreter.py +++ b/patterns/behavioral/interpreter/tests/test_interpreter.py @@ -34,3 +34,13 @@ def test_attack_is_rejected_not_executed(self) -> None: def test_names_are_rejected(self) -> None: with pytest.raises(ValueError): real_world.safe_eval("x + 1") + + def test_bool_constants_are_rejected(self) -> None: + # bool subclasses int; a safe evaluator must not compute True + 1. + with pytest.raises(ValueError, match="disallowed"): + real_world.safe_eval("True + 1") + + def test_hostile_nesting_gets_a_clean_error_not_a_crash(self) -> None: + bomb = "1" + " + 1" * 200 # deeper than MAX_DEPTH + with pytest.raises(ValueError, match="deeply nested"): + real_world.safe_eval(bomb) diff --git a/patterns/behavioral/memento/README.md b/patterns/behavioral/memento/README.md index e1da153..423cf61 100644 --- a/patterns/behavioral/memento/README.md +++ b/patterns/behavioral/memento/README.md @@ -8,6 +8,7 @@ symptoms: ["undo", "checkpoint and rollback", "save game", "restore previous sta verdict: use-with-care caveats: - "Immutable state makes the pattern nearly free: a snapshot is just keeping the old object. Design the state to be frozen and mementos fall out." + - "pickle.loads executes code while deserializing — only unpickle snapshots your own process produced; use JSON for anything crossing a trust boundary." - "Deep-copying big mutable graphs per keystroke is the naive cost; snapshot the smallest state that matters." stdlib_sightings: [copy.deepcopy, pickle.dumps, dataclasses.replace] --- diff --git a/patterns/behavioral/memento/real_world.py b/patterns/behavioral/memento/real_world.py index c17c7c0..740780a 100644 --- a/patterns/behavioral/memento/real_world.py +++ b/patterns/behavioral/memento/real_world.py @@ -2,6 +2,11 @@ dumps() produces an opaque snapshot; loads() restores an equivalent object -- checkpoint/rollback for anything picklable. + +SECURITY: ``pickle.loads`` executes code during deserialization. Only ever +unpickle snapshots your own process produced and stored somewhere untrusted +input cannot reach (CWE-502). For snapshots that cross a trust boundary, +serialize explicit state as JSON instead. """ from __future__ import annotations @@ -21,6 +26,7 @@ def checkpoint(game: Game) -> bytes: def rollback(snapshot: bytes) -> Game: + # Safe ONLY because `snapshot` came from checkpoint() in this process. restored = pickle.loads(snapshot) assert isinstance(restored, Game) return restored diff --git a/patterns/creational/singleton/pythonic.py b/patterns/creational/singleton/pythonic.py index 8764022..0d2d01b 100644 --- a/patterns/creational/singleton/pythonic.py +++ b/patterns/creational/singleton/pythonic.py @@ -30,7 +30,12 @@ def log(self, message: str) -> None: def get_logger() -> Logger: - """Build the shared instance on first call, then keep handing it back.""" + """Build the shared instance on first call, then keep handing it back. + + Not thread-safe: two threads racing the first call can each build a + Logger (one wins the slot). Harmless for a cheap object; guard with a + threading.Lock if construction has side effects. + """ global _lazy_instance if _lazy_instance is None: _lazy_instance = Logger() diff --git a/patterns/structural/facade/pythonic.py b/patterns/structural/facade/pythonic.py index ac99167..465ab01 100644 --- a/patterns/structural/facade/pythonic.py +++ b/patterns/structural/facade/pythonic.py @@ -81,6 +81,9 @@ def place_order( except PermissionError: warehouse.release(sku, quantity) # the step copy-paste always forgets raise + # Honest boundary: a crash below this line leaves the charge captured. + # Real systems make charge/label/notify a saga (compensate on failure) + # or an idempotent retry -- the facade pattern doesn't solve that part. label = shipping.create_label(sku, address) notifier.confirm(address, txn, label) return OrderResult(transaction_id=txn, shipping_label=label) diff --git a/patterns/structural/flyweight/pythonic.py b/patterns/structural/flyweight/pythonic.py index 0e2e55b..d97caf0 100644 --- a/patterns/structural/flyweight/pythonic.py +++ b/patterns/structural/flyweight/pythonic.py @@ -18,7 +18,11 @@ def get_card(rank: str, suit: str) -> tuple[str, str]: class Card: - """The __new__ form: ``Card('9', '♥') is Card('9', '♥')``.""" + """The __new__ form: ``Card('9', '♥') is Card('9', '♥')``. + + The pool is unbounded and unsynchronized: fine for a fixed domain like + 52 cards, wrong for unbounded user-supplied keys or racing threads. + """ _pool: ClassVar[dict[tuple[str, str], Card]] = {} diff --git a/src/design_patterns_mcp/sandbox.py b/src/design_patterns_mcp/sandbox.py index a997abd..1e69529 100644 --- a/src/design_patterns_mcp/sandbox.py +++ b/src/design_patterns_mcp/sandbox.py @@ -33,6 +33,8 @@ def run_example(catalog: Catalog, pattern_id: str, variant: str) -> RunResult: if variant not in variants: raise KeyError(f"{pattern_id} has no variant {variant!r} (has: {sorted(variants)})") path = variants[variant] # resolved by the catalog, never by the caller + if not path.is_file(): # a real check, not an assert: survives python -O + raise FileNotFoundError(f"catalog names {path} but it does not exist") repo_root = pattern.path.parents[2] module = f"patterns.{pattern.group}.{pattern.slug}.{variant}" @@ -54,8 +56,13 @@ def run_example(catalog: Catalog, pattern_id: str, variant: str) -> RunResult: ) except subprocess.TimeoutExpired as exc: out = exc.stdout.decode() if isinstance(exc.stdout, bytes) else (exc.stdout or "") - return RunResult(exit_code=-1, stdout=out[:MAX_OUTPUT_BYTES], stderr="", timed_out=True) - assert path.is_file() + err = exc.stderr.decode() if isinstance(exc.stderr, bytes) else (exc.stderr or "") + return RunResult( + exit_code=-1, + stdout=out[:MAX_OUTPUT_BYTES], + stderr=err[:MAX_OUTPUT_BYTES], + timed_out=True, + ) return RunResult( exit_code=completed.returncode, stdout=completed.stdout[:MAX_OUTPUT_BYTES], diff --git a/src/design_patterns_mcp/server.py b/src/design_patterns_mcp/server.py index 29a2077..1cab21f 100644 --- a/src/design_patterns_mcp/server.py +++ b/src/design_patterns_mcp/server.py @@ -7,6 +7,7 @@ from __future__ import annotations import argparse +from functools import lru_cache from typing import Any from mcp.server import MCPServer @@ -15,8 +16,18 @@ from design_patterns_mcp.sandbox import run_example as _run_example from design_patterns_mcp.search import SearchIndex -_catalog: Catalog = load_catalog() -_index = SearchIndex(_catalog) + +# Lazy initialization (see patterns/python/global_object): importing this +# module must not do disk I/O; the catalog loads on first use, once. +@lru_cache(maxsize=1) +def get_catalog() -> Catalog: + return load_catalog() + + +@lru_cache(maxsize=1) +def get_index() -> SearchIndex: + return SearchIndex(get_catalog()) + mcp = MCPServer( "python-design-patterns", @@ -63,7 +74,7 @@ def list_patterns(group: str | None = None, verdict: str | None = None) -> list[ """List catalog patterns, optionally filtered by group (creational, structural, behavioral, python, principle, modern) or verdict (pythonic, use-with-care, prefer-alternative).""" - patterns = _catalog.patterns + patterns = get_catalog().patterns if group is not None: patterns = tuple(p for p in patterns if p.group == group) if verdict is not None: @@ -77,9 +88,9 @@ def get_pattern(pattern_id: str, variant: str | None = None) -> dict[str, Any]: (e.g. 'structural/decorator'). variant: 'naive', 'pythonic', 'real_world', or 'all' to include example source code.""" try: - pattern = _catalog.get(pattern_id) + pattern = get_catalog().get(pattern_id) except KeyError: - known = ", ".join(_catalog.ids()) + known = ", ".join(get_catalog().ids()) raise ValueError(f"unknown pattern {pattern_id!r}; known ids: {known}") from None return _detail(pattern, variant) @@ -88,14 +99,14 @@ def get_pattern(pattern_id: str, variant: str | None = None) -> dict[str, Any]: def search_patterns(query: str, limit: int = 5) -> list[dict[str, Any]]: """Full-text search across pattern names, aliases, problems, symptoms, and prose. Returns the best matches with scores.""" - return [{**_summary(h.pattern), "score": h.score} for h in _index.search(query, limit)] + return [{**_summary(h.pattern), "score": h.score} for h in get_index().search(query, limit)] @mcp.tool() def run_example(pattern_id: str, variant: str) -> dict[str, Any]: """Execute one of a pattern's vendored example files ('naive', 'pythonic', 'real_world') in a sandboxed subprocess and return its real output.""" - result = _run_example(_catalog, pattern_id, variant) + result = _run_example(get_catalog(), pattern_id, variant) return { "exit_code": result.exit_code, "stdout": result.stdout, @@ -110,7 +121,7 @@ def recommend_pattern(problem_statement: str, limit: int = 3) -> list[dict[str, each with its caveats and verdict attached. A 'prefer-alternative' verdict means the pythonic variant shows what to write instead.""" recommendations = [] - for hit in _index.search(problem_statement, limit): + for hit in get_index().search(problem_statement, limit): p = hit.pattern rec = { **_summary(p), @@ -130,19 +141,19 @@ def recommend_pattern(problem_statement: str, limit: int = 3) -> list[dict[str, @mcp.resource("catalog://index") def catalog_index() -> str: """The whole catalog as JSON: every pattern's metadata and variants.""" - return _catalog.to_json() + return get_catalog().to_json() @mcp.resource("pattern://{group}/{slug}") def pattern_doc(group: str, slug: str) -> str: """One pattern's README prose.""" - return _catalog.get(f"{group}/{slug}").prose + return get_catalog().get(f"{group}/{slug}").prose @mcp.resource("pattern://{group}/{slug}/{variant}") def pattern_source(group: str, slug: str, variant: str) -> str: """One pattern's example source (naive | pythonic | real_world).""" - pattern = _catalog.get(f"{group}/{slug}") + pattern = get_catalog().get(f"{group}/{slug}") variants = pattern.variants() if variant not in variants: raise KeyError(f"{pattern.id} has no variant {variant!r}") @@ -152,7 +163,7 @@ def pattern_source(group: str, slug: str, variant: str) -> str: @mcp.prompt() def refactor_toward(pattern_id: str, code: str) -> str: """Ask for a refactor of the given code toward one catalog pattern.""" - pattern = _catalog.get(pattern_id) + pattern = get_catalog().get(pattern_id) caveats = "\n".join(f"- {c}" for c in pattern.caveats) return ( f"Refactor the following code toward the {pattern.name} pattern " @@ -165,7 +176,7 @@ def refactor_toward(pattern_id: str, code: str) -> str: @mcp.prompt() def explain_pattern(pattern_id: str, audience: str = "an intermediate Python developer") -> str: """Ask for an explanation of one pattern, tuned to an audience.""" - pattern = _catalog.get(pattern_id) + pattern = get_catalog().get(pattern_id) return ( f"Explain the {pattern.name} pattern to {audience}. Problem it solves: " f"{pattern.problem} Use the catalog's naive-vs-pythonic contrast, state "