Observer, State & Chain of Responsibility

from abc import ABC, abstractmethod


class Subscriber(ABC):
    """Anyone who wants to be a subscriber MUST have notify()."""
    @abstractmethod
    def notify(self, video: str) -> None: ...


class Avinash(Subscriber):
    def notify(self, video: str) -> None:
        print(f"Avinash is notified: {video}")


class Ajay(Subscriber):
    def notify(self, video: str) -> None:
        print(f"Ajay is notified: {video}")


class Shekhar(Subscriber):
    def notify(self, video: str) -> None:
        print(f"Shekhar is notified: {video}")


class YouTubeChannel:
    """The subject. It HAS a list of subscribers (aggregation)."""
    def __init__(self, name: str) -> None:
        self.name = name
        self._subscribers: list[Subscriber] = []

    def add_subscriber(self, s: Subscriber) -> None:      # registration
        self._subscribers.append(s)

    def remove_subscriber(self, s: Subscriber) -> None:   # unsubscribe
        self._subscribers.remove(s)

    def _notify_subscribers(self, video: str) -> None:    # private: only the channel calls it
        for s in self._subscribers:
            s.notify(video)          # same method for everyone

    def upload_video(self, video: str) -> None:           # the state change / event
        print(f"Video uploaded: {video}")
        self._notify_subscribers(video)


# usage
channel = YouTubeChannel("Pradeep Kumar")
ajay, shekhar, avinash = Ajay(), Shekhar(), Avinash()
for s in (ajay, shekhar, avinash):
    channel.add_subscriber(s)
channel.upload_video("Observer Pattern")
channel.remove_subscriber(shekhar)          # Shekhar unsubscribed
channel.upload_video("State Pattern")
# Video uploaded: Observer Pattern
# Ajay is notified: Observer Pattern
# Shekhar is notified: Observer Pattern
# Avinash is notified: Observer Pattern
# Video uploaded: State Pattern
# Ajay is notified: State Pattern
# Avinash is notified: State Pattern

from abc import ABC, abstractmethod


class Observer(ABC):
    @abstractmethod
    def notify(self, message: str) -> None: ...


class Subject(ABC):
    """Optional interface. Only needed if you have more than one kind of subject."""
    @abstractmethod
    def add_observer(self, o: Observer) -> None: ...
    @abstractmethod
    def remove_observer(self, o: Observer) -> None: ...
    @abstractmethod
    def notify_observers(self, message: str) -> None: ...


class Zomato(Subject):
    """Observable: customers watch it for item availability."""
    def __init__(self) -> None:
        self._observers: list[Observer] = []

    def add_observer(self, o: Observer) -> None:
        self._observers.append(o)

    def remove_observer(self, o: Observer) -> None:
        self._observers.remove(o)

    def notify_observers(self, message: str) -> None:
        for o in self._observers:
            o.notify(message)

    def item_available(self, item: str) -> None:   # state change
        self.notify_observers(f"{item} is available now!")


class Customer(Observer):
    """In Zomato, 'Customer' reads better than 'Observer'. Names are your choice."""
    def __init__(self, name: str) -> None:
        self.name = name

    def notify(self, message: str) -> None:
        print(f"{self.name} got: {message}")


# usage
zomato = Zomato()
zomato.add_observer(Customer("David"))
zomato.add_observer(Customer("Payal"))
zomato.item_available("Pizza")
# David got: Pizza is available now!
# Payal got: Pizza is available now!

class ATMMachine:                       # BAD: every method checks every state
    def __init__(self) -> None:
        self.state = "NO_CARD"

    def insert_card(self) -> None:
        if self.state == "NO_CARD":
            print("Card inserted."); self.state = "HAS_CARD"
        elif self.state == "HAS_CARD":
            print("Card already inserted.")
        # ...add a new state? edit EVERY method again

    def withdraw_cash(self, amount: int) -> None:
        if self.state == "NO_CARD":
            print("Insert the card first.")
        elif self.state == "HAS_CARD":
            print(f"Dispensing {amount}.")

from __future__ import annotations
from abc import ABC, abstractmethod


class ATMState(ABC):
    """All the 'buttons' of the ATM. Every state must say how each behaves."""
    def __init__(self, atm: ATMMachine) -> None:
        self.atm = atm              # needed to change the machine's state

    @abstractmethod
    def insert_card(self) -> None: ...
    @abstractmethod
    def eject_card(self) -> None: ...
    @abstractmethod
    def withdraw_cash(self, amount: int) -> None: ...


class NoCardState(ATMState):
    def insert_card(self) -> None:
        print("Card inserted successfully.")
        self.atm.set_state(self.atm.has_card_state)    # transition

    def eject_card(self) -> None:
        print("No card to eject.")

    def withdraw_cash(self, amount: int) -> None:
        print("Insert the card first.")


class HasCardState(ATMState):
    def insert_card(self) -> None:
        print("Card already inserted.")

    def eject_card(self) -> None:
        print("Card ejected.")
        self.atm.set_state(self.atm.no_card_state)     # transition

    def withdraw_cash(self, amount: int) -> None:
        print(f"Dispensing {amount}.")


