7 инженерных решений ООП Python для реальных проектов

# Вместо одного монолитного класса
class EventProcessor:
def __init__(self, storage, validator, enricher, notifier):
self.storage = storage
self.validator = validator
self.enricher = enricher
self.notifier = notifier

def process(self, event):
if self.validator.is_valid(event):
enriched = self.enricher.enrich(event)
self.storage.store(enriched)
self.notifier.notify(enriched)
return True
return False

from abc import ABC, abstractmethod
from typing import Dict, Any, List

class MedicalRecord(ABC):
@abstractmethod
def get_patient_id(self) -> str:
pass

@abstractmethod
def get_access_level(self) -> int:
pass

@abstractmethod
def to_audit_log(self) -> Dict[str, Any]:
pass

@abstractmethod
def to_dict(self) -> Dict[str, Any]:
pass

class PatientDiagnosis(MedicalRecord):
def __init__(self, patient_id: str, diagnosis_code: str,
notes: str, diagnosed_by: str):
self.patient_id = patient_id
self.diagnosis_code = diagnosis_code
self._notes = notes # Приватные данные
self.diagnosed_by = diagnosed_by

def get_patient_id(self) -> str:
return self.patient_id

def get_access_level(self) -> int:
return 3 # Требует высокого уровня доступа

def to_audit_log(self) -> Dict[str, Any]:
return {
"action": "access_diagnosis",
"patient_id": self.patient_id,
"diagnosis_code": self.diagnosis_code,
"accessed_by": "CURRENT_USER",
"timestamp": "CURRENT_TIME"
}

def to_dict(self) -> Dict[str, Any]:
return {
"patient_id": self.patient_id,
"diagnosis_code": self.diagnosis_code,
"notes": self._notes,
"diagnosed_by": self.diagnosed_by
}

class ReadOnlyRecord(MedicalRecord):
@abstractmethod
def to_dict(self) -> Dict[str, Any]:
pass

def is_modifiable(self) -> bool:
return False

class ModifiableRecord(MedicalRecord):
@abstractmethod
def to_dict(self) -> Dict[str, Any]:
pass

@abstractmethod
def update(self, data: Dict[str, Any]) -> None:
pass

def is_modifiable(self) -> bool:
return True

# Антипаттерн – слишком глубокая иерархия наследования
class BaseUser:
# Общая функциональность

class EmailUser(BaseUser):
# Функциональность пользователей с email

class SocialUser(BaseUser):
# Функциональность пользователей из соц. сетей

class PremiumEmailUser(EmailUser):
# Премиум пользователи с email

class PremiumSocialUser(SocialUser):
# Премиум пользователи из соц. сетей

class User:
def __init__(self, user_id, auth_provider=None, subscription=None):
self.user_id = user_id
self.auth_provider = auth_provider or EmailAuthProvider()
self.subscription = subscription or FreeSubscription()

def authenticate(self, credentials):
return self.auth_provider.authenticate(credentials)

def can_access(self, resource):
return self.subscription.has_access(resource)

# Различные провайдеры аутентификации
class AuthProvider(ABC):
@abstractmethod
def authenticate(self, credentials):
pass

class EmailAuthProvider(AuthProvider):
def authenticate(self, credentials):
# Email-специфичная аутентификация
return verify_email_password(credentials)

class SocialAuthProvider(AuthProvider):
def __init__(self, provider):
self.provider = provider

def authenticate(self, credentials):
# Аутентификация через указанную соц. сеть
return verify_social_token(self.provider, credentials)

# Различные типы подписок
class Subscription(ABC):
@abstractmethod
def has_access(self, resource):
pass

class FreeSubscription(Subscription):
def has_access(self, resource):
return resource.is_free()

class PremiumSubscription(Subscription):
def has_access(self, resource):
return True

# Премиум пользователь с email авторизацией
premium_email_user = User(
user_id="12345",
auth_provider=EmailAuthProvider(),
subscription=PremiumSubscription()
)

# Бесплатный пользователь через социальную сеть
free_social_user = User(
user_id="67890",
auth_provider=SocialAuthProvider(provider="twitter"),
subscription=FreeSubscription()
)

class ProcessorRegistry(type):
processors = {}

def __new__(mcs, name, bases, namespace):
cls = super().__new__(mcs, name, bases, namespace)
if name != 'DataProcessor': # не регистрируем базовый класс
processor_type = namespace.get('PROCESSOR_TYPE')
if processor_type:
mcs.processors[processor_type] = cls
return cls

@classmethod
def get_processor(mcs, processor_type):
return mcs.processors.get(processor_type)

class DataProcessor(metaclass=ProcessorRegistry):
"""Базовый класс для всех обработчиков данных"""

def process(self, data):
raise NotImplementedError

class JSONProcessor(DataProcessor):
PROCESSOR_TYPE = 'json'

def process(self, data):
# Обработка JSON данных
return json.loads(data)

class XMLProcessor(DataProcessor):
PROCESSOR_TYPE = 'xml'

def process(self, data):
# Обработка XML данных
return xmltodict.parse(data)

# Использование:
def process_data(data, format_type):
processor_cls = ProcessorRegistry.get_processor(format_type)
if not processor_cls:
raise ValueError(f"Unsupported format: {format_type}")

processor = processor_cls()
return processor.process(data)

class ValidatedField:
def __init__(self, name, validator=None):
self.name = name
self.private_name = f"_{name}"
self.validator = validator

def __get__(self, instance, owner):
if instance is None:
return self
return getattr(instance, self.private_name, None)

def __set__(self, instance, value):
if self.validator and not self.validator(value):
raise ValueError(f"Invalid value for {self.name}: {value}")
setattr(instance, self.private_name, value)

