Let's explore the Adapter Pattern, which helps incompatible interfaces work together. Think of it like a power adapter that lets you plug a US device into a European outlet.
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, Optional
from datetime import datetime
# Modern payment interface that our system expects
class PaymentProcessor(ABC):
@abstractmethod
def process_payment(self, amount: float) -> bool:
pass
@abstractmethod
def refund_payment(self, amount: float) -> bool:
pass
# Data structure for payment details
@dataclass
class PaymentDetails:
amount: float
card_number: str
expiry: str
cvv: str
# Legacy Payment System (Adaptee)
class LegacyPaymentSystem:
def __init__(self):
self.credentials = {"merchant_id": "12345", "api_key": "secret"}
def submit_long_card_number(self, card: str) -> bool:
return len(card) in [15, 16] and card.isdigit()
def verify_security_code(self, code: str) -> bool:
return len(code) in [3, 4] and code.isdigit()
def do_legacy_transaction(self, amount: int, card: str) -> Dict:
return {
"transaction_id": "legacy_123",
"status": "SUCCESS",
"amount": amount,
"timestamp": datetime.now()
}
# Third-party Payment System (Another Adaptee)
class StripePaymentSystem:
def __init__(self):
self.api_key = "stripe_secret_key"
def validate_card_details(self, card_number: str, exp_month: str,
exp_year: str, cvc: str) -> bool:
# Simulate Stripe's validation
return True
def create_charge(self, amount: int, currency: str = "USD") -> Dict:
return {
"id": "stripe_123",
"status": "succeeded",
"amount": amount,
"currency": currency
}
def create_refund(self, charge_id: str, amount: int) -> Dict:
return {
"id": "refund_123",
"status": "succeeded",
"amount": amount
}
# Adapter for Legacy Payment System
class LegacyPaymentAdapter(PaymentProcessor):
def __init__(self):
self.legacy_system = LegacyPaymentSystem()
self.last_transaction_id: Optional[str] = None
def process_payment(self, amount: float) -> bool:
# Convert amount to cents for legacy system
amount_cents = int(amount * 100)
# Simulate card validation and processing
card = "4111111111111111" # Example card
if not self.legacy_system.submit_long_card_number(card):
return False
result = self.legacy_system.do_legacy_transaction(amount_cents, card)
self.last_transaction_id = result["transaction_id"]
return result["status"] == "SUCCESS"
def refund_payment(self, amount: float) -> bool:
if not self.last_transaction_id:
return False
# Legacy system doesn't support refunds directly
print(f"Manual refund required for transaction {self.last_transaction_id}")
return True
# Adapter for Stripe Payment System
class StripePaymentAdapter(PaymentProcessor):
def __init__(self):
self.stripe = StripePaymentSystem()
self.last_charge_id: Optional[str] = None
def process_payment(self, amount: float) -> bool:
# Convert amount to cents for Stripe
amount_cents = int(amount * 100)
result = self.stripe.create_charge(amount_cents)
self.last_charge_id = result["id"]
return result["status"] == "succeeded"
def refund_payment(self, amount: float) -> bool:
if not self.last_charge_id:
return False
amount_cents = int(amount * 100)
result = self.stripe.create_refund(self.last_charge_id, amount_cents)
return result["status"] == "succeeded"
# Client code that uses the payment processor
class PaymentService:
def __init__(self, payment_processor: PaymentProcessor):
self.payment_processor = payment_processor
def make_payment(self, amount: float) -> bool:
print(f"\\\\nProcessing payment of ${amount:.2f}")
success = self.payment_processor.process_payment(amount)
if success:
print("Payment processed successfully")
else:
print("Payment processing failed")
return success
def make_refund(self, amount: float) -> bool:
print(f"\\\\nProcessing refund of ${amount:.2f}")
success = self.payment_processor.refund_payment(amount)
if success:
print("Refund processed successfully")
else:
print("Refund processing failed")
return success
def main():
# Using Legacy Payment System
print("=== Testing Legacy Payment System ===")
legacy_adapter = LegacyPaymentAdapter()
legacy_service = PaymentService(legacy_adapter)
legacy_service.make_payment(100.00)
legacy_service.make_refund(100.00)
# Using Stripe Payment System
print("\\\\n=== Testing Stripe Payment System ===")
stripe_adapter = StripePaymentAdapter()
stripe_service = PaymentService(stripe_adapter)
stripe_service.make_payment(150.00)
stripe_service.make_refund(150.00)
if __name__ == "__main__":
main()
The Adapter Pattern is crucial in real-world applications for several reasons:
Integration with Legacy Systems:
Real-world Python Examples:
# Python's built-in functions use adapters
list(range(5)) # range object adapted to list interface
dict(zip(['a', 'b'], [1, 2])) # zip adapted to dict interface
Key Benefits:
Would you like to:
Yes explain
Let me show you how Python's Database API (DB-API) uses the Adapter Pattern to make different databases work uniformly.
from abc import ABC, abstractmethod
from typing import List, Any, Dict
from dataclasses import dataclass
# Common interface for database operations
class DatabaseAdapter(ABC):
@abstractmethod
def connect(self) -> None:
pass
@abstractmethod
def execute(self, query: str, params: tuple = None) -> List[Any]:
pass
@abstractmethod
def close(self) -> None:
pass
# Simulated different database drivers
class MySQLDriver:
def mysql_connect(self, host: str, user: str, password: str) -> None:
print(f"MySQL: Connected to {host} as {user}")
def mysql_query(self, sql: str, parameters: tuple = None) -> List[Dict]:
print(f"MySQL: Executing: {sql}")
if parameters:
print(f"MySQL: With parameters: {parameters}")
return [{"id": 1, "name": "Sample MySQL Data"}]
def mysql_close(self) -> None:
print("MySQL: Connection closed")
class PostgreSQLDriver:
def pg_connect(self, connection_string: str) -> None:
print(f"PostgreSQL: Connected with {connection_string}")
def pg_execute(self, query: str, params: tuple = None) -> List[Dict]:
print(f"PostgreSQL: Executing: {query}")
if params:
print(f"PostgreSQL: With parameters: {params}")
return [{"id": 1, "name": "Sample PostgreSQL Data"}]
def pg_disconnect(self) -> None:
print("PostgreSQL: Connection closed")
class SQLiteDriver:
def sqlite_open(self, db_file: str) -> None:
print(f"SQLite: Opened database {db_file}")
def sqlite_run(self, sql_query: str, bindings: tuple = None) -> List[Dict]:
print(f"SQLite: Running: {sql_query}")
if bindings:
print(f"SQLite: With bindings: {bindings}")
return [{"id": 1, "name": "Sample SQLite Data"}]
def sqlite_shutdown(self) -> None:
print("SQLite: Database closed")
# Adapters for each database system
class MySQLAdapter(DatabaseAdapter):
def __init__(self, host: str, user: str, password: str):
self.driver = MySQLDriver()
self.host = host
self.user = user
self.password = password
def connect(self) -> None:
self.driver.mysql_connect(self.host, self.user, self.password)
def execute(self, query: str, params: tuple = None) -> List[Any]:
return self.driver.mysql_query(query, params)
def close(self) -> None:
self.driver.mysql_close()
class PostgreSQLAdapter(DatabaseAdapter):
def __init__(self, connection_string: str):
self.driver = PostgreSQLDriver()
self.connection_string = connection_string
def connect(self) -> None:
self.driver.pg_connect(self.connection_string)
def execute(self, query: str, params: tuple = None) -> List[Any]:
return self.driver.pg_execute(query, params)
def close(self) -> None:
self.driver.pg_disconnect()
class SQLiteAdapter(DatabaseAdapter):
def __init__(self, db_file: str):
self.driver = SQLiteDriver()
self.db_file = db_file
def connect(self) -> None:
self.driver.sqlite_open(self.db_file)
def execute(self, query: str, params: tuple = None) -> List[Any]:
return self.driver.sqlite_run(query, params)
def close(self) -> None:
self.driver.sqlite_shutdown()
# Generic database operations class that works with any adapter
class DatabaseOperations:
def __init__(self, db_adapter: DatabaseAdapter):
self.db = db_adapter
def fetch_user_data(self, user_id: int) -> List[Dict]:
self.db.connect()
query = "SELECT * FROM users WHERE id = %s"
result = self.db.execute(query, (user_id,))
self.db.close()
return result
def main():
# Test with MySQL
print("=== Testing MySQL ===")
mysql_adapter = MySQLAdapter(
host="localhost",
user="root",
password="password"
)
mysql_ops = DatabaseOperations(mysql_adapter)
mysql_ops.fetch_user_data(1)
# Test with PostgreSQL
print("\\\\n=== Testing PostgreSQL ===")
postgres_adapter = PostgreSQLAdapter(
connection_string="postgresql://user:pass@localhost/db"
)
postgres_ops = DatabaseOperations(postgres_adapter)
postgres_ops.fetch_user_data(1)
# Test with SQLite
print("\\\\n=== Testing SQLite ===")
sqlite_adapter = SQLiteAdapter(db_file="database.db")
sqlite_ops = DatabaseOperations(sqlite_adapter)
sqlite_ops.fetch_user_data(1)
if __name__ == "__main__":
main()
Let me explain why this pattern is so important in database operations:
Python's DB-API Specification (PEP 249):
# All these different database libraries follow the same interface
import sqlite3
import psycopg2 # PostgreSQL
import mysql.connector
# They all support operations like:
connection = db.connect()
cursor = connection.cursor()
cursor.execute("SELECT * FROM table")