class ATMMachine:
    """Context. Knows all its states, holds the current one, forwards every call."""
    def __init__(self) -> None:
        self.no_card_state = NoCardState(self)       # created once, reused
        self.has_card_state = HasCardState(self)
        self.current_state: ATMState = self.no_card_state

    def set_state(self, state: ATMState) -> None:
        self.current_state = state

    # public buttons: no logic here, just delegate
    def insert_card(self) -> None:
        self.current_state.insert_card()

    def eject_card(self) -> None:
        self.current_state.eject_card()

    def withdraw_cash(self, amount: int) -> None:
        self.current_state.withdraw_cash(amount)


# usage
atm = ATMMachine()
atm.withdraw_cash(500)   # Insert the card first.
atm.insert_card()        # Card inserted successfully.
atm.insert_card()        # Card already inserted.
atm.withdraw_cash(500)   # Dispensing 500.
atm.eject_card()         # Card ejected.
atm.eject_card()         # No card to eject.

from __future__ import annotations
from abc import ABC, abstractmethod


class TrafficLightState(ABC):
    @abstractmethod
    def change(self, light: TrafficLight) -> None: ...
    @abstractmethod
    def describe(self) -> str: ...


class RedLight(TrafficLightState):
    def change(self, light: TrafficLight) -> None:
        light.set_state(LightFactory.get("GREEN"))
    def describe(self) -> str:
        return "RED: stop"


class GreenLight(TrafficLightState):
    def change(self, light: TrafficLight) -> None:
        light.set_state(LightFactory.get("YELLOW"))
    def describe(self) -> str:
        return "GREEN: go"


class YellowLight(TrafficLightState):
    def change(self, light: TrafficLight) -> None:
        light.set_state(LightFactory.get("RED"))
    def describe(self) -> str:
        return "YELLOW: slow down"


class LightFactory:
    """Creates each state once, then hands out the same object (hash map cache)."""
    _cache: dict[str, TrafficLightState] = {}
    _types = {"RED": RedLight, "GREEN": GreenLight, "YELLOW": YellowLight}

    @classmethod
    def get(cls, color: str) -> TrafficLightState:
        if color not in cls._cache:
            cls._cache[color] = cls._types[color]()
        return cls._cache[color]


class TrafficLight:
    """Context: just forwards to the current state."""
    def __init__(self) -> None:
        self._state = LightFactory.get("RED")

    def set_state(self, state: TrafficLightState) -> None:
        self._state = state

    def change(self) -> None:
        self._state.change(self)

    def describe(self) -> str:
        return self._state.describe()


# usage
light = TrafficLight()
for _ in range(4):
    print(light.describe())
    light.change()
print(LightFactory.get("RED") is LightFactory.get("RED"))
# RED: stop
# GREEN: go
# YELLOW: slow down
# RED: stop
# True   (same object reused, nothing new created)

from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional


@dataclass
class SupportTicket:
    issue: str
    priority: int          # 1 = easy ... 3 = hard


class SupportHandler(ABC):
    """Base class: holds the 'next' link and the common pass-along logic."""
    def __init__(self) -> None:
        self._next_handler: Optional[SupportHandler] = None

    def set_next_handler(self, nxt: SupportHandler) -> None:
        self._next_handler = nxt

    def handle_request(self, ticket: SupportTicket) -> None:
        if self.can_handle(ticket):
            self.process(ticket)
        elif self._next_handler is not None:
            print(f"{type(self).__name__} cannot handle, passing to next handler")
            self._next_handler.handle_request(ticket)
        else:
            print(f"No handler available for: {ticket.issue}")

    @abstractmethod
    def can_handle(self, ticket: SupportTicket) -> bool: ...

    @abstractmethod
    def process(self, ticket: SupportTicket) -> None: ...


class Level1Support(SupportHandler):
    def can_handle(self, t: SupportTicket) -> bool:
        return t.priority <= 1
    def process(self, t: SupportTicket) -> None:
        print(f"Level 1 resolved: {t.issue}")


class Level2Support(SupportHandler):
    def can_handle(self, t: SupportTicket) -> bool:
        return t.priority <= 2
    def process(self, t: SupportTicket) -> None:
        print(f"Level 2 resolved: {t.issue}")


class Level3Support(SupportHandler):
    def can_handle(self, t: SupportTicket) -> bool:
        return t.priority <= 3
    def process(self, t: SupportTicket) -> None:
        print(f"Level 3 resolved: {t.issue}")


# usage: build the chain L1 -> L2 -> L3, always start at L1
level1, level2, level3 = Level1Support(), Level2Support(), Level3Support()
level1.set_next_handler(level2)
level2.set_next_handler(level3)

level1.handle_request(SupportTicket("Login issue", 1))
level1.handle_request(SupportTicket("Server down", 3))
level1.handle_request(SupportTicket("Data centre fire", 5))
# Level 1 resolved: Login issue
# Level1Support cannot handle, passing to next handler
# Level2Support cannot handle, passing to next handler
# Level 3 resolved: Server down
# Level1Support cannot handle, passing to next handler
# Level2Support cannot handle, passing to next handler
# No handler available for: Data centre fire