The Adapter Pattern is a structural design pattern that allows objects with incompatible interfaces to collaborate. It acts as a bridge between two incompatible interfaces by wrapping an existing class with a new interface that clients expect to use.
Problem Statement
In software development, you often encounter situations where you need to use a class that doesn’t have the interface your code expects. This commonly happens when:
- Integrating third-party libraries with different interfaces
- Working with legacy code that cannot be modified
- Connecting systems with different data formats or protocols
- Reusing existing functionality in new contexts
The traditional solution would be to modify the existing class, but this isn’t always possible or desirable due to:
- Lack of access to source code
- Risk of breaking existing functionality
- Violation of the Open/Closed Principle
Solution
The Adapter Pattern solves this problem by creating an adapter class that:
- Implements the interface expected by the client
- Contains a reference to the adaptee (the class being adapted)
- Translates calls from the client interface to the adaptee’s interface
Client ──────> Target Interface ──────> Adapter ──────> Adaptee
uses implements wraps
Structure
┌─────────────┐ ┌─────────────────┐ ┌─────────────┐
│ Client │───▶│ Target Interface│◀───│ Adapter │
└─────────────┘ └─────────────────┘ └─────────────┘
│
▼
┌─────────────┐
│ Adaptee │
└─────────────┘
Key Components
- Target Interface: The interface that clients expect to use
- Adapter: The class that implements the target interface and wraps the adaptee
- Adaptee: The existing class that needs to be adapted
- Client: The code that uses the target interface
Benefits
- Compatibility: Enables collaboration between incompatible interfaces
- Reusability: Allows the reuse of existing functionality without modification
- Separation of Concerns: Keeps interface conversion logic separate from business logic
- Flexibility: Supports multiple adaptees through different adapter implementations
When to Use
- When you need to use an existing class with an incompatible interface
- When creating a reusable class that should work with unforeseen classes
- When you need to use several existing subclasses but cannot adapt their interface by subclassing
- When integrating third-party libraries or legacy systems
Implementation Considerations
The Adapter Pattern can be implemented in two ways:
- Object Adapter: Uses composition to wrap the adaptee
- Class Adapter: Uses multiple inheritance (where supported)
Object Adapter is generally preferred because:
- It doesn’t require multiple inheritance
- It can adapt a class and all its subclasses
- It’s more flexible and follows the composition over inheritance principle
Common Variations
- Two-way Adapter: Can work as both target and adaptee
- Pluggable Adapter: Uses introspection to adapt to multiple interfaces
- Class Adapter: Uses inheritance instead of composition (less common in Python)
Basic Example
This example demonstrates a media player scenario where we need to integrate a legacy AdvancedMediaPlayer class with a new MediaPlayer interface.
from abc import ABC, abstractmethod
from typing import Protocol
# Target Interface - What the client expects
class MediaPlayer(Protocol):
"""Protocol defining the interface expected by clients."""
def play(self, audio_type: str, filename: str) -> None:
"""Play audio file of specified type."""
pass
# Adaptee - Existing class with incompatible interface
class AdvancedMediaPlayer:
"""Legacy media player with different interface."""
def play_vlc(self, filename: str) -> None:
"""Play VLC format file."""
print(f"Playing VLC file: {filename}")
def play_mp4(self, filename: str) -> None:
"""Play MP4 format file."""
print(f"Playing MP4 file: {filename}")
# Adapter - Bridges the gap between Target and Adaptee
class MediaAdapter:
"""Adapter that makes AdvancedMediaPlayer compatible with MediaPlayer interface."""
def __init__(self) -> None:
self.advanced_player = AdvancedMediaPlayer()
def play(self, audio_type: str, filename: str) -> None:
"""Adapt the play method to work with advanced player."""
if audio_type.lower() == "vlc":
self.advanced_player.play_vlc(filename)
elif audio_type.lower() == "mp4":
self.advanced_player.play_mp4(filename)
else:
print(f"Unsupported format: {audio_type}")
# Client - Uses the Target Interface
class AudioPlayer:
"""Client class that uses MediaPlayer interface."""
def __init__(self) -> None:
self.adapter = MediaAdapter()
def play(self, audio_type: str, filename: str) -> None:
"""Play audio using appropriate player."""
if audio_type.lower() == "mp3":
print(f"Playing MP3 file: {filename}")
else:
# Use adapter for other formats
self.adapter.play(audio_type, filename)
# Usage Example
def main() -> None:
"""Demonstrate the Adapter Pattern."""
player = AudioPlayer()
# Direct support for MP3
player.play("mp3", "song.mp3")
# Adapted support for VLC and MP4
player.play("vlc", "movie.vlc")
player.play("mp4", "video.mp4")
# Unsupported format
player.play("avi", "clip.avi")
if __name__ == "__main__":
main()
Key Points:
- MediaPlayer Protocol: Defines the interface that clients expect
- AdvancedMediaPlayer: The existing class with incompatible methods (play_vlc, play_mp4)
- MediaAdapter: Bridges the gap by implementing MediaPlayer and wrapping AdvancedMediaPlayer
- AudioPlayer: The client that uses the standardized interface
The adapter translates the generic play(audio_type, filename) call into specific method calls on the adaptee based on the audio type.
Payment Gateway Example
This example shows how to use the Adapter Pattern to integrate multiple payment gateways with different APIs into a unified interface.
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from decimal import Decimal
from dataclasses import dataclass
# Common data structures
@dataclass
class PaymentRequest:
"""Standard payment request structure."""
amount: Decimal
currency: str
customer_id: str
description: str
@dataclass
class PaymentResult:
"""Standard payment result structure."""
success: bool
transaction_id: str
message: str
gateway_response: Optional[Dict[str, Any]] = None
# Target Interface - What our application expects
class PaymentGateway(ABC):
"""Abstract payment gateway interface."""
@abstractmethod
def process_payment(self, request: PaymentRequest) -> PaymentResult:
"""Process a payment request."""
pass
@abstractmethod
def get_gateway_name(self) -> str:
"""Get the name of the payment gateway."""
pass
# Adaptee 1 - PayPal SDK (incompatible interface)
class PayPalSDK:
"""Simulated PayPal SDK with different interface."""
def charge_customer(self, customer: str, amount_cents: int, memo: str) -> Dict[str, Any]:
"""PayPal's method signature (amount in cents)."""
if amount_cents < 100: # Minimum $1.00
return {
"status": "failed",
"error": "Minimum amount is $1.00",
"paypal_id": None
}
return {
"status": "completed",
"paypal_id": f"PP_{customer}_{amount_cents}",
"fee_cents": amount_cents * 3 // 100 # 3% fee
}
# Adaptee 2 - Stripe SDK (different incompatible interface)
class StripeAPI:
"""Simulated Stripe API with different interface."""
def create_charge(self, **kwargs) -> Dict[str, str]:
"""Stripe's method signature (amount in dollars)."""
amount = kwargs.get('amount_dollars', 0)
customer = kwargs.get('customer_token', '')
if amount < 0.50: # Minimum $0.50
return {
"id": "",
"status": "failed",
"failure_code": "amount_too_small"
}
return {
"id": f"ch_{customer}_{int(amount * 100)}",
"status": "succeeded",
"balance_transaction": f"txn_{customer}"
}
# Adapter 1 - PayPal Adapter
class PayPalAdapter(PaymentGateway):
"""Adapter for PayPal SDK."""
def __init__(self) -> None:
self.paypal = PayPalSDK()
def process_payment(self, request: PaymentRequest) -> PaymentResult:
"""Adapt PaymentRequest to PayPal SDK format."""
# Convert amount to cents (PayPal expects cents)
amount_cents = int(request.amount * 100)
# Call PayPal SDK with adapted parameters
result = self.paypal.charge_customer(
customer=request.customer_id,
amount_cents=amount_cents,
memo=request.description
)
# Convert PayPal response to standard format
if result["status"] == "completed":
return PaymentResult(
success=True,
transaction_id=result["paypal_id"],
message="Payment processed successfully",
gateway_response=result
)
else:
return PaymentResult(
success=False,
transaction_id="",
message=result.get("error", "Payment failed"),
gateway_response=result
)
def get_gateway_name(self) -> str:
return "PayPal"
# Adapter 2 - Stripe Adapter
class StripeAdapter(PaymentGateway):
"""Adapter for Stripe API."""
def __init__(self) -> None:
self.stripe = StripeAPI()
def process_payment(self, request: PaymentRequest) -> PaymentResult:
"""Adapt PaymentRequest to Stripe API format."""
# Call Stripe API with adapted parameters
result = self.stripe.create_charge(
amount_dollars=float(request.amount),
customer_token=request.customer_id,
description=request.description,
currency=request.currency
)
# Convert Stripe response to standard format
if result["status"] == "succeeded":
return PaymentResult(
success=True,
transaction_id=result["id"],
message="Charge completed successfully",
gateway_response=result
)
else:
return PaymentResult(
success=False,
transaction_id="",
message=f"Payment failed: {result.get('failure_code', 'unknown_error')}",
gateway_response=result
)
def get_gateway_name(self) -> str:
return "Stripe"
# Client - Payment processor that uses the standard interface
class PaymentProcessor:
"""Client that processes payments using standard interface."""
def __init__(self, gateway: PaymentGateway) -> None:
self.gateway = gateway
def process_customer_payment(self, request: PaymentRequest) -> None:
"""Process payment and handle result."""
print(f"\nProcessing payment via {self.gateway.get_gateway_name()}")
print(f"Amount: {request.currency} {request.amount}")
print(f"Customer: {request.customer_id}")
result = self.gateway.process_payment(request)
if result.success:
print(f"✓ Payment successful!")
print(f" Transaction ID: {result.transaction_id}")
print(f" Message: {result.message}")
else:
print(f"✗ Payment failed!")
print(f" Message: {result.message}")
# Usage Example
def main() -> None:
"""Demonstrate payment gateway adapters."""
# Create payment request
payment_request = PaymentRequest(
amount=Decimal("25.99"),
currency="USD",
customer_id="cust_12345",
description="Online purchase"
)
# Test with PayPal adapter
paypal_gateway = PayPalAdapter()
paypal_processor = PaymentProcessor(paypal_gateway)
paypal_processor.process_customer_payment(payment_request)
# Test with Stripe adapter
stripe_gateway = StripeAdapter()
stripe_processor = PaymentProcessor(stripe_gateway)
stripe_processor.process_customer_payment(payment_request)
# Test error case (amount too small)
small_payment = PaymentRequest(
amount=Decimal("0.25"),
currency="USD",
customer_id="cust_67890",
description="Small purchase"
)
print("\n" + "="*50)
print("Testing error handling:")
paypal_processor.process_customer_payment(small_payment)
stripe_processor.process_customer_payment(small_payment)
if __name__ == "__main__":
main()
Key Components:
- PaymentGateway: Abstract base class defining the standard interface
- PaymentRequest/PaymentResult: Data transfer objects for consistent data structures
- PayPalSDK/StripeAPI: Simulated third-party SDKs with different interfaces
- PayPalAdapter/StripeAdapter: Adapters that translate between the standard interface and each SDK
- PaymentProcessor: Client code that works with any payment gateway through the standard interface
Adaptation Details:
- PayPal: Expects amount in cents, returns status with PayPal-specific fields
- Stripe: Expects amount in dollars, returns different status codes and field names
- Adapters: Handle the data transformation and error mapping transparently
Database Connection Example
This example demonstrates how to use the Adapter Pattern to create a unified interface for different database systems with varying connection methods and query interfaces.
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
import json
# Data structures
@dataclass
class DatabaseConfig:
"""Database configuration."""
host: str
port: int
database: str
username: str
password: str
@dataclass
class QueryResult:
"""Standardized query result."""
success: bool
data: List[Dict[str, Any]]
affected_rows: int
error_message: Optional[str] = None
# Target Interface - What our application expects
class DatabaseConnection(ABC):
"""Abstract database connection interface."""
@abstractmethod
def connect(self, config: DatabaseConfig) -> bool:
"""Establish database connection."""
pass
@abstractmethod
def execute_query(self, query: str, params: Optional[Dict[str, Any]] = None) -> QueryResult:
"""Execute a query and return standardized results."""
pass
@abstractmethod
def disconnect(self) -> None:
"""Close database connection."""
pass
@abstractmethod
def get_database_type(self) -> str:
"""Get the type of database."""
pass
# Adaptee 1 - MySQL-like database class
class MySQLDatabase:
"""Simulated MySQL database with specific interface."""
def __init__(self) -> None:
self.connected = False
self.connection_string = ""
def mysql_connect(self, host: str, port: int, db: str, user: str, passwd: str) -> Tuple[bool, str]:
"""MySQL-specific connection method."""
self.connection_string = f"mysql://{user}@{host}:{port}/{db}"
self.connected = True
return True, "Connected successfully"
def mysql_query(self, sql: str, values: Optional[List[Any]] = None) -> Dict[str, Any]:
"""MySQL-specific query method."""
if not self.connected:
return {"error": "Not connected", "rows": [], "count": 0}
# Simulate different query types
if sql.strip().upper().startswith("SELECT"):
# Simulate SELECT results
return {
"rows": [
{"id": 1, "name": "John Doe", "email": "john@example.com"},
{"id": 2, "name": "Jane Smith", "email": "jane@example.com"}
],
"count": 2,
"error": None
}
elif sql.strip().upper().startswith(("INSERT", "UPDATE", "DELETE")):
# Simulate modification results
return {
"rows": [],
"count": 1, # affected rows
"error": None
}
else:
return {"error": "Invalid query", "rows": [], "count": 0}
def mysql_close(self) -> None:
"""MySQL-specific close method."""
self.connected = False
# Adaptee 2 - PostgreSQL-like database class
class PostgreSQLDatabase:
"""Simulated PostgreSQL database with different interface."""
def __init__(self) -> None:
self.is_connected = False
self.connection_info = {}
def pg_connect(self, connection_params: Dict[str, Any]) -> Dict[str, Any]:
"""PostgreSQL-specific connection method."""
self.connection_info = connection_params
self.is_connected = True
return {"status": "OK", "message": "Connection established"}
def pg_execute(self, statement: str, parameters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""PostgreSQL-specific execution method."""
if not self.is_connected:
return {"status": "ERROR", "message": "No connection", "result": None, "rowcount": 0}
# Simulate different query types
if statement.strip().upper().startswith("SELECT"):
return {
"status": "OK",
"result": [
{"user_id": 101, "username": "alice", "active": True},
{"user_id": 102, "username": "bob", "active": False}
],
"rowcount": 2,
"message": "Query executed"
}
elif statement.strip().upper().startswith(("INSERT", "UPDATE", "DELETE")):
return {
"status": "OK",
"result": None,
"rowcount": 3, # affected rows
"message": "Command executed"
}
else:
return {"status": "ERROR", "message": "Syntax error", "result": None, "rowcount": 0}
def pg_disconnect(self) -> None:
"""PostgreSQL-specific disconnect method."""
self.is_connected = False
# Adapter 1 - MySQL Adapter
class MySQLAdapter(DatabaseConnection):
"""Adapter for MySQL database."""
def __init__(self) -> None:
self.mysql_db = MySQLDatabase()
def connect(self, config: DatabaseConfig) -> bool:
"""Adapt connection to MySQL format."""
success, message = self.mysql_db.mysql_connect(
host=config.host,
port=config.port,
db=config.database,
user=config.username,
passwd=config.password
)
return success
def execute_query(self, query: str, params: Optional[Dict[str, Any]] = None) -> QueryResult:
"""Adapt query execution to MySQL format."""
# Convert dict params to list if needed (MySQL expects list)
param_list = list(params.values()) if params else None
result = self.mysql_db.mysql_query(query, param_list)
if result["error"]:
return QueryResult(
success=False,
data=[],
affected_rows=0,
error_message=result["error"]
)
else:
return QueryResult(
success=True,
data=result["rows"],
affected_rows=result["count"]
)
def disconnect(self) -> None:
"""Adapt disconnection to MySQL format."""
self.mysql_db.mysql_close()
def get_database_type(self) -> str:
return "MySQL"
# Adapter 2 - PostgreSQL Adapter
class PostgreSQLAdapter(DatabaseConnection):
"""Adapter for PostgreSQL database."""
def __init__(self) -> None:
self.pg_db = PostgreSQLDatabase()
def connect(self, config: DatabaseConfig) -> bool:
"""Adapt connection to PostgreSQL format."""
connection_params = {
"host": config.host,
"port": config.port,
"database": config.database,
"user": config.username,
"password": config.password
}
result = self.pg_db.pg_connect(connection_params)
return result["status"] == "OK"
def execute_query(self, query: str, params: Optional[Dict[str, Any]] = None) -> QueryResult:
"""Adapt query execution to PostgreSQL format."""
result = self.pg_db.pg_execute(query, params)
if result["status"] == "ERROR":
return QueryResult(
success=False,
data=[],
affected_rows=0,
error_message=result["message"]
)
else:
return QueryResult(
success=True,
data=result["result"] or [],
affected_rows=result["rowcount"]
)
def disconnect(self) -> None:
"""Adapt disconnection to PostgreSQL format."""
self.pg_db.pg_disconnect()
def get_database_type(self) -> str:
return "PostgreSQL"
# Client - Data access layer that uses the standard interface
class DataAccessLayer:
"""Client that performs database operations through standard interface."""
def __init__(self, db_connection: DatabaseConnection) -> None:
self.db = db_connection
def initialize_connection(self, config: DatabaseConfig) -> bool:
"""Initialize database connection."""
print(f"Connecting to {self.db.get_database_type()} database...")
success = self.db.connect(config)
if success:
print("✓ Connection established successfully")
else:
print("✗ Connection failed")
return success
def get_users(self) -> List[Dict[str, Any]]:
"""Retrieve users from database."""
print(f"\nQuerying users from {self.db.get_database_type()}...")
result = self.db.execute_query("SELECT * FROM users")
if result.success:
print(f"✓ Retrieved {len(result.data)} users")
return result.data
else:
print(f"✗ Query failed: {result.error_message}")
return []
def update_user_status(self, user_id: int, active: bool) -> bool:
"""Update user status."""
print(f"\nUpdating user {user_id} status to {'active' if active else 'inactive'}...")
query = "UPDATE users SET active = %(active)s WHERE id = %(user_id)s"
params = {"active": active, "user_id": user_id}
result = self.db.execute_query(query, params)
if result.success:
print(f"✓ Updated {result.affected_rows} user(s)")
return True
else:
print(f"✗ Update failed: {result.error_message}")
return False
def cleanup(self) -> None:
"""Close database connection."""
print(f"\nClosing {self.db.get_database_type()} connection...")
self.db.disconnect()
print("✓ Connection closed")
# Usage Example
def main() -> None:
"""Demonstrate database connection adapters."""
# Database configuration
config = DatabaseConfig(
host="localhost",
port=5432,
database="myapp",
username="user",
password="password"
)
print("="*60)
print("DATABASE ADAPTER PATTERN DEMONSTRATION")
print("="*60)
# Test with MySQL adapter
print("\n" + "-"*30)
print("Testing MySQL Database")
print("-"*30)
mysql_adapter = MySQLAdapter()
mysql_dal = DataAccessLayer(mysql_adapter)
if mysql_dal.initialize_connection(config):
users = mysql_dal.get_users()
if users:
print("Users found:")
for user in users:
print(f" - {user}")
mysql_dal.update_user_status(1, False)
mysql_dal.cleanup()
# Test with PostgreSQL adapter
print("\n" + "-"*30)
print("Testing PostgreSQL Database")
print("-"*30)
pg_adapter = PostgreSQLAdapter()
pg_dal = DataAccessLayer(pg_adapter)
if pg_dal.initialize_connection(config):
users = pg_dal.get_users()
if users:
print("Users found:")
for user in users:
print(f" - {user}")
pg_dal.update_user_status(101, True)
pg_dal.cleanup()
if __name__ == "__main__":
main()
Key Aspects:
- DatabaseConnection: Abstract interface providing standard database operations
- MySQLDatabase/PostgreSQLDatabase: Simulated database classes with different APIs and conventions
- MySQLAdapter/PostgreSQLAdapter: Adapters that translate the standard interface to database-specific calls
- DataAccessLayer: Client code that can work with any database through the unified interface
Adaptation Challenges Addressed:
- Connection Methods: MySQL uses separate parameters vs PostgreSQL uses a config dictionary
- Query Parameters: MySQL expects positional parameters vs PostgreSQL uses named parameters
- Result Formats: Different field names and structures in query results
- Error Handling: Different error reporting mechanisms across databases