Abstract Classes in Python — Stop Silent Method Failures

# ============================================================
# PART 1: The problem — a regular base class with no enforcement
# ============================================================

class Notifier:
    """
    Intends every subclass to override send().
    But Python has no mechanism to enforce that intent here.
    The comment is the only contract. Comments are not contracts.
    """

    def send(self, message: str, recipient: str) -> bool:
        # This does nothing and returns None (falsy).
        # Python won't raise any error if a subclass skips this.
        pass


class EmailNotifier(Notifier):
    def send(self, message: str, recipient: str) -> bool:
        print(f"[EMAIL] To: {recipient} | {message}")
        return True


class SlackNotifier(Notifier):
    # Developer forgot send(). Python does not care.
    # This class definition is completely valid as far as the interpreter knows.
    pass


# Both instantiate without error
email = EmailNotifier()
slack = SlackNotifier()  # Should fail — but doesn't

result_email = email.send("Deploy complete", "alice@example.com")
result_slack = slack.send("Deploy complete", "#alerts")  # Silently does nothing

print(f"Email delivered: {result_email}")  # True
print(f"Slack delivered: {result_slack}")  # None — falsy, silent failure

# In production: if result_slack: log_delivery() — this never runs.
# The alert is never logged as failed either. It just disappears.

from abc import ABC, abstractmethod
from datetime import datetime
from typing import Optional


class Notifier(ABC):
    """
    Abstract base class for all notification channels.

    Contract:
      - send() must be implemented by every concrete subclass
      - channel_name must be declared as a property by every subclass

    Any subclass that omits either of these raises TypeError at instantiation
    time — before any notification attempt is made, before any business logic
    runs, and long before any alert can be silently dropped.
    """

    def __init__(self, sender_id: str) -> None:
        """
        Abstract classes CAN have constructors.
        This initialises shared state every notifier needs.
        Subclasses call super().__init__(sender_id) to reuse this.
        """
        self._sender_id = sender_id
        self._dispatch_count = 0

    @abstractmethod
    def send(self, message: str, recipient: str) -> bool:
        """
        Send a notification to a recipient.

        Returns True if delivery was confirmed, False if it failed.
        Must NEVER return None — callers rely on the boolean result.

        Abstract methods CAN have a body. Subclasses can call super().send()
        to get the shared audit logging below without reimplementing it.
        The override is still mandatory — this body is opt-in, not default.
        """
        # Shared audit logic — any subclass can opt in via super().send()
        self._dispatch_count += 1
        print(
            f"  [AUDIT] Channel={self.channel_name} "
            f"Sender={self._sender_id} "
            f"At={datetime.utcnow().strftime('%H:%M:%S')} UTC "
            f"Attempt=#{self._dispatch_count}"
        )
        return False  # Subclass overrides the meaningful return value

    @property
    @abstractmethod
    def channel_name(self) -> str:
        """
        Human-readable name for this notification channel.

        CORRECT stacking order:
          @property       ← outermost, line 1 above def
          @abstractmethod ← innermost, line 2 above def
          def channel_name(self) -> str:

        WRONG order (@abstractmethod above @property):
          silently breaks enforcement in Python 3.3+ with no error.
        """
        ...

    def get_stats(self) -> dict:
        """Concrete shared method — all subclasses inherit this for free."""
        return {
            "channel": self.channel_name,
            "sender_id": self._sender_id,
            "total_dispatches": self._dispatch_count,
        }


class EmailNotifier(Notifier):
    """Sends notifications via email."""

    def __init__(self, sender_id: str, smtp_host: str) -> None:
        super().__init__(sender_id)  # initialise shared state from ABC
        self._smtp_host = smtp_host

    @property
    def channel_name(self) -> str:
        return "Email"

    def send(self, message: str, recipient: str) -> bool:
        super().send(message, recipient)  # runs shared audit logging
        print(f"  [Email] SMTP:{self._smtp_host} To:{recipient} | {message[:80]}")
        return True


