Что писал на ООП?

Опыт с объектно-ориентированным программированием

ООП — это основа моей разработки. Я использую его везде: от web приложений до микросервисов.

Проекты с применением ООП

1. E-commerce платформа (Django ORM)

Структура объектов для интернет-магазина:

from django.db import models
from abc import ABC, abstractmethod

# Domain Models
class Product(models.Model):
    """Продукт в каталоге"""
    name = models.CharField(max_length=255)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    stock = models.PositiveIntegerField()
    
    def decrease_stock(self, quantity: int):
        if quantity > self.stock:
            raise ValueError("Not enough stock")
        self.stock -= quantity
        self.save()

class Order(models.Model):
    """Заказ с полным жизненным циклом"""
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    status = models.CharField(
        max_length=20,
        choices=[('pending', 'Pending'), ('paid', 'Paid'), ('shipped', 'Shipped')]
    )
    created_at = models.DateTimeField(auto_now_add=True)
    
    def apply_coupon(self, coupon_code: str):
        """Применить купон к заказу"""
        coupon = Coupon.objects.get(code=coupon_code)
        if coupon.is_valid():
            self.discount = coupon.discount
            self.save()
    
    def process_payment(self, payment_method):
        """Обработать платёж с использованием Strategy паттерна"""
        processor = PaymentProcessorFactory.create(payment_method)
        result = processor.process(self)
        
        if result.success:
            self.status = 'paid'
            self.save()
        return result

class OrderItem(models.Model):
    """Товар в заказе"""
    order = models.ForeignKey(Order, on_delete=models.CASCADE)
    product = models.ForeignKey(Product, on_delete=models.CASCADE)
    quantity = models.PositiveIntegerField()
    price_at_purchase = models.DecimalField(max_digits=10, decimal_places=2)

2. Payment Processing System (Strategy + Factory)

from abc import ABC, abstractmethod
from enum import Enum
from decimal import Decimal
from dataclasses import dataclass
from typing import Optional

@dataclass
class PaymentResult:
    success: bool
    transaction_id: str
    message: str
    error: Optional[str] = None

class PaymentMethod(ABC):
    """Абстрактный базовый класс для методов платежа"""
    
    @abstractmethod
    def validate(self, payment_data: dict) -> bool:
        """Валидировать данные платежа"""
        pass
    
    @abstractmethod
    def process(self, amount: Decimal, currency: str) -> PaymentResult:
        """Обработать платёж"""
        pass
    
    @abstractmethod
    def refund(self, transaction_id: str) -> PaymentResult:
        """Вернуть деньги"""
        pass

class CreditCardPayment(PaymentMethod):
    def __init__(self, card_number: str, cvv: str, expiry: str):
        self.card_number = card_number
        self.cvv = cvv
        self.expiry = expiry
    
    def validate(self, payment_data: dict) -> bool:
        # Используем Luhn алгоритм
        return self._luhn_check(self.card_number)
    
    def process(self, amount: Decimal, currency: str) -> PaymentResult:
        try:
            # Вызываем payment gateway (Stripe, PayPal и т.д.)
            response = stripe.Charge.create(
                amount=int(amount * 100),
                currency=currency,
                source=self.card_number
            )
            return PaymentResult(
                success=True,
                transaction_id=response['id'],
                message="Payment processed"
            )
        except Exception as e:
            return PaymentResult(
                success=False,
                transaction_id="",
                message="Payment failed",
                error=str(e)
            )
    
    def refund(self, transaction_id: str) -> PaymentResult:
        # Логика возврата
        pass

class PayPalPayment(PaymentMethod):
    def __init__(self, email: str):
        self.email = email
    
    def validate(self, payment_data: dict) -> bool:
        return '@' in self.email
    
    def process(self, amount: Decimal, currency: str) -> PaymentResult:
        # PayPal API
        pass
    
    def refund(self, transaction_id: str) -> PaymentResult:
        pass

