Dependency Inversion Principle

HighLevel → creates → LowLevel

HighLevel → depends on → Abstraction ← implements ← LowLevel

class MySQLDatabase:
    def connect(self):
        print("Connecting to MySQL...")
    
    def query(self, sql):
        print(f"MySQL executing: {sql}")
        return [{"id": 1, "name": "Alice"}]

class UserRepository:
    def __init__(self):
        # 🚨 Directly creates MySQL - tightly coupled!
        self.database = MySQLDatabase()
    
    def find_user(self, user_id):
        self.database.connect()
        return self.database.query(f"SELECT * FROM users WHERE id = {user_id}")

# Problems:
# 1. Can't switch to PostgreSQL without changing UserRepository
# 2. Can't test without a real MySQL database
# 3. Can't use a different database for different environments

# Want to switch to PostgreSQL? Must modify UserRepository!
# Want to test? Must have MySQL running!

from abc import ABC, abstractmethod

# 1. Define the abstraction (interface)
class Database(ABC):
    @abstractmethod
    def connect(self):
        pass
    
    @abstractmethod
    def query(self, sql):
        pass

# 2. Implement concrete versions
class MySQLDatabase(Database):
    def connect(self):
        print("🐬 Connecting to MySQL...")
    
    def query(self, sql):
        print(f"🐬 MySQL: {sql}")
        return [{"id": 1, "name": "Alice"}]

class PostgreSQLDatabase(Database):
    def connect(self):
        print("🐘 Connecting to PostgreSQL...")
    
    def query(self, sql):
        print(f"🐘 PostgreSQL: {sql}")
        return [{"id": 1, "name": "Alice"}]

class MockDatabase(Database):
    """For testing!"""
    def connect(self):
        print("🧪 Mock database connected")
    
    def query(self, sql):
        print(f"🧪 Mock query: {sql}")
        return [{"id": 99, "name": "Test User"}]

# 3. Depend on abstraction, receive via injection
class UserRepository:
    def __init__(self, database: Database):  # 🎯 Accepts ANY Database!
        self.database = database
    
    def find_user(self, user_id):
        self.database.connect()
        return self.database.query(f"SELECT * FROM users WHERE id = {user_id}")

# 4. Usage - inject whatever you need!
print("=== Production with MySQL ===")
mysql_repo = UserRepository(MySQLDatabase())
mysql_repo.find_user(1)

print("\n=== Production with PostgreSQL ===")
postgres_repo = UserRepository(PostgreSQLDatabase())
postgres_repo.find_user(1)

print("\n=== Testing with Mock ===")
test_repo = UserRepository(MockDatabase())
test_repo.find_user(1)

# 🎉 Same UserRepository code, different databases!

import smtplib

class NotificationService:
    def __init__(self):
        # 🚨 Directly creating SMTP - tight coupling!
        self.server = smtplib.SMTP('smtp.gmail.com', 587)
        self.server.login('[email protected]', 'password')
    
    def send(self, to, message):
        self.server.send_message(message, to_addrs=[to])

# Problems:
# - Can't switch to SendGrid, Mailgun, or AWS SES easily
# - Can't test without sending real emails!
# - Gmail credentials hardcoded

from abc import ABC, abstractmethod

class EmailProvider(ABC):
    @abstractmethod
    def send(self, to: str, subject: str, body: str):
        pass

class GmailProvider(EmailProvider):
    def __init__(self, username, password):
        self.username = username
        self.password = password
    
    def send(self, to, subject, body):
        print(f"📧 Gmail: Sending to {to}")
        print(f"   Subject: {subject}")
        # Actual SMTP logic here

class SendGridProvider(EmailProvider):
    def __init__(self, api_key):
        self.api_key = api_key
    
    def send(self, to, subject, body):
        print(f"📧 SendGrid: Sending to {to}")
        print(f"   Subject: {subject}")
        # SendGrid API logic here

class AWSEmailProvider(EmailProvider):
    def __init__(self, access_key, secret_key, region):
        self.region = region
    
    def send(self, to, subject, body):
        print(f"📧 AWS SES ({self.region}): Sending to {to}")
        print(f"   Subject: {subject}")
        # AWS SES logic here

class MockEmailProvider(EmailProvider):
    """For testing - doesn't send real emails!"""
    def __init__(self):
        self.sent_emails = []
    
    def send(self, to, subject, body):
        print(f"🧪 Mock: Would send to {to}")
        self.sent_emails.append({"to": to, "subject": subject, "body": body})

class NotificationService:
    def __init__(self, email_provider: EmailProvider):  # 🎯 Inject provider
        self.email = email_provider
    
    def notify_user(self, user_email, message):
        self.email.send(user_email, "Notification", message)

# Usage - swap providers easily!
print("=== Using Gmail ===")
gmail_notifier = NotificationService(GmailProvider("user", "pass"))
gmail_notifier.notify_user("[email protected]", "Your order shipped!")

print("\n=== Using SendGrid ===")
sendgrid_notifier = NotificationService(SendGridProvider("api-key-123"))
sendgrid_notifier.notify_user("[email protected]", "Your order shipped!")

print("\n=== Testing ===")
mock = MockEmailProvider()
test_notifier = NotificationService(mock)
test_notifier.notify_user("[email protected]", "Test message")
print(f"Captured {len(mock.sent_emails)} emails!")  # Can verify emails!

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class PaymentResult:
    success: bool
    transaction_id: str
    message: str