class SMSNotifier(Notifier):
    """Sends notifications via SMS with a 160-character limit."""

    @property
    def channel_name(self) -> str:
        return "SMS"

    def send(self, message: str, recipient: str) -> bool:
        super().send(message, recipient)
        truncated = message[:160]
        print(f"  [SMS] To:{recipient} | {truncated}")
        return True


class SlackNotifier(Notifier):
    """Broken — forgot to implement send()."""

    @property
    def channel_name(self) -> str:
        return "Slack"
    # send() is missing — ABC will catch this at instantiation time


# ============================================================
# Demo
# ============================================================

print("=== Creating valid notifiers ===")
email = EmailNotifier(sender_id="platform-svc", smtp_host="smtp.company.com")
sms = SMSNotifier(sender_id="platform-svc")

print("\n=== Sending notifications ===")
email.send("Deployment to prod-eu-west-1 succeeded.", "oncall@company.com")
print()
sms.send("Your verification code is 829341.", "+14155550199")

print("\n=== Stats (concrete inherited method) ===")
print(email.get_stats())
print(sms.get_stats())

print("\n=== Attempting to instantiate broken SlackNotifier ===")
try:
    slack = SlackNotifier(sender_id="platform-svc")
except TypeError as e:
    print(f"TypeError caught — ABC enforcement working: {e}")

print("\n=== Attempting to instantiate the abstract base itself ===")
try:
    base = Notifier(sender_id="test")
except TypeError as e:
    print(f"TypeError caught — cannot instantiate abstract class: {e}")

print("\n=== Missing abstract methods on the base ===")
print(f"Notifier.__abstractmethods__ = {Notifier.__abstractmethods__}")

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Optional


@dataclass
class PaymentResult:
    success: bool
    transaction_id: Optional[str]
    amount: float
    currency: str
    error_message: Optional[str] = None


class PaymentProcessor(ABC):
    """
    Abstract base for all payment processors.

    The process_payment() template method enforces the sequence:
      1. validate_card()  — must not be skippable
      2. charge()         — only runs if validation passes
      3. _log_transaction() — shared audit infrastructure

    Subclasses implement the domain-specific steps.
    The sequence is owned by this class and cannot drift.
    """

    def __init__(self, api_key: str, environment: str = "production") -> None:
        """Shared initialisation — every processor needs credentials and env."""
        if not api_key:
            raise ValueError("API key cannot be empty")
        self._api_key = api_key
        self._environment = environment
        self._transaction_count = 0

    # ── Abstract properties — configuration contracts ──

    @property
    @abstractmethod
    def processor_name(self) -> str:
        """Human-readable processor name (e.g., 'Stripe', 'PayPal')."""
        ...

    @property
    @abstractmethod
    def supported_currencies(self) -> list[str]:
        """List of ISO 4217 currency codes this processor accepts."""
        ...

    # ── Abstract methods — critical path ──

    @abstractmethod
    def validate_card(self, card_token: str) -> bool:
        """
        Validate payment method before attempting charge.
        Must return True if valid, False if invalid.
        Must NEVER return None.
        """
        ...

    @abstractmethod
    def charge(
        self, amount: float, currency: str, card_token: str
    ) -> PaymentResult:
        """
        Execute the charge against the payment gateway.
        Must return a PaymentResult — never raise silently.
        """
        ...

    @abstractmethod
    def refund(self, transaction_id: str, amount: float) -> PaymentResult:
        """Issue a full or partial refund."""
        ...

    # ── Template method — fixed pipeline, cannot be overridden ──

    def process_payment(
        self, amount: float, currency: str, card_token: str
    ) -> PaymentResult:
        """
        The fixed payment pipeline. Validation always precedes charge.
        No subclass can reorder or skip steps.
        """
        if currency not in self.supported_currencies:
            return PaymentResult(
                success=False,
                transaction_id=None,
                amount=amount,
                currency=currency,
                error_message=(
                    f"{self.processor_name} does not support {currency}. "
                    f"Supported: {self.supported_currencies}"
                )
            )

        if not self.validate_card(card_token):
            return PaymentResult(
                success=False,
                transaction_id=None,
                amount=amount,
                currency=currency,
                error_message="Card validation failed"
            )

        result = self.charge(amount, currency, card_token)
        self._log_transaction(result)
        return result

    # ── Concrete shared infrastructure ──

    def _log_transaction(self, result: PaymentResult) -> None:
        """Shared audit logging — all processors inherit this."""
        self._transaction_count += 1
        status = "SUCCESS" if result.success else "FAILED"
        print(
            f"[AUDIT] {self.processor_name} | {status} | "
            f"{result.currency} {result.amount:.2f} | "
            f"TxID={result.transaction_id} | "
            f"Env={self._environment} | "
            f"At={datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC"
        )

    def get_session_stats(self) -> dict:
        return {
            "processor": self.processor_name,
            "environment": self._environment,
            "transactions_this_session": self._transaction_count,
            "supported_currencies": self.supported_currencies,
        }