class PaymentProcessorFactory:
    """Factory для создания payment processors"""
    
    _processors = {
        'credit_card': CreditCardPayment,
        'paypal': PayPalPayment,
        'apple_pay': ApplePayPayment,
    }
    
    @classmethod
    def register(cls, method_name: str, processor_class):
        cls._processors[method_name] = processor_class
    
    @classmethod
    def create(cls, method: str, **kwargs) -> PaymentMethod:
        if method not in cls._processors:
            raise ValueError(f"Unknown payment method: {method}")
        return cls._processors[method](**kwargs)

# Использование
payment = PaymentProcessorFactory.create(
    'credit_card',
    card_number='4111111111111111',
    cvv='123',
    expiry='12/25'
)
result = payment.process(Decimal('99.99'), 'USD')

3. Task Queue System (Observer Pattern)

from typing import Callable, List
from enum import Enum
import asyncio

class TaskStatus(Enum):
    PENDING = 'pending'
    RUNNING = 'running'
    COMPLETED = 'completed'
    FAILED = 'failed'

class Task:
    """Задача с наблюдателями для отслеживания статуса"""
    
    def __init__(self, task_id: str, func: Callable, args=None):
        self.task_id = task_id
        self.func = func
        self.args = args or ()
        self.status = TaskStatus.PENDING
        self._observers: List[Callable] = []
        self.result = None
        self.error = None
    
    def attach_observer(self, observer: Callable):
        """Добавить наблюдателя"""
        self._observers.append(observer)
    
    def detach_observer(self, observer: Callable):
        """Удалить наблюдателя"""
        self._observers.remove(observer)
    
    def _notify_observers(self):
        """Уведомить всех наблюдателей"""
        for observer in self._observers:
            observer(self)
    
    async def execute(self):
        """Выполнить задачу и уведомить наблюдателей"""
        try:
            self.status = TaskStatus.RUNNING
            self._notify_observers()
            
            self.result = await self.func(*self.args)
            self.status = TaskStatus.COMPLETED
        except Exception as e:
            self.error = str(e)
            self.status = TaskStatus.FAILED
        finally:
            self._notify_observers()

class TaskQueue:
    """Очередь задач"""
    
    def __init__(self, max_workers: int = 4):
        self.queue: asyncio.Queue = asyncio.Queue()
        self.max_workers = max_workers
    
    async def enqueue(self, task: Task):
        """Добавить задачу в очередь"""
        await self.queue.put(task)
    
    async def start_workers(self):
        """Запустить рабочих"""
        workers = [
            asyncio.create_task(self._worker())
            for _ in range(self.max_workers)
        ]
        await asyncio.gather(*workers)
    
    async def _worker(self):
        """Рабочий поток"""
        while True:
            task = await self.queue.get()
            await task.execute()
            self.queue.task_done()

# Использование
async def send_email(to: str, subject: str):
    print(f"Sending email to {to}: {subject}")
    await asyncio.sleep(2)

def on_task_status_changed(task: Task):
    print(f"Task {task.task_id} status: {task.status.value}")
    if task.status == TaskStatus.COMPLETED:
        print(f"Result: {task.result}")
    elif task.status == TaskStatus.FAILED:
        print(f"Error: {task.error}")

task = Task('email-1', send_email, ('user@example.com', 'Hello'))
task.attach_observer(on_task_status_changed)
await task_queue.enqueue(task)

4. User Management System (Inheritance)

from datetime import datetime, timedelta
from typing import List

class User:
    """Базовый класс пользователя"""
    
    def __init__(self, user_id: str, email: str, name: str):
        self.user_id = user_id
        self.email = email
        self.name = name
        self.created_at = datetime.now()
    
    def update_profile(self, **kwargs):
        for key, value in kwargs.items():
            if hasattr(self, key):
                setattr(self, key, value)
    
    def get_profile(self) -> dict:
        return {
            'user_id': self.user_id,
            'email': self.email,
            'name': self.name
        }

class PremiumUser(User):
    """Premium пользователь с дополнительными возможностями"""
    
    def __init__(self, user_id: str, email: str, name: str, subscription_date: datetime = None):
        super().__init__(user_id, email, name)
        self.subscription_date = subscription_date or datetime.now()
        self.subscription_level = 'premium'
    
    def is_subscription_active(self) -> bool:
        """Проверить, активна ли подписка"""
        # Подписка действительна 1 год
        expiry = self.subscription_date + timedelta(days=365)
        return datetime.now() < expiry
    
    def get_benefits(self) -> List[str]:
        """Получить привилегии premium пользователя"""
        return [
            'Unlimited storage',
            'Priority support',
            'Advanced analytics',
            'Custom branding'
        ]

