-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbridge.py
More file actions
66 lines (43 loc) · 1.87 KB
/
Copy pathbridge.py
File metadata and controls
66 lines (43 loc) · 1.87 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
"""The Bridge without ceremony: composition plus an injected implementor.
Two independent axes — what to say (alert, digest) and how to deliver it
(email, Slack, SMS). The transport is a ``Protocol`` injected into dataclass
notifiers: M notifiers + N transports cover M x N combinations, and "outage
alert to Slack" is a constructor call, not a class.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import ClassVar, Protocol
class Transport(Protocol):
"""The implementor side of the bridge."""
def deliver(self, recipient: str, text: str) -> None: ...
@dataclass
class EmailTransport:
outbox: list[str] = field(default_factory=list)
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 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 digest(self, items: list[str]) -> None:
summary = f"{len(items)} updates: " + "; ".join(items)
self.transport.deliver(self.recipient, summary)