# ── Concrete implementation: Stripe ──

class StripeProcessor(PaymentProcessor):

    @property
    def processor_name(self) -> str:
        return "Stripe"

    @property
    def supported_currencies(self) -> list[str]:
        return ["USD", "EUR", "GBP", "CAD", "AUD"]

    def validate_card(self, card_token: str) -> bool:
        is_valid = card_token.startswith("tok_") and len(card_token) > 6
        print(f"  [Stripe] Card validation: {'PASSED' if is_valid else 'FAILED'}")
        return is_valid

    def charge(
        self, amount: float, currency: str, card_token: str
    ) -> PaymentResult:
        print(f"  [Stripe] Charging {currency} {amount:.2f} via {card_token}")
        return PaymentResult(
            success=True,
            transaction_id="ch_stripe_" + card_token[-6:],
            amount=amount,
            currency=currency
        )

    def refund(self, transaction_id: str, amount: float) -> PaymentResult:
        print(f"  [Stripe] Issuing refund of {amount:.2f} for {transaction_id}")
        return PaymentResult(
            success=True,
            transaction_id="re_" + transaction_id,
            amount=amount,
            currency="USD"
        )


# ── Broken processor — forgot charge() ──

class BrokenProcessor(PaymentProcessor):
    """Simulates the 'forgot an abstract method' bug."""

    @property
    def processor_name(self) -> str:
        return "BrokenPay"

    @property
    def supported_currencies(self) -> list[str]:
        return ["USD"]

    def validate_card(self, card_token: str) -> bool:
        return True

    def refund(self, transaction_id: str, amount: float) -> PaymentResult:
        return PaymentResult(True, transaction_id, amount, "USD")

    # charge() is missing — ABC catches this


# ── Demo ──

print("=== Payment Processing Demo ===")
stripe = StripeProcessor(api_key="sk_live_abc123", environment="production")

result = stripe.process_payment(149.99, "USD", "tok_visa_4242")
print(f"  Result: success={result.success} txid={result.transaction_id}\n")

result = stripe.process_payment(49.99, "JPY", "tok_visa_4242")  # unsupported currency
print(f"  Result: success={result.success} error='{result.error_message}'\n")

print(stripe.get_session_stats())

print("\n=== Broken processor — ABC enforcement ===")
try:
    broken = BrokenProcessor(api_key="test_key")
except TypeError as e:
    print(f"TypeError at instantiation: {e}")
    print("The uncharged order never happened. ABC caught it before any customer was affected.")

from abc import ABC, abstractmethod
from typing import Protocol, runtime_checkable


# ── Option 1: typing.Protocol — pure contract, no inheritance required ──

