- Introduction to Abstraction
- Understanding Abstraction Concepts
- Abstract Base Classes (ABC)
- Interface Design Patterns
- Metaclasses and Advanced Abstraction
- Context Managers and Protocols
- Real-World Applications
- Best Practices
- Common Pitfalls
- Exercises and Practice
Abstraction is one of the four fundamental principles of Object-Oriented Programming (OOP). It's the process of hiding complex implementation details while exposing only the essential features and functionality to the user. Abstraction allows you to focus on what an object does rather than how it does it.
🎯 Key Benefits:
- Simplification: Hide complex implementation details
- Modularity: Create clean, well-defined interfaces
- Maintainability: Changes to implementation don't affect users
- Reusability: Abstract interfaces can be implemented in multiple ways
- Consistency: Enforce common behavior across related classes
Think of abstraction like driving a car:
- You know how to use the steering wheel, pedals, and gear shift (interface)
- You don't need to understand the engine, transmission, or fuel injection (implementation)
- Different cars have the same interface but different implementations
- The interface remains consistent even as technology changes
# Level 1: Concrete Implementation
class BasicCalculator:
def add(self, a, b):
return a + b
def subtract(self, a, b):
return a - b
# Level 2: Interface Definition
from abc import ABC, abstractmethod
class Calculator(ABC):
@abstractmethod
def calculate(self, operation, a, b):
pass
# Level 3: Multiple Implementations
class SimpleCalculator(Calculator):
def calculate(self, operation, a, b):
if operation == "add":
return a + b
elif operation == "subtract":
return a - b
else:
raise ValueError("Unsupported operation")
class AdvancedCalculator(Calculator):
def calculate(self, operation, a, b):
operations = {
"add": lambda x, y: x + y,
"subtract": lambda x, y: x - y,
"multiply": lambda x, y: x * y,
"divide": lambda x, y: x / y if y != 0 else float('inf')
}
return operations.get(operation, lambda x, y: 0)(a, b)| Aspect | Abstraction | Encapsulation | Inheritance | Polymorphism |
|---|---|---|---|---|
| Purpose | Hide complexity | Hide implementation | Code reuse | Multiple forms |
| Focus | What it does | How it's protected | IS-A relationship | Same interface |
| Level | Interface design | Data protection | Class hierarchy | Method behavior |
| Example | Abstract methods | Private attributes | Parent/Child | Method overriding |
# 1. Data Abstraction
class Point:
def __init__(self, x, y):
self._x = x
self._y = y
@property
def coordinates(self):
return (self._x, self._y)
def distance_from_origin(self):
return (self._x ** 2 + self._y ** 2) ** 0.5
# 2. Procedural Abstraction
class FileProcessor:
def process_file(self, filename):
data = self._read_file(filename)
processed = self._process_data(data)
self._save_results(processed)
def _read_file(self, filename):
# Hidden implementation
pass
def _process_data(self, data):
# Hidden implementation
pass
def _save_results(self, data):
# Hidden implementation
pass
# 3. Interface Abstraction
from abc import ABC, abstractmethod
class Drawable(ABC):
@abstractmethod
def draw(self):
pass
@abstractmethod
def get_area(self):
passPython's abc module provides infrastructure for defining Abstract Base Classes, which are classes that cannot be instantiated directly and must be subclassed.
from abc import ABC, abstractmethod
class Shape(ABC):
"""Abstract base class for all shapes"""
@abstractmethod
def area(self):
"""Calculate and return the area of the shape"""
pass
@abstractmethod
def perimeter(self):
"""Calculate and return the perimeter of the shape"""
pass
# Concrete method (shared implementation)
def describe(self):
return f"This is a {self.__class__.__name__} with area {self.area():.2f}"
class Circle(Shape):
def __init__(self, radius):
self.radius = radius
def area(self):
import math
return math.pi * self.radius ** 2
def perimeter(self):
import math
return 2 * math.pi * self.radius
class Rectangle(Shape):
def __init__(self, width, height):
self.width = width
self.height = height
def area(self):
return self.width * self.height
def perimeter(self):
return 2 * (self.width + self.height)
# Usage
circle = Circle(5)
rectangle = Rectangle(4, 6)
print(circle.describe()) # This is a Circle with area 78.54
print(rectangle.describe()) # This is a Rectangle with area 24.00
# shape = Shape() # ❌ TypeError: Can't instantiate abstract classfrom abc import ABC, abstractmethod
class Vehicle(ABC):
def __init__(self, make, model):
self._make = make
self._model = model
@property
@abstractmethod
def max_speed(self):
"""Maximum speed of the vehicle"""
pass
@property
@abstractmethod
def fuel_type(self):
"""Type of fuel used by the vehicle"""
pass
@abstractmethod
def start_engine(self):
"""Start the vehicle's engine"""
pass
def get_info(self):
return f"{self._make} {self._model} - Max Speed: {self.max_speed} km/h"
class Car(Vehicle):
@property
def max_speed(self):
return 200
@property
def fuel_type(self):
return "Gasoline"
def start_engine(self):
return "Car engine started with key"
class ElectricCar(Vehicle):
@property
def max_speed(self):
return 250
@property
def fuel_type(self):
return "Electric"
def start_engine(self):
return "Electric motor activated silently"
# Usage
car = Car("Toyota", "Camry")
electric = ElectricCar("Tesla", "Model 3")
print(car.get_info()) # Toyota Camry - Max Speed: 200 km/h
print(electric.start_engine()) # Electric motor activated silentlyfrom abc import ABC, abstractmethod
class DatabaseConnection(ABC):
"""Abstract base class for database connections"""
@abstractmethod
def connect(self):
"""Establish connection to database"""
pass
@abstractmethod
def disconnect(self):
"""Close database connection"""
pass
@abstractmethod
def execute_query(self, query):
"""Execute a query and return results"""
pass
@abstractmethod
def execute_transaction(self, queries):
"""Execute multiple queries in a transaction"""
pass
# Template method (concrete implementation using abstract methods)
def query_with_connection(self, query):
try:
self.connect()
result = self.execute_query(query)
return result
finally:
self.disconnect()
class MySQLConnection(DatabaseConnection):
def __init__(self, host, database):
self.host = host
self.database = database
self.connected = False
def connect(self):
print(f"Connecting to MySQL at {self.host}/{self.database}")
self.connected = True
def disconnect(self):
print("Disconnecting from MySQL")
self.connected = False
def execute_query(self, query):
if not self.connected:
raise RuntimeError("Not connected to database")
print(f"Executing MySQL query: {query}")
return f"MySQL result for: {query}"
def execute_transaction(self, queries):
if not self.connected:
raise RuntimeError("Not connected to database")
print("Starting MySQL transaction")
results = []
for query in queries:
results.append(self.execute_query(query))
print("Committing MySQL transaction")
return results
class PostgreSQLConnection(DatabaseConnection):
def __init__(self, host, database):
self.host = host
self.database = database
self.connected = False
def connect(self):
print(f"Connecting to PostgreSQL at {self.host}/{self.database}")
self.connected = True
def disconnect(self):
print("Disconnecting from PostgreSQL")
self.connected = False
def execute_query(self, query):
if not self.connected:
raise RuntimeError("Not connected to database")
print(f"Executing PostgreSQL query: {query}")
return f"PostgreSQL result for: {query}"
def execute_transaction(self, queries):
if not self.connected:
raise RuntimeError("Not connected to database")
print("Starting PostgreSQL transaction")
results = []
for query in queries:
results.append(self.execute_query(query))
print("Committing PostgreSQL transaction")
return results
# Usage
mysql_db = MySQLConnection("localhost", "myapp")
postgres_db = PostgreSQLConnection("localhost", "myapp")
# Using template method
result1 = mysql_db.query_with_connection("SELECT * FROM users")
result2 = postgres_db.query_with_connection("SELECT * FROM products")
print(f"MySQL result: {result1}")
print(f"PostgreSQL result: {result2}")Python 3.8+ introduced typing.Protocol for structural subtyping (duck typing with type hints):
from typing import Protocol
class Drawable(Protocol):
def draw(self) -> str:
...
def get_area(self) -> float:
...
class Movable(Protocol):
def move(self, x: float, y: float) -> None:
...
# Classes that implement the protocols (no explicit inheritance needed)
class Circle:
def __init__(self, radius: float):
self.radius = radius
self.x = 0.0
self.y = 0.0
def draw(self) -> str:
return f"Drawing circle at ({self.x}, {self.y}) with radius {self.radius}"
def get_area(self) -> float:
import math
return math.pi * self.radius ** 2
def move(self, x: float, y: float) -> None:
self.x = x
self.y = y
class Rectangle:
def __init__(self, width: float, height: float):
self.width = width
self.height = height
self.x = 0.0
self.y = 0.0
def draw(self) -> str:
return f"Drawing rectangle at ({self.x}, {self.y}) with size {self.width}x{self.height}"
def get_area(self) -> float:
return self.width * self.height
def move(self, x: float, y: float) -> None:
self.x = x
self.y = y
# Functions that work with protocols
def render_shape(shape: Drawable) -> None:
print(shape.draw())
print(f"Area: {shape.get_area():.2f}")
def move_object(obj: Movable, x: float, y: float) -> None:
obj.move(x, y)
print(f"Moved object to ({x}, {y})")
# Usage
circle = Circle(5)
rectangle = Rectangle(4, 6)
render_shape(circle) # Works because Circle implements Drawable
render_shape(rectangle) # Works because Rectangle implements Drawable
move_object(circle, 10, 20) # Works because Circle implements Movable
move_object(rectangle, 5, 15) # Works because Rectangle implements MovableMixins provide a way to add functionality to classes through multiple inheritance:
class TimestampMixin:
"""Mixin to add timestamp functionality"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
import datetime
self.created_at = datetime.datetime.now()
self.updated_at = self.created_at
def touch(self):
"""Update the timestamp"""
import datetime
self.updated_at = datetime.datetime.now()
def age_in_seconds(self):
"""Get age in seconds"""
import datetime
return (datetime.datetime.now() - self.created_at).total_seconds()
class SerializableMixin:
"""Mixin to add serialization functionality"""
def to_dict(self):
"""Convert object to dictionary"""
result = {}
for key, value in self.__dict__.items():
if not key.startswith('_'):
if hasattr(value, 'to_dict'):
result[key] = value.to_dict()
else:
result[key] = str(value)
return result
def from_dict(self, data):
"""Load object from dictionary"""
for key, value in data.items():
if hasattr(self, key):
setattr(self, key, value)
class ValidatedMixin:
"""Mixin to add validation functionality"""
def validate(self):
"""Validate object state"""
errors = []
for attr_name in dir(self):
if attr_name.startswith('validate_'):
validator = getattr(self, attr_name)
if callable(validator):
try:
validator()
except ValueError as e:
errors.append(str(e))
if errors:
raise ValueError(f"Validation errors: {', '.join(errors)}")
return True
# Using mixins
class User(TimestampMixin, SerializableMixin, ValidatedMixin):
def __init__(self, username, email, age):
self.username = username
self.email = email
self.age = age
super().__init__() # Initialize mixins
def validate_username(self):
if not self.username or len(self.username) < 3:
raise ValueError("Username must be at least 3 characters")
def validate_email(self):
if '@' not in self.email:
raise ValueError("Invalid email format")
def validate_age(self):
if self.age < 0 or self.age > 150:
raise ValueError("Age must be between 0 and 150")
# Usage
user = User("alice", "alice@example.com", 25)
print(f"User created at: {user.created_at}")
print(f"User age: {user.age_in_seconds():.2f} seconds")
user.validate() # Validates all fields
print("User data:", user.to_dict())
user.touch() # Update timestamp
print(f"User updated at: {user.updated_at}")from abc import ABC, abstractmethod
class SortingStrategy(ABC):
"""Abstract strategy for sorting algorithms"""
@abstractmethod
def sort(self, data):
pass
@property
@abstractmethod
def name(self):
pass
class BubbleSort(SortingStrategy):
@property
def name(self):
return "Bubble Sort"
def sort(self, data):
arr = data.copy()
n = len(arr)
for i in range(n):
for j in range(0, n - i - 1):
if arr[j] > arr[j + 1]:
arr[j], arr[j + 1] = arr[j + 1], arr[j]
return arr
class QuickSort(SortingStrategy):
@property
def name(self):
return "Quick Sort"
def sort(self, data):
if len(data) <= 1:
return data
pivot = data[len(data) // 2]
left = [x for x in data if x < pivot]
middle = [x for x in data if x == pivot]
right = [x for x in data if x > pivot]
return self.sort(left) + middle + self.sort(right)
class MergeSort(SortingStrategy):
@property
def name(self):
return "Merge Sort"
def sort(self, data):
if len(data) <= 1:
return data
mid = len(data) // 2
left = self.sort(data[:mid])
right = self.sort(data[mid:])
return self._merge(left, right)
def _merge(self, left, right):
result = []
i = j = 0
while i < len(left) and j < len(right):
if left[i] <= right[j]:
result.append(left[i])
i += 1
else:
result.append(right[j])
j += 1
result.extend(left[i:])
result.extend(right[j:])
return result
class Sorter:
"""Context class that uses sorting strategies"""
def __init__(self, strategy: SortingStrategy):
self._strategy = strategy
def set_strategy(self, strategy: SortingStrategy):
self._strategy = strategy
def sort_data(self, data):
print(f"Using {self._strategy.name}")
return self._strategy.sort(data)
# Usage
data = [64, 34, 25, 12, 22, 11, 90]
print(f"Original data: {data}")
sorter = Sorter(BubbleSort())
result1 = sorter.sort_data(data)
print(f"Bubble sorted: {result1}")
sorter.set_strategy(QuickSort())
result2 = sorter.sort_data(data)
print(f"Quick sorted: {result2}")
sorter.set_strategy(MergeSort())
result3 = sorter.sort_data(data)
print(f"Merge sorted: {result3}")Metaclasses provide the highest level of abstraction in Python, allowing you to control class creation itself.
class LoggingMeta(type):
"""Metaclass that adds logging to all method calls"""
def __new__(mcs, name, bases, attrs):
# Wrap all callable attributes (methods) with logging
for attr_name, attr_value in attrs.items():
if callable(attr_value) and not attr_name.startswith('_'):
attrs[attr_name] = mcs._add_logging(attr_value, attr_name)
return super().__new__(mcs, name, bases, attrs)
@staticmethod
def _add_logging(func, func_name):
def wrapper(*args, **kwargs):
print(f"Calling {func_name} with args={args[1:]} kwargs={kwargs}")
result = func(*args, **kwargs)
print(f"{func_name} returned: {result}")
return result
return wrapper
class Calculator(metaclass=LoggingMeta):
def add(self, a, b):
return a + b
def multiply(self, a, b):
return a * b
def divide(self, a, b):
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
# Usage
calc = Calculator()
result1 = calc.add(5, 3) # Automatically logged
result2 = calc.multiply(4, 7) # Automatically loggedclass SingletonMeta(type):
"""Metaclass that creates singleton instances"""
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class DatabaseConnection(metaclass=SingletonMeta):
def __init__(self):
if not hasattr(self, 'initialized'):
self.connection_string = "database://localhost:5432/mydb"
self.connected = False
self.initialized = True
def connect(self):
if not self.connected:
print(f"Connecting to {self.connection_string}")
self.connected = True
else:
print("Already connected")
def disconnect(self):
if self.connected:
print("Disconnecting from database")
self.connected = False
else:
print("Already disconnected")
# Usage
db1 = DatabaseConnection()
db2 = DatabaseConnection()
print(f"db1 is db2: {db1 is db2}") # True - same instance
db1.connect()
db2.disconnect() # Affects the same instanceclass RegistryMeta(type):
"""Metaclass that maintains a registry of all classes"""
registry = {}
def __new__(mcs, name, bases, attrs):
cls = super().__new__(mcs, name, bases, attrs)
mcs.registry[name] = cls
return cls
@classmethod
def get_class(mcs, name):
return mcs.registry.get(name)
@classmethod
def list_classes(mcs):
return list(mcs.registry.keys())
class Plugin(metaclass=RegistryMeta):
"""Base class for all plugins"""
@abstractmethod
def execute(self):
pass
class EmailPlugin(Plugin):
def execute(self):
return "Sending email..."
class SMSPlugin(Plugin):
def execute(self):
return "Sending SMS..."
class PushNotificationPlugin(Plugin):
def execute(self):
return "Sending push notification..."
# Usage
print("Available plugins:", RegistryMeta.list_classes())
# Dynamically create plugin instances
plugin_name = "EmailPlugin"
plugin_class = RegistryMeta.get_class(plugin_name)
if plugin_class:
plugin = plugin_class()
print(plugin.execute())Context managers provide a clean way to manage resources and implement the context management protocol.
class FileManager:
"""Context manager for file operations"""
def __init__(self, filename, mode='r'):
self.filename = filename
self.mode = mode
self.file = None
def __enter__(self):
print(f"Opening file: {self.filename}")
self.file = open(self.filename, self.mode)
return self.file
def __exit__(self, exc_type, exc_value, traceback):
print(f"Closing file: {self.filename}")
if self.file:
self.file.close()
if exc_type is not None:
print(f"Exception occurred: {exc_type.__name__}: {exc_value}")
# Return False to propagate exceptions
return False
# Usage
try:
with FileManager("test.txt", "w") as f:
f.write("Hello, World!")
# File is automatically closed when exiting the with block
except FileNotFoundError as e:
print(f"File error: {e}")class DatabaseTransaction:
"""Context manager for database transactions"""
def __init__(self, connection):
self.connection = connection
self.transaction_started = False
def __enter__(self):
print("Starting database transaction")
self.connection.begin_transaction()
self.transaction_started = True
return self
def __exit__(self, exc_type, exc_value, traceback):
if self.transaction_started:
if exc_type is None:
print("Committing transaction")
self.connection.commit()
else:
print(f"Rolling back transaction due to {exc_type.__name__}")
self.connection.rollback()
return False # Don't suppress exceptions
class MockConnection:
"""Mock database connection for demonstration"""
def begin_transaction(self):
print("BEGIN TRANSACTION")
def commit(self):
print("COMMIT")
def rollback(self):
print("ROLLBACK")
def execute(self, query):
print(f"Executing: {query}")
if "ERROR" in query:
raise RuntimeError("Database error")
# Usage
connection = MockConnection()
# Successful transaction
with DatabaseTransaction(connection) as tx:
connection.execute("INSERT INTO users VALUES (1, 'Alice')")
connection.execute("INSERT INTO users VALUES (2, 'Bob')")
# Failed transaction
try:
with DatabaseTransaction(connection) as tx:
connection.execute("INSERT INTO users VALUES (3, 'Charlie')")
connection.execute("ERROR: Invalid query") # This will cause rollback
except RuntimeError:
print("Transaction failed and was rolled back")from contextlib import contextmanager
import time
@contextmanager
def timer(operation_name):
"""Context manager to time operations"""
print(f"Starting {operation_name}")
start_time = time.time()
try:
yield
finally:
end_time = time.time()
duration = end_time - start_time
print(f"{operation_name} completed in {duration:.4f} seconds")
@contextmanager
def temporary_setting(obj, attr_name, temp_value):
"""Context manager to temporarily change an object's attribute"""
original_value = getattr(obj, attr_name)
setattr(obj, attr_name, temp_value)
try:
yield obj
finally:
setattr(obj, attr_name, original_value)
class Settings:
def __init__(self):
self.debug = False
self.log_level = "INFO"
# Usage
settings = Settings()
print(f"Original debug setting: {settings.debug}")
with temporary_setting(settings, 'debug', True):
print(f"Temporary debug setting: {settings.debug}")
with timer("Complex operation"):
# Simulate some work
time.sleep(0.1)
result = sum(range(1000000))
print(f"Final debug setting: {settings.debug}")from abc import ABC, abstractmethod
from typing import Dict, List, Any
class Plugin(ABC):
"""Abstract base class for all plugins"""
@property
@abstractmethod
def name(self) -> str:
pass
@property
@abstractmethod
def version(self) -> str:
pass
@abstractmethod
def initialize(self, config: Dict[str, Any]) -> None:
pass
@abstractmethod
def execute(self, data: Any) -> Any:
pass
@abstractmethod
def cleanup(self) -> None:
pass
class PluginManager:
"""Manager for loading and executing plugins"""
def __init__(self):
self._plugins: Dict[str, Plugin] = {}
self._initialized_plugins: List[str] = []
def register_plugin(self, plugin: Plugin) -> None:
self._plugins[plugin.name] = plugin
print(f"Registered plugin: {plugin.name} v{plugin.version}")
def initialize_plugin(self, name: str, config: Dict[str, Any] = None) -> None:
if name in self._plugins:
self._plugins[name].initialize(config or {})
self._initialized_plugins.append(name)
print(f"Initialized plugin: {name}")
else:
raise ValueError(f"Plugin {name} not found")
def execute_plugin(self, name: str, data: Any) -> Any:
if name not in self._initialized_plugins:
raise RuntimeError(f"Plugin {name} not initialized")
return self._plugins[name].execute(data)
def cleanup_all(self) -> None:
for name in self._initialized_plugins:
self._plugins[name].cleanup()
print(f"Cleaned up plugin: {name}")
self._initialized_plugins.clear()
# Concrete plugin implementations
class DataValidationPlugin(Plugin):
@property
def name(self) -> str:
return "DataValidator"
@property
def version(self) -> str:
return "1.0.0"
def initialize(self, config: Dict[str, Any]) -> None:
self.rules = config.get('validation_rules', [])
print(f"DataValidator initialized with {len(self.rules)} rules")
def execute(self, data: Any) -> Any:
print(f"Validating data: {data}")
# Simulate validation
if isinstance(data, dict) and 'email' in data:
if '@' not in data['email']:
raise ValueError("Invalid email format")
return {"valid": True, "data": data}
def cleanup(self) -> None:
self.rules = []
print("DataValidator cleaned up")
class DataTransformPlugin(Plugin):
@property
def name(self) -> str:
return "DataTransformer"
@property
def version(self) -> str:
return "2.1.0"
def initialize(self, config: Dict[str, Any]) -> None:
self.transformations = config.get('transformations', {})
print(f"DataTransformer initialized with {len(self.transformations)} transformations")
def execute(self, data: Any) -> Any:
print(f"Transforming data: {data}")
if isinstance(data, dict):
transformed = data.copy()
for key, value in transformed.items():
if isinstance(value, str):
transformed[key] = value.upper()
return transformed
return data
def cleanup(self) -> None:
self.transformations = {}
print("DataTransformer cleaned up")
# Usage
manager = PluginManager()
# Register plugins
validator = DataValidationPlugin()
transformer = DataTransformPlugin()
manager.register_plugin(validator)
manager.register_plugin(transformer)
# Initialize plugins
manager.initialize_plugin("DataValidator", {"validation_rules": ["email", "required"]})
manager.initialize_plugin("DataTransformer", {"transformations": {"case": "upper"}})
# Execute plugins
test_data = {"name": "alice", "email": "alice@example.com"}
validated_data = manager.execute_plugin("DataValidator", test_data)
transformed_data = manager.execute_plugin("DataTransformer", validated_data["data"])
print(f"Final result: {transformed_data}")
# Cleanup
manager.cleanup_all()from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import json
class APIClient(ABC):
"""Abstract base class for API clients"""
def __init__(self, base_url: str, api_key: Optional[str] = None):
self.base_url = base_url
self.api_key = api_key
@abstractmethod
def authenticate(self) -> bool:
pass
@abstractmethod
def make_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
pass
@abstractmethod
def handle_error(self, error_code: int, error_message: str) -> None:
pass
# Template method
def execute_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
if not self.authenticate():
raise RuntimeError("Authentication failed")
try:
return self.make_request(method, endpoint, data)
except Exception as e:
self.handle_error(500, str(e))
raise
class RESTAPIClient(APIClient):
"""REST API client implementation"""
def __init__(self, base_url: str, api_key: Optional[str] = None):
super().__init__(base_url, api_key)
self.session_token = None
def authenticate(self) -> bool:
if self.api_key:
print(f"Authenticating with API key: {self.api_key[:8]}...")
# Simulate authentication
self.session_token = f"token_{self.api_key[:8]}"
return True
return False
def make_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
url = f"{self.base_url}/{endpoint.lstrip('/')}"
print(f"Making {method} request to {url}")
if data:
print(f"Request data: {json.dumps(data, indent=2)}")
# Simulate API response
if method.upper() == "GET":
return {"status": "success", "data": {"id": 1, "name": "Test"}}
elif method.upper() == "POST":
return {"status": "created", "data": {"id": 2, **data}}
else:
return {"status": "success"}
def handle_error(self, error_code: int, error_message: str) -> None:
print(f"REST API Error {error_code}: {error_message}")
class GraphQLAPIClient(APIClient):
"""GraphQL API client implementation"""
def authenticate(self) -> bool:
if self.api_key:
print(f"Authenticating GraphQL with API key: {self.api_key[:8]}...")
return True
return False
def make_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
# GraphQL typically uses POST for all requests
url = f"{self.base_url}/graphql"
print(f"Making GraphQL request to {url}")
if data and 'query' in data:
print(f"GraphQL Query: {data['query']}")
# Simulate GraphQL response
return {
"data": {"user": {"id": 1, "name": "Alice"}},
"errors": None
}
def handle_error(self, error_code: int, error_message: str) -> None:
print(f"GraphQL Error {error_code}: {error_message}")
# Usage
rest_client = RESTAPIClient("https://api.example.com", "secret_key_123")
graphql_client = GraphQLAPIClient("https://api.example.com", "secret_key_456")
# REST API calls
rest_response = rest_client.execute_request("GET", "/users/1")
print(f"REST Response: {rest_response}")
rest_create = rest_client.execute_request("POST", "/users", {"name": "Bob", "email": "bob@example.com"})
print(f"REST Create: {rest_create}")
# GraphQL API calls
graphql_response = graphql_client.execute_request("POST", "/graphql", {
"query": "query { user(id: 1) { id name email } }"
})
print(f"GraphQL Response: {graphql_response}")from abc import ABC, abstractmethod
from typing import Any, List, Iterator
class DataProcessor(ABC):
"""Abstract base class for data processors"""
@abstractmethod
def process(self, data: Any) -> Any:
pass
@abstractmethod
def can_process(self, data: Any) -> bool:
pass
class Pipeline:
"""Data processing pipeline"""
def __init__(self):
self._processors: List[DataProcessor] = []
def add_processor(self, processor: DataProcessor) -> None:
self._processors.append(processor)
def process(self, data: Any) -> Any:
result = data
for processor in self._processors:
if processor.can_process(result):
print(f"Processing with {processor.__class__.__name__}")
result = processor.process(result)
else:
print(f"Skipping {processor.__class__.__name__} - cannot process data")
return result
# Concrete processors
class StringCleanupProcessor(DataProcessor):
def can_process(self, data: Any) -> bool:
return isinstance(data, str)
def process(self, data: Any) -> Any:
return data.strip().lower()
class NumberValidationProcessor(DataProcessor):
def can_process(self, data: Any) -> bool:
return isinstance(data, (int, float))
def process(self, data: Any) -> Any:
if data < 0:
raise ValueError("Negative numbers not allowed")
return data
class ListSortProcessor(DataProcessor):
def can_process(self, data: Any) -> bool:
return isinstance(data, list)
def process(self, data: Any) -> Any:
return sorted(data)
class DictKeyNormalizationProcessor(DataProcessor):
def can_process(self, data: Any) -> bool:
return isinstance(data, dict)
def process(self, data: Any) -> Any:
return {key.lower().replace(' ', '_'): value for key, value in data.items()}
# Usage
pipeline = Pipeline()
pipeline.add_processor(StringCleanupProcessor())
pipeline.add_processor(NumberValidationProcessor())
pipeline.add_processor(ListSortProcessor())
pipeline.add_processor(DictKeyNormalizationProcessor())
# Test different data types
test_cases = [
" HELLO WORLD ",
42,
[3, 1, 4, 1, 5, 9],
{"First Name": "Alice", "Last Name": "Smith", "Age": 30}
]
for data in test_cases:
print(f"\nProcessing: {data}")
try:
result = pipeline.process(data)
print(f"Result: {result}")
except Exception as e:
print(f"Error: {e}")✅ Good:
from abc import ABC, abstractmethod
class PaymentProcessor(ABC):
"""Clear, focused interface for payment processing"""
@abstractmethod
def process_payment(self, amount: float, currency: str) -> bool:
"""Process a payment and return success status"""
pass
@abstractmethod
def refund_payment(self, transaction_id: str) -> bool:
"""Refund a payment and return success status"""
pass
@abstractmethod
def get_transaction_status(self, transaction_id: str) -> str:
"""Get the status of a transaction"""
pass❌ Bad:
class PaymentThing:
"""Unclear interface with mixed responsibilities"""
def do_stuff(self, data): # Vague method name
pass
def process_payment(self, amount, currency, user_data, shipping_info, tax_info):
# Too many parameters, unclear responsibility
pass
def send_email_and_process_payment(self, email, amount):
# Mixed responsibilities
pass✅ Good:
class EmailService(ABC):
@abstractmethod
def send_email(self, to: str, subject: str, body: str) -> bool:
pass
class SMTPEmailService(EmailService):
def send_email(self, to: str, subject: str, body: str) -> bool:
print(f"Sending email via SMTP to {to}")
return True
class NotificationService:
"""Uses composition instead of inheritance"""
def __init__(self, email_service: EmailService):
self._email_service = email_service
def notify_user(self, user_email: str, message: str) -> bool:
return self._email_service.send_email(
to=user_email,
subject="Notification",
body=message
)
# Usage
email_service = SMTPEmailService()
notification_service = NotificationService(email_service)
notification_service.notify_user("user@example.com", "Hello!")✅ Good:
class DataStorage(ABC):
"""Stable interface that won't change frequently"""
@abstractmethod
def save(self, key: str, data: Any) -> bool:
pass
@abstractmethod
def load(self, key: str) -> Any:
pass
@abstractmethod
def delete(self, key: str) -> bool:
pass
@abstractmethod
def exists(self, key: str) -> bool:
pass❌ Bad:
class DataStorage(ABC):
"""Unstable interface that changes frequently"""
@abstractmethod
def save_with_compression_and_encryption(self, key: str, data: Any, compress: bool, encrypt: bool) -> bool:
# Too specific, will need to change when requirements change
passfrom abc import ABC, abstractmethod
from typing import List, Optional
class SearchEngine(ABC):
"""
Abstract base class for search engines.
This class defines the interface that all search engines must implement.
It provides a consistent way to search, index, and manage documents
regardless of the underlying search technology.
"""
@abstractmethod
def index_document(self, doc_id: str, content: str, metadata: Optional[dict] = None) -> bool:
"""
Index a document for searching.
Args:
doc_id: Unique identifier for the document
content: The text content to be indexed
metadata: Optional metadata associated with the document
Returns:
True if indexing was successful, False otherwise
Raises:
ValueError: If doc_id is empty or None
RuntimeError: If the search engine is not initialized
"""
pass
@abstractmethod
def search(self, query: str, limit: int = 10) -> List[dict]:
"""
Search for documents matching the query.
Args:
query: The search query string
limit: Maximum number of results to return (default: 10)
Returns:
List of dictionaries containing search results with keys:
- 'doc_id': Document identifier
- 'score': Relevance score (0.0 to 1.0)
- 'content': Document content
- 'metadata': Document metadata (if available)
Raises:
ValueError: If query is empty or None
RuntimeError: If the search engine is not initialized
"""
pass❌ Bad:
# Too many layers of abstraction
class AbstractDataAccessLayerFactoryInterface(ABC):
@abstractmethod
def create_abstract_data_access_layer(self) -> 'AbstractDataAccessLayer':
pass
class AbstractDataAccessLayer(ABC):
@abstractmethod
def get_abstract_data_repository(self) -> 'AbstractDataRepository':
pass
class AbstractDataRepository(ABC):
@abstractmethod
def find_by_abstract_criteria(self, criteria: 'AbstractCriteria') -> 'AbstractResult':
pass✅ Good:
# Simple, clear abstraction
class UserRepository(ABC):
@abstractmethod
def find_by_id(self, user_id: int) -> Optional['User']:
pass
@abstractmethod
def save(self, user: 'User') -> bool:
pass❌ Bad:
class DatabaseConnection(ABC):
@abstractmethod
def execute_sql(self, sql: str) -> Any:
# Leaks SQL implementation detail
pass
@abstractmethod
def get_mysql_connection(self) -> Any:
# Leaks specific database type
pass✅ Good:
class DatabaseConnection(ABC):
@abstractmethod
def execute_query(self, query: 'Query') -> 'Result':
# Generic interface that doesn't leak implementation
pass
@abstractmethod
def begin_transaction(self) -> 'Transaction':
# Database-agnostic operation
pass❌ Bad:
class Shape(ABC):
@abstractmethod
def area(self) -> float:
pass
# Missing other essential methods that all shapes should have
# like perimeter, draw, etc.✅ Good:
class Shape(ABC):
@abstractmethod
def area(self) -> float:
pass
@abstractmethod
def perimeter(self) -> float:
pass
@abstractmethod
def contains_point(self, x: float, y: float) -> bool:
pass
def describe(self) -> str:
# Concrete method using abstract methods
return f"{self.__class__.__name__} with area {self.area():.2f}"-
Basic Abstract Class
- Create an abstract
Animalclass with abstract methodsmake_sound()andmove() - Implement concrete classes
Dog,Cat, andBird - Add a concrete method
describe()that uses the abstract methods
- Create an abstract
-
Simple Interface
- Create an abstract
Drawableinterface with adraw()method - Implement classes
Circle,Rectangle, andTriangle - Create a function that can draw any drawable object
- Create an abstract
-
Template Method Pattern
- Create an abstract
DataProcessorclass with a template methodprocess_file() - The template method should call abstract methods:
read_file(),process_data(),write_file() - Implement concrete processors for CSV and JSON files
- Create an abstract
-
Strategy Pattern
- Create an abstract
CompressionStrategyclass - Implement different compression algorithms (ZIP, GZIP, etc.)
- Create a
FileCompressorclass that uses these strategies
- Create an abstract
-
Plugin System
- Design a plugin architecture using abstract base classes
- Create a plugin manager that can load, initialize, and execute plugins
- Implement sample plugins for different data transformations
-
Context Manager Framework
- Create an abstract base class for context managers
- Implement concrete context managers for different resources
- Add support for nested context managers
-
Metaclass-Based Framework
- Create a metaclass that automatically registers classes
- Build a command pattern framework using the metaclass
- Add automatic validation and documentation generation
-
Protocol-Based Design
- Design a system using
typing.Protocolfor structural subtyping - Create multiple implementations that satisfy the protocols
- Demonstrate how protocols enable duck typing with type safety
- Design a system using
Abstraction is a powerful principle that helps you:
- Hide complexity while exposing essential functionality
- Create flexible designs that can accommodate change
- Establish contracts between different parts of your system
- Enable polymorphism and code reuse
- Improve maintainability through clear interfaces
- Use ABC for formal contracts: When you need to enforce specific methods
- Leverage protocols for flexibility: When you want duck typing with type hints
- Apply metaclasses sparingly: Only when you need to control class creation
- Design stable interfaces: Keep abstractions focused and unlikely to change
- Document thoroughly: Abstract classes need clear documentation
- Practice with the provided exercises
- Study real-world frameworks to see abstraction in action
- Experiment with different abstraction techniques
- Focus on creating clean, maintainable interfaces
Remember: Good abstraction is about finding the right level of detail—not too specific that it becomes inflexible, not too general that it becomes meaningless.