def email_validator(value):
import re
pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
return bool(re.match(pattern, value))

def positive_integer(value):
return isinstance(value, int) and value > 0

class User:
email = ValidatedField('email', validator=email_validator)
age = ValidatedField('age', validator=positive_integer)

def __init__(self, email=None, age=None):
self.email = email
self.age = age

class ComputedProperty:
def __init__(self, calculation_func, dependencies=None):
self.calculation_func = calculation_func
self.dependencies = dependencies or []
self.cache_attr_name = f"_{calculation_func.__name__}_cache"

def __get__(self, instance, owner):
if instance is None:
return self

# Проверяем, есть ли кэшированное значение
if hasattr(instance, self.cache_attr_name):
return getattr(instance, self.cache_attr_name)

# Вычисляем и кэшируем значение
result = self.calculation_func(instance)
setattr(instance, self.cache_attr_name, result)
return result

def invalidate_cache(self, instance):
if hasattr(instance, self.cache_attr_name):
delattr(instance, self.cache_attr_name)

class Product:
def __init__(self, base_price, tax_rate, discount=0):
self.base_price = base_price
self.tax_rate = tax_rate
self.discount = discount

@ComputedProperty
def final_price(self):
"""Вычисляет итоговую цену с учетом налогов и скидок"""
price_after_discount = self.base_price * (1 – self.discount / 100)
return price_after_discount * (1 + self.tax_rate / 100)

class UserRepository:
def get_by_id(self, user_id):
try:
return User.objects.get(id=user_id)
except User.DoesNotExist:
return None

def get_active_users(self):
return User.objects.filter(is_active=True)

def save(self, user):
user.save()
return user

class UserService:
def __init__(self, user_repository=None):
self.user_repository = user_repository or UserRepository()

def activate_user(self, user_id):
user = self.user_repository.get_by_id(user_id)
if not user:
raise ValueError(f"User with id {user_id} not found")

user.is_active = True
self.user_repository.save(user)
return user

from abc import ABC, abstractmethod

class PaymentStrategy(ABC):
@abstractmethod
def pay(self, amount):
pass

class CreditCardStrategy(PaymentStrategy):
def __init__(self, card_number, cvv, expiry_date):
self.card_number = card_number
self.cvv = cvv
self.expiry_date = expiry_date

def pay(self, amount):
# Логика оплаты кредитной картой
print(f"Paid {amount} using Credit Card: {self.card_number}")
return True

class PayPalStrategy(PaymentStrategy):
def __init__(self, email, password):
self.email = email
self.password = password

def pay(self, amount):
# Логика оплаты через PayPal
print(f"Paid {amount} using PayPal: {self.email}")
return True

class ShoppingCart:
def __init__(self):
self.items = []
self.payment_strategy = None

def add_item(self, item):
self.items.append(item)

def calculate_total(self):
return sum(item.price for item in self.items)

def set_payment_strategy(self, payment_strategy):
self.payment_strategy = payment_strategy

def checkout(self):
if not self.payment_strategy:
raise ValueError("Payment strategy not set")

amount = self.calculate_total()
return self.payment_strategy.pay(amount)

class EventManager:
def __init__(self):
self.subscribers = {}

def subscribe(self, event_type, listener):
if event_type not in self.subscribers:
self.subscribers[event_type] = []
self.subscribers[event_type].append(listener)

def unsubscribe(self, event_type, listener):
if event_type in self.subscribers:
self.subscribers[event_type].remove(listener)

def notify(self, event_type, data):
if event_type not in self.subscribers:
return

for listener in self.subscribers[event_type]:
listener(data)

# Пример использования в реальном приложении
class OrderService:
def __init__(self, event_manager):
self.event_manager = event_manager

def place_order(self, order):
# Бизнес-логика размещения заказа
order_id = self._save_order(order)

# Уведомляем подписчиков о новом заказе
self.event_manager.notify("order.created", {
"order_id": order_id,
"customer_id": order.customer_id,
"amount": order.total_amount
})

return order_id

def _save_order(self, order):
# Логика сохранения заказа в базу данных
return 12345 # ID сохраненного заказа

# Обработчики событий
def notify_customer(order_data):
print(f"Sending notification to customer {order_data['customer_id']}")

def update_inventory(order_data):
print(f"Updating inventory for order {order_data['order_id']}")

def generate_invoice(order_data):
print(f"Generating invoice for ${order_data['amount']}")

# Настройка системы
event_manager = EventManager()
event_manager.subscribe("order.created", notify_customer)
event_manager.subscribe("order.created", update_inventory)
event_manager.subscribe("order.created", generate_invoice)

order_service = OrderService(event_manager)

class Command(ABC):
@abstractmethod
def execute(self):
pass

class SendEmailCommand(Command):
def __init__(self, to, subject, body):
self.to = to
self.subject = subject
self.body = body

def execute(self):
# Логика отправки email
print(f"Sending email to {self.to}")
return True

class GenerateReportCommand(Command):
def __init__(self, report_type, user_id, parameters):
self.report_type = report_type
self.user_id = user_id
self.parameters = parameters

def execute(self):
# Логика генерации отчета
print(f"Generating {self.report_type} report for user {self.user_id}")
return {"report_url": f"/reports/{self.report_type}_{self.user_id}.pdf"}

# Интеграция с Celery
@app.task
def execute_command(command_data):
command_type = command_data.pop('command_type')

commands = {
'send_email': SendEmailCommand,
'generate_report': GenerateReportCommand
}

if command_type not in commands:
raise ValueError(f"Unknown command type: {command_type}")

command = commands[command_type](**command_data)
return command.execute()

# Использование
def schedule_email(to, subject, body):
execute_command.delay({
'command_type': 'send_email',
'to': to,
'subject': subject,
'body': body
})