@runtime_checkable
class NotifierProtocol(Protocol):
    """
    A pure capability contract using Protocol.

    Any object that has send() and channel_name with matching signatures
    satisfies this Protocol — no inheritance from NotifierProtocol required.
    This is structural subtyping: shape matters, not lineage.
    """

    def send(self, message: str, recipient: str) -> bool:
        ...

    @property
    def channel_name(self) -> str:
        ...


# ── Option 2: ABC — contract + shared state + shared implementation ──

class AbstractNotifier(ABC):
    """
    An ABC that combines the Protocol contract with shared infrastructure.

    Use this when you want:
      - Instantiation-time TypeError enforcement (not just mypy warnings)
      - Shared state (__init__ fields) inherited by all subclasses
      - Shared concrete methods (logging, stats) all subclasses get for free
    """

    def __init__(self, sender_id: str) -> None:
        self._sender_id = sender_id
        self._sent_count = 0

    @abstractmethod
    def send(self, message: str, recipient: str) -> bool:
        ...

    @property
    @abstractmethod
    def channel_name(self) -> str:
        ...

    def get_stats(self) -> dict:
        """Shared concrete method — all subclasses inherit this."""
        return {
            "sender_id": self._sender_id,
            "channel": self.channel_name,
            "sent": self._sent_count,
        }


# ── Satisfies Protocol without inheriting from it ──

class ThirdPartyEmailClient:
    """
    From a third-party library — we cannot modify this class.
    It has the right methods, so it satisfies NotifierProtocol
    without any inheritance from our code.
    """

    @property
    def channel_name(self) -> str:
        return "ThirdPartyEmail"

    def send(self, message: str, recipient: str) -> bool:
        print(f"  [ThirdPartyEmail] To:{recipient} | {message}")
        return True


# ── Satisfies ABC by inheriting from it ──

class SlackNotifier(AbstractNotifier):

    @property
    def channel_name(self) -> str:
        return "Slack"

    def send(self, message: str, recipient: str) -> bool:
        self._sent_count += 1
        print(f"  [Slack] To:{recipient} | {message}")
        return True


# ── Demo: both work as notification channels ──

print("=== Protocol isinstance check ===")
client = ThirdPartyEmailClient()
print(f"ThirdPartyEmailClient satisfies NotifierProtocol: {isinstance(client, NotifierProtocol)}")
# True — because it has the right method signatures, regardless of inheritance

print("\n=== ABC enforcement ===")
slack = SlackNotifier(sender_id="platform-svc")
slack.send("Deployment complete", "#alerts")
print(slack.get_stats())  # Inherited concrete method

print("\n=== Which approach to use? ===")
print("Protocol: third-party integration, no shared state, type checking via mypy")
print("ABC:      shared state needed, instantiation-time enforcement, shared infrastructure")


# ── The gotcha: Protocol isinstance without @runtime_checkable ──
# Without @runtime_checkable, isinstance(obj, NotifierProtocol) raises TypeError.
# Add @runtime_checkable when you need runtime isinstance checks.
# Without it, Protocol is a static analysis tool only.

// io.thecodeforge — python tutorial

from abc import ABC, abstractmethod

class PaymentHandler(ABC):
    _fee_percentage = 0.02  # Shared state

    @abstractmethod
    def validate(self, payload: dict) -> bool:
        pass

    @abstractmethod
    def execute(self, payload: dict) -> dict:
        pass

    def apply_fee(self, amount: float) -> float:
        return amount * (1 - self._fee_percentage)

class StripeHandler(PaymentHandler):
    def validate(self, payload):
        return "token" in payload

    def execute(self, payload):
        return {"status": "charged", "amount": self.apply_fee(payload["amount"])}

// io.thecodeforge — python tutorial

from abc import ABC, abstractmethod

class DataConnector(ABC):
    @property
    @abstractmethod
    def connection_string(self) -> str:
        pass

    def connect(self):
        # Concrete method using abstract property
        print(f"Connecting to {self.connection_string}...")

