Notification System & URL Shortener

import itertools
from abc import ABC, abstractmethod
from concurrent.futures import ThreadPoolExecutor, wait
from dataclasses import dataclass
from enum import Enum
from typing import Optional

# ---------- enums ----------
class NotificationType(Enum):
    TRANSACTIONAL = "TRANSACTIONAL"
    PROMOTIONAL = "PROMOTIONAL"
    INFO = "INFO"

class Priority(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3

# ---------- the message (many optional fields -> Builder) ----------
@dataclass(frozen=True)
class Notification:
    id: int
    message: str
    user_id: Optional[str] = None        # promotional broadcasts may have no user
    type: NotificationType = NotificationType.INFO
    priority: Priority = Priority.MEDIUM
    scheduled_at: Optional[float] = None  # epoch seconds; None = send now

class NotificationBuilder:
    _ids = itertools.count(1)

    def __init__(self, message: str):
        self._fields = {"message": message}

    def to(self, user_id: str):
        self._fields["user_id"] = user_id
        return self

    def of_type(self, t: NotificationType):
        self._fields["type"] = t
        return self

    def with_priority(self, p: Priority):
        self._fields["priority"] = p
        return self

    def build(self) -> Notification:
        return Notification(id=next(self._ids), **self._fields)

# ---------- channels ----------
class NotificationChannel(ABC):
    @abstractmethod
    def send_notification(self, n: Notification) -> None: ...

class SmsChannel(NotificationChannel):
    def send_notification(self, n):
        print(f"[SMS]   to {n.user_id}: {n.message}")

class EmailChannel(NotificationChannel):
    def send_notification(self, n):
        print(f"[EMAIL] to {n.user_id}: {n.message}")

class WhatsAppChannel(NotificationChannel):
    def send_notification(self, n):
        print(f"[WA]    to {n.user_id}: {n.message}")

class NotificationChannelFactory:
    """One place that knows how to create every channel."""
    _registry = {"SMS": SmsChannel, "EMAIL": EmailChannel, "WHATSAPP": WhatsAppChannel}

    @classmethod
    def get(cls, name: str) -> NotificationChannel:
        return cls._registry[name]()

# ---------- repository (not a bare dict inside the service) ----------
class UserPreferenceRepository:
    def __init__(self):
        self._prefs: dict[str, set[str]] = {}    # stands in for a DB table

    def save(self, user_id: str, channels: set[str]) -> None:
        self._prefs[user_id] = channels

    def get(self, user_id: str) -> list[NotificationChannel]:
        names = sorted(self._prefs.get(user_id, {"EMAIL"}))   # default: email
        return [NotificationChannelFactory.get(name) for name in names]

# ---------- dispatching strategy (the fan-out) ----------
class NotificationDispatchStrategy(ABC):
    @abstractmethod
    def dispatch(self, n: Notification, channels: list[NotificationChannel]) -> None: ...

class SyncDispatcher(NotificationDispatchStrategy):
    def dispatch(self, n, channels):
        for channel in channels:          # waits for each channel before the next
            channel.send_notification(n)

class AsyncDispatcher(NotificationDispatchStrategy):
    def __init__(self, workers: int = 5):
        self._pool = ThreadPoolExecutor(max_workers=workers)   # Java: Executors.newFixedThreadPool(5)

    def dispatch(self, n, channels):
        futures = [self._pool.submit(c.send_notification, n) for c in channels]
        wait(futures)                     # demo only; a real service would just return

# ---------- facade the source systems call ----------
class NotificationService:
    def __init__(self, prefs: UserPreferenceRepository,
                 strategy: NotificationDispatchStrategy):
        self.prefs, self.strategy = prefs, strategy

    def send_notification(self, n: Notification) -> None:
        channels = self.prefs.get(n.user_id)          # 1) who wants what
        self.strategy.dispatch(n, channels)           # 2) fan out

# usage
repo = UserPreferenceRepository()
repo.save("u1", {"SMS", "EMAIL"})
service = NotificationService(repo, AsyncDispatcher())
otp = (NotificationBuilder("Your OTP is 4821").to("u1")
       .of_type(NotificationType.TRANSACTIONAL).with_priority(Priority.HIGH).build())
service.send_notification(otp)
# [EMAIL] to u1: Your OTP is 4821
# [SMS]   to u1: Your OTP is 4821      (order may vary - they run in parallel)

# Extension (named in class as out-of-scope): retry with the Decorator pattern.
# Uses NotificationChannel / Notification / EmailChannel from the code above.
import time

class RetryingChannel(NotificationChannel):
    """Wraps ANY channel and adds retries. Same interface, extra behaviour."""

    def __init__(self, inner: NotificationChannel, attempts: int = 3, backoff: float = 0.1):
        self.inner, self.attempts, self.backoff = inner, attempts, backoff

    def send_notification(self, n: Notification) -> None:
        for attempt in range(1, self.attempts + 1):
            try:
                return self.inner.send_notification(n)      # delegate to the real channel
            except Exception as err:
                print(f"attempt {attempt} failed: {err}")
                if attempt == self.attempts:
                    raise                                    # give up: caller decides (DLQ, log...)
                time.sleep(self.backoff * 2 ** (attempt - 1))  # exponential backoff

class FlakyEmail(EmailChannel):          # fails twice, then works - to see the retry
    calls = 0
    def send_notification(self, n):
        FlakyEmail.calls += 1
        if FlakyEmail.calls < 3:
            raise ConnectionError("SMTP timeout")
        super().send_notification(n)

# usage
RetryingChannel(FlakyEmail()).send_notification(NotificationBuilder("Hi").to("u1").build())
# attempt 1 failed: SMTP timeout
# attempt 2 failed: SMTP timeout
# [EMAIL] to u1: Hi

import hashlib
import itertools
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass

# ---------- the unit of the system ----------
@dataclass
class UrlMapping:                      # an object, not just a dict entry
    short_code: str
    long_url: str
    click_count: int = 0

class UrlRepository:                   # stands in for the DB
    def __init__(self):
        self._by_code: dict[str, UrlMapping] = {}
        self._by_long: dict[str, UrlMapping] = {}

    def save(self, m: UrlMapping) -> None:
        self._by_code[m.short_code] = m
        self._by_long[m.long_url] = m

    def by_code(self, code: str) -> UrlMapping | None:
        return self._by_code.get(code)

    def by_long(self, url: str) -> UrlMapping | None:
        return self._by_long.get(url)

# ---------- Strategy: how to make a short code ----------
class ShortCodeStrategy(ABC):
    @abstractmethod
    def generate(self, long_url: str) -> str: ...

class HashStrategy(ShortCodeStrategy):
    def generate(self, long_url: str) -> str:
        return hashlib.md5(long_url.encode()).hexdigest()[:7]   # collisions possible: must check

class Base62Strategy(ShortCodeStrategy):
    ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"

    def __init__(self, start: int = 100_000):
        self._counter = itertools.count(start)   # unique id per URL (a DB sequence in real life)
        self._lock = threading.Lock()

    def generate(self, long_url: str) -> str:
        with self._lock:
            n = next(self._counter)
        code = ""
        while n:
            n, r = divmod(n, 62)
            code = self.ALPHABET[r] + code
        return code

# ---------- Observer: who cares about clicks ----------
class ClickObserver(ABC):
    @abstractmethod
    def notify(self, mapping: UrlMapping) -> None: ...

class ClickCountObserver(ClickObserver):
    def notify(self, mapping):
        mapping.click_count += 1

class AnalyticsObserver(ClickObserver):
    def notify(self, mapping):
        print(f"analytics: {mapping.short_code} clicked")

# ---------- orchestrator ----------
class UrlShortenerService:
    BASE = "https://sho.rt/"

    def __init__(self, repo: UrlRepository, strategy: ShortCodeStrategy):
        self.repo, self.strategy = repo, strategy
        self.observers: list[ClickObserver] = []

    def subscribe(self, o: ClickObserver) -> None:
        self.observers.append(o)

    def shorten(self, long_url: str) -> str:
        existing = self.repo.by_long(long_url)
        if existing:                                  # same long URL -> same short URL
            return self.BASE + existing.short_code
        code = self.strategy.generate(long_url)
        self.repo.save(UrlMapping(code, long_url))
        return self.BASE + code

    def resolve(self, short_code: str) -> str:
        m = self.repo.by_code(short_code)
        if m is None:
            raise KeyError("404 - unknown short code")
        for o in self.observers:                      # fan the click event out
            o.notify(m)
        return m.long_url                             # the web layer answers 302 Location: long_url

# usage
svc = UrlShortenerService(UrlRepository(), Base62Strategy())
svc.subscribe(ClickCountObserver())
svc.subscribe(AnalyticsObserver())
short = svc.shorten("https://docs.example.com/weekend-plan?id=42")
print(short)                                          # https://sho.rt/q0U
code = short.rsplit("/", 1)[1]
print(svc.resolve(code))                              # analytics: q0U clicked / long URL
print(svc.repo.by_code(code).click_count)             # 1