class AdminUser(User):
    """Администратор с особыми правами"""
    
    def __init__(self, user_id: str, email: str, name: str, role: str = 'admin'):
        super().__init__(user_id, email, name)
        self.role = role
        self.permissions = self._get_default_permissions()
    
    def _get_default_permissions(self) -> List[str]:
        """Получить стандартные права админа"""
        return [
            'delete_user',
            'edit_user',
            'view_analytics',
            'manage_subscriptions'
        ]
    
    def grant_permission(self, permission: str):
        if permission not in self.permissions:
            self.permissions.append(permission)
    
    def revoke_permission(self, permission: str):
        if permission in self.permissions:
            self.permissions.remove(permission)
    
    def has_permission(self, permission: str) -> bool:
        return permission in self.permissions

# Использование
admin = AdminUser('admin-1', 'admin@example.com', 'John Admin')
print(admin.has_permission('delete_user'))  # True

premium = PremiumUser('user-1', 'user@example.com', 'Jane Premium')
print(premium.get_benefits())  # ['Unlimited storage', ...]

5. API Route Handling (Composition)

from typing import Dict, Callable
from fastapi import FastAPI, Request, Response
from abc import ABC, abstractmethod

class BaseHandler(ABC):
    """Базовый обработчик запроса"""
    
    @abstractmethod
    async def handle(self, request: Request) -> Response:
        pass

class UserHandler(BaseHandler):
    def __init__(self, db, logger):
        self.db = db
        self.logger = logger
    
    async def handle(self, request: Request) -> Response:
        user_id = request.path_params['user_id']
        user = self.db.get_user(user_id)
        self.logger.info(f"Fetched user: {user_id}")
        return Response(user.to_json())

class ProductHandler(BaseHandler):
    def __init__(self, db, cache, logger):
        self.db = db
        self.cache = cache
        self.logger = logger
    
    async def handle(self, request: Request) -> Response:
        product_id = request.path_params['product_id']
        
        # Проверяем кэш
        cached = self.cache.get(f"product:{product_id}")
        if cached:
            return Response(cached)
        
        # Получаем из БД
        product = self.db.get_product(product_id)
        self.cache.set(f"product:{product_id}", product.to_json())
        self.logger.info(f"Fetched product: {product_id}")
        return Response(product.to_json())

class Router:
    """Роутер запросов к обработчикам"""
    
    def __init__(self):
        self.handlers: Dict[str, BaseHandler] = {}
    
    def register(self, path: str, handler: BaseHandler):
        self.handlers[path] = handler
    
    async def route(self, request: Request) -> Response:
        path = request.url.path
        if path in self.handlers:
            return await self.handlers[path].handle(request)
        return Response(status_code=404)

# Использование
router = Router()
router.register('/users/{user_id}', UserHandler(db, logger))
router.register('/products/{product_id}', ProductHandler(db, cache, logger))

Что я изучил через ООП

1. Design Patterns:

   • Strategy (Payment processing)
   • Factory (Creating objects)
   • Observer (Task notifications)
   • Singleton (Database connection)
   • Decorator (Adding functionality)

2. SOLID Principles:

   • Single Responsibility — каждый класс имеет одну задачу
   • Open/Closed — открыт для расширения, закрыт для модификации
   • Liskov Substitution — дочерние классы заменяемы
   • Interface Segregation — узкие интерфейсы
   • Dependency Inversion — зависимости через interfaces

3. Advanced Concepts:

   • Inheritance и Composition
   • Polymorphism для гибких систем
   • Encapsulation для безопасности
   • Abstract Base Classes для контрактов

Где ООП помогает

• Масштабируемость — легко добавлять новые features
• Тестируемость — изолированные unit tests
• Поддерживаемость — четкая структура кода
• Переиспользуемость — компоненты работают в разных проектах
• Командная разработка — контракты между модулями

ООП — это не просто синтаксис Python, это способ мышления о архитектуре приложений.