class PostgresConnector(DataConnector):
    @property
    def connection_string(self):
        return "postgresql://localhost:5432/prod"

pg = PostgresConnector()
pg.connect()

// io.thecodeforge — python tutorial

from abc import ABC, abstractmethod

class PaymentGateway(ABC):
    @abstractmethod
    def process_payment(self, amount: float) -> str:
        """Process payment, return transaction ID."""

    @abstractmethod
    def refund(self, transaction_id: str) -> bool:
        """Refund transaction, return success status."""

class StripeGateway(PaymentGateway):
    def process_payment(self, amount: float) -> str:
        return f"txn_stripe_{amount}"

    def refund(self, transaction_id: str) -> bool:
        return True

# This fails at instantiation — missing refund
# class BadGateway(PaymentGateway):
#     pass

g = StripeGateway()
print(g.process_payment(99.99))

// io.thecodeforge — python tutorial

from abc import ABC, abstractmethod

class NotificationChannel(ABC):
    @abstractmethod
    def send(self, message: str) -> bool:
        pass

class EmailChannel(NotificationChannel):
    def send(self, message: str) -> bool:
        print(f"Email: {message}")
        return True

class SMSChannel(NotificationChannel):
    def send(self, message: str) -> bool:
        print(f"SMS: {message}")
        return True

def notify_all(channels: list[NotificationChannel], msg: str):
    for ch in channels:
        ch.send(msg)

channels = [EmailChannel(), SMSChannel()]
notify_all(channels, "Deploy complete")

// io.thecodeforge — python tutorial
from abc import ABC, abstractmethod

class PaymentProcessorBase(ABC):
    """Best practice: single-responsibility ABC."""
    
    @abstractmethod
    def charge(self, amount: float) -> str:
        pass
    
    @abstractmethod
    def refund(self, transaction_id: str) -> bool:
        pass
    
    # Never add concrete methods here unless truly shared.
    # Use a mixin or separate helper class instead.

class StripeProcessor(PaymentProcessorBase):
    def charge(self, amount: float) -> str:
        return f"Stripe charged ${amount}"
    
    def refund(self, transaction_id: str) -> bool:
        return True

// io.thecodeforge — python tutorial
from abc import ABC, abstractmethod

class DataExporter(ABC):
    def export(self, source: str) -> str:
        data = self._extract(source)
        transformed = self._transform(data)
        return self._load(transformed)
    
    @abstractmethod
    def _extract(self, source: str) -> list:
        pass
    
    @abstractmethod
    def _transform(self, data: list) -> list:
        pass
    
    @abstractmethod
    def _load(self, data: list) -> str:
        pass

class CSVExporter(DataExporter):
    def _extract(self, source: str) -> list:
        return ["row1", "row2"]
    
    def _transform(self, data: list) -> list:
        return [d.upper() for d in data]
    
    def _load(self, data: list) -> str:
        return "\n".join(data)

// io.thecodeforge — python tutorial
from abc import ABC, abstractmethod

class PaymentProcessor(ABC):
    @abstractmethod
    def charge(self, amount: float) -> str:
        pass

class StripeProcessor(PaymentProcessor):
    def charge(self, amount: float) -> str:
        return f"Stripe charged ${amount}"

# Uncomment to see enforced contract:
# broken = PaymentProcessor()  # TypeError: Can't instantiate abstract class

// io.thecodeforge — python tutorial
from abc import ABC, abstractmethod

class Logger(ABC):
    @abstractmethod
    def log(self, msg: str) -> None:
        pass

class FileLogger(Logger):
    def log(self, msg: str) -> None:
        print(f"[FILE] {msg}")

fl = FileLogger()
fl.log("Server started")  # Output: [FILE] Server started

class BrokenLogger(Logger):
    pass

try:
    bl = BrokenLogger()
except TypeError as e:
    print(e)  # Output: Can't instantiate abstract class...