class PaymentGateway(ABC):
    @abstractmethod
    def process(self, amount: float, card_token: str) -> PaymentResult:
        pass
    
    @abstractmethod
    def refund(self, transaction_id: str) -> PaymentResult:
        pass

class StripeGateway(PaymentGateway):
    def __init__(self, api_key):
        self.api_key = api_key
    
    def process(self, amount, card_token):
        print(f"💳 Stripe: Processing ${amount}")
        return PaymentResult(True, "stripe_tx_123", "Payment successful")
    
    def refund(self, transaction_id):
        print(f"💳 Stripe: Refunding {transaction_id}")
        return PaymentResult(True, transaction_id, "Refund successful")

class PayPalGateway(PaymentGateway):
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
    
    def process(self, amount, card_token):
        print(f"🅿️ PayPal: Processing ${amount}")
        return PaymentResult(True, "paypal_tx_456", "Payment successful")
    
    def refund(self, transaction_id):
        print(f"🅿️ PayPal: Refunding {transaction_id}")
        return PaymentResult(True, transaction_id, "Refund successful")

class SquareGateway(PaymentGateway):
    def __init__(self, access_token):
        self.access_token = access_token
    
    def process(self, amount, card_token):
        print(f"⬜ Square: Processing ${amount}")
        return PaymentResult(True, "square_tx_789", "Payment successful")
    
    def refund(self, transaction_id):
        print(f"⬜ Square: Refunding {transaction_id}")
        return PaymentResult(True, transaction_id, "Refund successful")

class MockPaymentGateway(PaymentGateway):
    """For testing!"""
    def __init__(self, should_succeed=True):
        self.should_succeed = should_succeed
        self.processed = []
    
    def process(self, amount, card_token):
        self.processed.append(amount)
        if self.should_succeed:
            return PaymentResult(True, "mock_tx_000", "Mock success")
        return PaymentResult(False, "", "Mock failure")
    
    def refund(self, transaction_id):
        return PaymentResult(True, transaction_id, "Mock refund")

# Order service depends on abstraction
class OrderService:
    def __init__(self, payment_gateway: PaymentGateway):  # 🎯 Injected!
        self.payment = payment_gateway
    
    def checkout(self, order_total, card_token):
        print(f"🛒 Processing order for ${order_total}")
        result = self.payment.process(order_total, card_token)
        
        if result.success:
            print(f"✅ Order confirmed! Transaction: {result.transaction_id}")
        else:
            print(f"❌ Order failed: {result.message}")
        
        return result

# 🎉 Easy to switch payment providers!

# Production - use Stripe
order_service = OrderService(StripeGateway("sk_live_xxx"))
order_service.checkout(99.99, "tok_visa")

# Different environment - use PayPal
print()
paypal_service = OrderService(PayPalGateway("client_id", "secret"))
paypal_service.checkout(49.99, "paypal_tok")

# Testing - use mock
print("\n=== Testing ===")
mock = MockPaymentGateway(should_succeed=True)
test_service = OrderService(mock)
test_service.checkout(100.00, "test_token")
print(f"Total processed in tests: ${sum(mock.processed)}")

class UserService:
    def __init__(self, user_repo: UserRepository, email_service: EmailService):
        self.users = user_repo
        self.email = email_service

class UserService:
    def __init__(self):
        self._user_repo = None
    
    def set_repository(self, repo: UserRepository):
        self._user_repo = repo

class UserService:
    def create_user(self, data, email_service: EmailService):
        # Email service passed only when needed
        email_service.send_welcome(data['email'])

class Container:
    """Simple dependency injection container"""
    
    def __init__(self):
        self._services = {}
    
    def register(self, interface, implementation):
        self._services[interface] = implementation
    
    def resolve(self, interface):
        if interface not in self._services:
            raise KeyError(f"Service {interface} not registered")
        return self._services[interface]

# Setup container
container = Container()

# Register implementations
if ENVIRONMENT == "production":
    container.register("database", MySQLDatabase())
    container.register("email", SendGridProvider("api_key"))
    container.register("payment", StripeGateway("stripe_key"))
else:
    container.register("database", MockDatabase())
    container.register("email", MockEmailProvider())
    container.register("payment", MockPaymentGateway())

# Resolve dependencies
db = container.resolve("database")
email = container.resolve("email")

# Create services with resolved dependencies
user_repo = UserRepository(db)
notification = NotificationService(email)

WITHOUT DIP (Tight Coupling):
┌─────────────────────────────────┐
│         OrderService            │
│                                 │
│  ┌─────────────────────────┐    │
│  │   StripeGateway         │◄───┼── Hardcoded inside!
│  │   (concrete class)      │    │
│  └─────────────────────────┘    │
│                                 │
│  - Can't change gateway         │
│  - Can't test without Stripe    │
│  - Coupled to implementation    │
└─────────────────────────────────┘

WITH DIP (Loose Coupling):
┌─────────────────────────────────┐
│         OrderService            │
│                                 │
│  ┌─────────────────────────┐    │
│  │   PaymentGateway        │◄───┼── Interface (abstraction)
│  │   (abstract interface)  │    │
│  └─────────────────────────┘    │
└─────────────────────────────────┘
           ▲
           │ implements
    ┌──────┼──────┬─────────────┐
    │      │      │             │
┌───┴───┐ ┌┴────┐ ┌┴─────────┐ ┌┴────────┐
│Stripe │ │PayPal│ │  Square │ │MockPay │
└───────┘ └─────┘ └─────────┘ └────────┘

- Easy to swap implementations
- Easy to test with mocks
- Decoupled from details