Bài 19: Python Best Practices (Phần 2)
SOLID Principles
SOLID là 5 principles giúp code maintainable, scalable, và testable.
1. Single Responsibility Principle (SRP)
"A class should have only one reason to change."
Mỗi class chỉ làm một việc duy nhất.
❌ Violating SRP
class User: """User class doing too many things.""" def __init__(self, username, email): self.username = username self.email = email def save_to_database(self): """Save user to database - DB responsibility""" # Database logic pass def send_welcome_email(self): """Send email - Email responsibility""" # Email logic pass def generate_report(self): """Generate report - Reporting responsibility""" # Report logic pass def validate_email(self): """Validate email - Validation responsibility""" # Validation logic pass
✅ Following SRP
class User: """User model - only handles user data.""" def __init__(self, username: str, email: str): self.username = username self.email = email def get_display_name(self) -> str: """Get user display name.""" return self.username class UserRepository: """Handles user database operations.""" def save(self, user: User): """Save user to database.""" # Database logic pass def find_by_username(self, username: str) -> User: """Find user by username.""" # Database query pass class EmailService: """Handles email operations.""" def send_welcome_email(self, user: User): """Send welcome email to user.""" # Email logic pass class UserValidator: """Validates user data.""" def validate_email(self, email: str) -> bool: """Validate email format.""" import re pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$' return bool(re.match(pattern, email)) class ReportGenerator: """Generates reports.""" def generate_user_report(self, user: User) -> str: """Generate user report.""" return f"Report for {user.username}" # Sử dụnguser = User("john_doe", "[email protected]")repo = UserRepository()email_service = EmailService() repo.save(user)email_service.send_welcome_email(user)
2. Open/Closed Principle (OCP)
"Software entities should be open for extension but closed for modification."
Extend functionality mà không modify existing code.
❌ Violating OCP
class PaymentProcessor: """Payment processor that needs modification for new payment methods.""" def process_payment(self, amount: float, method: str): if method == "credit_card": # Credit card logic print(f"Processing ${amount} via credit card") elif method == "paypal": # PayPal logic print(f"Processing ${amount} via PayPal") elif method == "bitcoin": # Need to modify for new method! # Bitcoin logic print(f"Processing ${amount} via Bitcoin") else: raise ValueError(f"Unknown payment method: {method}")
✅ Following OCP
from abc import ABC, abstractmethod class PaymentMethod(ABC): """Abstract payment method.""" @abstractmethod def process(self, amount: float): """Process payment.""" pass class CreditCardPayment(PaymentMethod): """Credit card payment implementation.""" def process(self, amount: float): print(f"Processing ${amount} via credit card") class PayPalPayment(PaymentMethod): """PayPal payment implementation.""" def process(self, amount: float): print(f"Processing ${amount} via PayPal") class BitcoinPayment(PaymentMethod): """Bitcoin payment implementation - NEW, no modification needed!""" def process(self, amount: float): print(f"Processing ${amount} via Bitcoin") class PaymentProcessor: """Payment processor - closed for modification, open for extension.""" def process_payment(self, amount: float, method: PaymentMethod): """Process payment using any payment method.""" method.process(amount) # Sử dụngprocessor = PaymentProcessor() # Existing methodsprocessor.process_payment(100, CreditCardPayment())processor.process_payment(200, PayPalPayment()) # New method - no changes to PaymentProcessor!processor.process_payment(300, BitcoinPayment())
3. Liskov Substitution Principle (LSP)
"Subtypes must be substitutable for their base types."
Child classes không được thay đổi behavior của parent class.
❌ Violating LSP
class Bird: """Base bird class.""" def fly(self): """All birds can fly.""" print("Flying...") class Sparrow(Bird): """Sparrow can fly - OK.""" pass class Penguin(Bird): """Penguin cannot fly - VIOLATES LSP!""" def fly(self): raise Exception("Penguins cannot fly!") # Problem: Cannot substitute Penguin for Birddef make_bird_fly(bird: Bird): bird.fly() make_bird_fly(Sparrow()) # ✅ Worksmake_bird_fly(Penguin()) # ❌ Breaks!
✅ Following LSP
from abc import ABC, abstractmethod class Bird(ABC): """Base bird class.""" @abstractmethod def move(self): """All birds can move.""" pass class FlyingBird(Bird): """Birds that can fly.""" def move(self): self.fly() def fly(self): print("Flying...") class FlightlessBird(Bird): """Birds that cannot fly.""" def move(self): self.walk() def walk(self): print("Walking...") class Sparrow(FlyingBird): """Sparrow implementation.""" pass class Penguin(FlightlessBird): """Penguin implementation.""" pass # Now substitutabledef make_bird_move(bird: Bird): bird.move() make_bird_move(Sparrow()) # ✅ Flying...make_bird_move(Penguin()) # ✅ Walking...
4. Interface Segregation Principle (ISP)
"Clients should not be forced to depend on interfaces they don't use."
Chia interfaces thành smaller, specific ones.
❌ Violating ISP
from abc import ABC, abstractmethod class Worker(ABC): """Fat interface - not all workers need all methods.""" @abstractmethod def work(self): pass @abstractmethod def eat(self): pass @abstractmethod def sleep(self): pass class Human(Worker): """Human needs all methods - OK.""" def work(self): print("Working...") def eat(self): print("Eating...") def sleep(self): print("Sleeping...") class Robot(Worker): """Robot doesn't eat/sleep - FORCED to implement!""" def work(self): print("Working...") def eat(self): raise NotImplementedError("Robots don't eat") def sleep(self): raise NotImplementedError("Robots don't sleep")
✅ Following ISP
from abc import ABC, abstractmethod class Workable(ABC): """Work interface.""" @abstractmethod def work(self): pass class Eatable(ABC): """Eat interface.""" @abstractmethod def eat(self): pass class Sleepable(ABC): """Sleep interface.""" @abstractmethod def sleep(self): pass class Human(Workable, Eatable, Sleepable): """Human implements all interfaces.""" def work(self): print("Working...") def eat(self): print("Eating...") def sleep(self): print("Sleeping...") class Robot(Workable): """Robot only implements Workable.""" def work(self): print("Working 24/7...") # Usagedef make_work(worker: Workable): worker.work() def make_eat(eater: Eatable): eater.eat() human = Human()robot = Robot() make_work(human) # ✅make_work(robot) # ✅make_eat(human) # ✅# make_eat(robot) # ❌ Type error - Robot is not Eatable
5. Dependency Inversion Principle (DIP)
"Depend on abstractions, not on concretions."
High-level modules không nên depend on low-level modules.
❌ Violating DIP
class MySQLDatabase: """Concrete MySQL implementation.""" def connect(self): print("Connecting to MySQL...") def query(self, sql: str): print(f"Executing MySQL query: {sql}") class UserService: """UserService depends on concrete MySQLDatabase.""" def __init__(self): self.db = MySQLDatabase() # Tight coupling! def get_user(self, user_id: int): self.db.connect() self.db.query(f"SELECT * FROM users WHERE id = {user_id}") # Problem: Hard to switch databases or test
✅ Following DIP
from abc import ABC, abstractmethod class Database(ABC): """Database abstraction.""" @abstractmethod def connect(self): pass @abstractmethod def query(self, sql: str): pass class MySQLDatabase(Database): """MySQL implementation.""" def connect(self): print("Connecting to MySQL...") def query(self, sql: str): print(f"Executing MySQL query: {sql}") class PostgreSQLDatabase(Database): """PostgreSQL implementation.""" def connect(self): print("Connecting to PostgreSQL...") def query(self, sql: str): print(f"Executing PostgreSQL query: {sql}") class UserService: """UserService depends on Database abstraction.""" def __init__(self, database: Database): self.db = database # Inject dependency! def get_user(self, user_id: int): self.db.connect() self.db.query(f"SELECT * FROM users WHERE id = {user_id}") # Usage - easy to switch implementationsmysql_service = UserService(MySQLDatabase())postgres_service = UserService(PostgreSQLDatabase()) # Easy to test with mockclass MockDatabase(Database): def connect(self): pass def query(self, sql: str): return {"id": 1, "name": "Test"} test_service = UserService(MockDatabase())
Design Patterns
1. Singleton Pattern
Ensure một class chỉ có một instance duy nhất.
class Singleton: """Singleton implementation using __new__.""" _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance # Thread-safe Singletonimport threading class ThreadSafeSingleton: """Thread-safe singleton.""" _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: # Double-checked locking if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance # Singleton decoratordef singleton(cls): """Singleton decorator.""" instances = {} def get_instance(*args, **kwargs): if cls not in instances: instances[cls] = cls(*args, **kwargs) return instances[cls] return get_instance @singletonclass DatabaseConnection: """Database connection as singleton.""" def __init__(self): print("Creating database connection...") self.connection = "connected" # Usagedb1 = DatabaseConnection()db2 = DatabaseConnection()print(db1 is db2) # True - same instance
2. Factory Pattern
Tạo objects mà không specify exact class.
from abc import ABC, abstractmethodfrom typing import Dict, Type class Animal(ABC): """Abstract animal.""" @abstractmethod def speak(self) -> str: pass class Dog(Animal): """Dog implementation.""" def speak(self) -> str: return "Woof!" class Cat(Animal): """Cat implementation.""" def speak(self) -> str: return "Meow!" class Cow(Animal): """Cow implementation.""" def speak(self) -> str: return "Moo!" class AnimalFactory: """Factory to create animals.""" _animals: Dict[str, Type[Animal]] = { 'dog': Dog, 'cat': Cat, 'cow': Cow, } @classmethod def create_animal(cls, animal_type: str) -> Animal: """Create animal by type.""" animal_class = cls._animals.get(animal_type.lower()) if not animal_class: raise ValueError(f"Unknown animal type: {animal_type}") return animal_class() @classmethod def register_animal(cls, animal_type: str, animal_class: Type[Animal]): """Register new animal type.""" cls._animals[animal_type.lower()] = animal_class # Usagefactory = AnimalFactory() dog = factory.create_animal('dog')print(dog.speak()) # Woof! cat = factory.create_animal('cat')print(cat.speak()) # Meow! # Register new animal typeclass Bird(Animal): def speak(self) -> str: return "Chirp!" factory.register_animal('bird', Bird)bird = factory.create_animal('bird')print(bird.speak()) # Chirp!
3. Observer Pattern
Notify observers khi state changes.
from abc import ABC, abstractmethodfrom typing import List class Observer(ABC): """Observer interface.""" @abstractmethod def update(self, subject: 'Subject'): """Called when subject changes.""" pass class Subject: """Subject that notifies observers.""" def __init__(self): self._observers: List[Observer] = [] self._state = None def attach(self, observer: Observer): """Attach observer.""" if observer not in self._observers: self._observers.append(observer) def detach(self, observer: Observer): """Detach observer.""" self._observers.remove(observer) def notify(self): """Notify all observers.""" for observer in self._observers: observer.update(self) @property def state(self): return self._state @state.setter def state(self, value): """Set state and notify observers.""" self._state = value self.notify() class EmailObserver(Observer): """Send email when subject changes.""" def update(self, subject: Subject): print(f"Email: State changed to {subject.state}") class SMSObserver(Observer): """Send SMS when subject changes.""" def update(self, subject: Subject): print(f"SMS: State changed to {subject.state}") class LogObserver(Observer): """Log when subject changes.""" def update(self, subject: Subject): print(f"LOG: State changed to {subject.state}") # Usagesubject = Subject() email_observer = EmailObserver()sms_observer = SMSObserver()log_observer = LogObserver() subject.attach(email_observer)subject.attach(sms_observer)subject.attach(log_observer) # Change state - all observers notifiedsubject.state = "active"# Output:# Email: State changed to active# SMS: State changed to active# LOG: State changed to active # Detach observersubject.detach(sms_observer)subject.state = "inactive"# Output:# Email: State changed to inactive# LOG: State changed to inactive
4. Strategy Pattern
Encapsulate algorithms và make them interchangeable.
from abc import ABC, abstractmethod class SortStrategy(ABC): """Strategy interface for sorting.""" @abstractmethod def sort(self, data: list) -> list: pass class BubbleSort(SortStrategy): """Bubble sort implementation.""" def sort(self, data: list) -> list: data = data.copy() n = len(data) for i in range(n): for j in range(0, n-i-1): if data[j] > data[j+1]: data[j], data[j+1] = data[j+1], data[j] return data class QuickSort(SortStrategy): """Quick sort implementation.""" def sort(self, data: list) -> list: 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(SortStrategy): """Merge sort implementation.""" def sort(self, data: list) -> list: 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: list, right: list) -> list: 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 that uses sorting strategy.""" def __init__(self, strategy: SortStrategy): self._strategy = strategy def set_strategy(self, strategy: SortStrategy): """Change sorting strategy at runtime.""" self._strategy = strategy def sort(self, data: list) -> list: """Sort data using current strategy.""" return self._strategy.sort(data) # Usagedata = [64, 34, 25, 12, 22, 11, 90] sorter = Sorter(BubbleSort())print(sorter.sort(data)) # Bubble sort sorter.set_strategy(QuickSort())print(sorter.sort(data)) # Quick sort sorter.set_strategy(MergeSort())print(sorter.sort(data)) # Merge sort
5. Decorator Pattern
Add behavior to objects dynamically.
from abc import ABC, abstractmethod class Coffee(ABC): """Coffee interface.""" @abstractmethod def cost(self) -> float: pass @abstractmethod def description(self) -> str: pass class SimpleCoffee(Coffee): """Basic coffee.""" def cost(self) -> float: return 2.0 def description(self) -> str: return "Simple Coffee" class CoffeeDecorator(Coffee): """Base decorator.""" def __init__(self, coffee: Coffee): self._coffee = coffee def cost(self) -> float: return self._coffee.cost() def description(self) -> str: return self._coffee.description() class MilkDecorator(CoffeeDecorator): """Add milk.""" def cost(self) -> float: return self._coffee.cost() + 0.5 def description(self) -> str: return self._coffee.description() + ", Milk" class SugarDecorator(CoffeeDecorator): """Add sugar.""" def cost(self) -> float: return self._coffee.cost() + 0.2 def description(self) -> str: return self._coffee.description() + ", Sugar" class VanillaDecorator(CoffeeDecorator): """Add vanilla.""" def cost(self) -> float: return self._coffee.cost() + 0.7 def description(self) -> str: return self._coffee.description() + ", Vanilla" # Usagecoffee = SimpleCoffee()print(f"{coffee.description()}: ${coffee.cost()}")# Simple Coffee: $2.0 # Add milkcoffee = MilkDecorator(coffee)print(f"{coffee.description()}: ${coffee.cost()}")# Simple Coffee, Milk: $2.5 # Add sugarcoffee = SugarDecorator(coffee)print(f"{coffee.description()}: ${coffee.cost()}")# Simple Coffee, Milk, Sugar: $2.7 # Add vanillacoffee = VanillaDecorator(coffee)print(f"{coffee.description()}: ${coffee.cost()}")# Simple Coffee, Milk, Sugar, Vanilla: $3.4
2 Ứng Dụng Thực Tế
1. Plugin System với Factory và Strategy
from abc import ABC, abstractmethodfrom typing import Dict, Type, Any class Plugin(ABC): """Plugin interface.""" @abstractmethod def execute(self, data: Any) -> Any: """Execute plugin logic.""" pass @abstractmethod def get_name(self) -> str: """Get plugin name.""" pass class DataValidationPlugin(Plugin): """Data validation plugin.""" def execute(self, data: Dict) -> Dict: # Validate data if not data.get('email'): raise ValueError("Email is required") return data def get_name(self) -> str: return "data_validation" class DataTransformPlugin(Plugin): """Data transformation plugin.""" def execute(self, data: Dict) -> Dict: # Transform data data['email'] = data['email'].lower() return data def get_name(self) -> str: return "data_transform" class DataEnrichmentPlugin(Plugin): """Data enrichment plugin.""" def execute(self, data: Dict) -> Dict: # Enrich data data['processed_at'] = "2024-01-01" return data def get_name(self) -> str: return "data_enrichment" class PluginFactory: """Factory to create and manage plugins.""" _plugins: Dict[str, Type[Plugin]] = {} @classmethod def register(cls, plugin_class: Type[Plugin]): """Register plugin.""" plugin = plugin_class() cls._plugins[plugin.get_name()] = plugin_class @classmethod def create(cls, plugin_name: str) -> Plugin: """Create plugin instance.""" plugin_class = cls._plugins.get(plugin_name) if not plugin_class: raise ValueError(f"Unknown plugin: {plugin_name}") return plugin_class() @classmethod def get_available_plugins(cls) -> list: """Get list of available plugins.""" return list(cls._plugins.keys()) class PluginPipeline: """Execute plugins in sequence.""" def __init__(self): self.plugins: list[Plugin] = [] def add_plugin(self, plugin: Plugin): """Add plugin to pipeline.""" self.plugins.append(plugin) def execute(self, data: Any) -> Any: """Execute all plugins in order.""" result = data for plugin in self.plugins: print(f"Executing plugin: {plugin.get_name()}") result = plugin.execute(result) return result # Register pluginsPluginFactory.register(DataValidationPlugin)PluginFactory.register(DataTransformPlugin)PluginFactory.register(DataEnrichmentPlugin) # Usagepipeline = PluginPipeline() # Add plugins dynamicallyfor plugin_name in ['data_validation', 'data_transform', 'data_enrichment']: plugin = PluginFactory.create(plugin_name) pipeline.add_plugin(plugin) # Process datadata = {'email': '[email protected]'}result = pipeline.execute(data)print(result)# {'email': '[email protected]', 'processed_at': '2024-01-01'}
2. Event-Driven Architecture với Observer
from abc import ABC, abstractmethodfrom typing import Dict, List, Anyfrom datetime import datetime class Event: """Event data.""" def __init__(self, event_type: str, data: Dict[str, Any]): self.type = event_type self.data = data self.timestamp = datetime.now() class EventHandler(ABC): """Event handler interface.""" @abstractmethod def handle(self, event: Event): """Handle event.""" pass @abstractmethod def can_handle(self, event_type: str) -> bool: """Check if handler can handle event type.""" pass class UserCreatedHandler(EventHandler): """Handle user created events.""" def handle(self, event: Event): print(f"User created: {event.data['username']}") # Send welcome email, create profile, etc. def can_handle(self, event_type: str) -> bool: return event_type == "user.created" class UserUpdatedHandler(EventHandler): """Handle user updated events.""" def handle(self, event: Event): print(f"User updated: {event.data['username']}") # Update cache, notify watchers, etc. def can_handle(self, event_type: str) -> bool: return event_type == "user.updated" class OrderPlacedHandler(EventHandler): """Handle order placed events.""" def handle(self, event: Event): print(f"Order placed: #{event.data['order_id']}") # Process payment, send confirmation, update inventory def can_handle(self, event_type: str) -> bool: return event_type == "order.placed" class EventBus: """Event bus for publishing and subscribing to events.""" def __init__(self): self._handlers: List[EventHandler] = [] self._event_log: List[Event] = [] def subscribe(self, handler: EventHandler): """Subscribe handler to events.""" self._handlers.append(handler) def publish(self, event: Event): """Publish event to all relevant handlers.""" print(f"\n[EventBus] Publishing event: {event.type}") # Log event self._event_log.append(event) # Notify handlers handled = False for handler in self._handlers: if handler.can_handle(event.type): handler.handle(event) handled = True if not handled: print(f"[EventBus] No handlers for event: {event.type}") def get_event_log(self) -> List[Event]: """Get event history.""" return self._event_log # Usageevent_bus = EventBus() # Subscribe handlersevent_bus.subscribe(UserCreatedHandler())event_bus.subscribe(UserUpdatedHandler())event_bus.subscribe(OrderPlacedHandler()) # Publish eventsevent_bus.publish(Event("user.created", { "username": "john_doe", "email": "[email protected]"})) event_bus.publish(Event("order.placed", { "order_id": "12345", "total": 99.99})) event_bus.publish(Event("user.updated", { "username": "john_doe", "email": "[email protected]"})) # View event logprint(f"\nTotal events: {len(event_bus.get_event_log())}")
Best Practices Summary
Code Quality Checklist
# ✅ Follow PEP 8# ✅ Use type hints# ✅ Write docstrings# ✅ Keep functions small (< 50 lines)# ✅ Single responsibility per class# ✅ Dependency injection# ✅ Write tests# ✅ Handle exceptions properly# ✅ Use meaningful names# ✅ Don't repeat yourself (DRY)# ✅ Keep it simple (KISS)# ✅ You aren't gonna need it (YAGNI) def good_function(user_id: int, email: str) -> bool: """ Send welcome email to user. Args: user_id: User's unique identifier email: User's email address Returns: True if email sent successfully, False otherwise Raises: ValueError: If email format is invalid """ if not _is_valid_email(email): raise ValueError(f"Invalid email: {email}") try: _send_email(email, "Welcome!") return True except EmailError as e: logger.error(f"Failed to send email: {e}") return False
Bài Tập Thực Hành
Bài 1: Refactor with SOLID
Refactor một class lớn để follow SOLID principles.
Bài 2: Implement Singleton
Tạo thread-safe Singleton cho database connection.
Bài 3: Factory Pattern
Implement factory pattern cho notification system (email, SMS, push).
Bài 4: Observer Pattern
Tạo event system với multiple observers.
Bài 5: Strategy Pattern
Implement strategy pattern cho data export (CSV, JSON, XML).
Tóm Tắt
Trong Part 2 chúng ta đã học:
✅ SOLID Principles:
- Single Responsibility (SRP)
- Open/Closed (OCP)
- Liskov Substitution (LSP)
- Interface Segregation (ISP)
- Dependency Inversion (DIP)
✅ Design Patterns:
- Singleton (single instance)
- Factory (object creation)
- Observer (event notification)
- Strategy (interchangeable algorithms)
- Decorator (add behavior dynamically)
✅ Real Applications:
- Plugin system
- Event-driven architecture
✅ Best Practices Checklist
Best practices giúp code maintainable, testable, và scalable! 🚀
Bài tiếp theo: Bài 20: Working with APIs - Học cách làm việc với REST APIs! 🌐