Как использовал полиморфизм в своих проектах?

Полиморфизм в моих проектах

Использую полиморфизм практически во всех больших проектах. Вот конкретные примеры.

1. Payment Gateway система (Fintech)

Когда я работал в финтехе, нужно было поддерживать несколько платёжных систем: Stripe, PayPal, YandexKassa.

Подход с полиморфизмом:

from abc import ABC, abstractmethod
from typing import Dict

class PaymentGateway(ABC):
    @abstractmethod
    def charge(self, amount: int, customer_id: str) -> Dict:
        """Снять деньги со счёта"""
        pass
    
    @abstractmethod
    def refund(self, transaction_id: str) -> Dict:
        """Вернуть деньги"""
        pass

class StripeGateway(PaymentGateway):
    def charge(self, amount: int, customer_id: str) -> Dict:
        # Stripe API call
        import stripe
        stripe.api_key = self.api_key
        charge = stripe.Charge.create(
            amount=amount,
            currency="usd",
            customer=customer_id
        )
        return {"transaction_id": charge.id, "status": "success"}
    
    def refund(self, transaction_id: str) -> Dict:
        import stripe
        refund = stripe.Refund.create(charge=transaction_id)
        return {"status": "refunded"}

class PayPalGateway(PaymentGateway):
    def charge(self, amount: int, customer_id: str) -> Dict:
        # PayPal API call
        result = self.paypal_client.charge(
            customer_id=customer_id,
            amount=amount
        )
        return {"transaction_id": result.id, "status": "success"}
    
    def refund(self, transaction_id: str) -> Dict:
        result = self.paypal_client.refund(transaction_id)
        return {"status": "refunded"}

class PaymentService:
    def __init__(self, gateway: PaymentGateway):
        self.gateway = gateway  # Любой gateway!
    
    def process_subscription(self, customer_id: str, amount: int):
        result = self.gateway.charge(amount, customer_id)
        if result["status"] == "success":
            # Логирование, создание подписки в БД
            return {"subscription_id": "sub_123"}
        raise Exception("Payment failed")

Преимущества:

• Переключение между Stripe и PayPal в конфиге
• Тестировать легко (mock gateway)
• Добавить новый gateway — один класс

2. Report Generator система (Data Analytics)

Система генерирования отчётов в разных форматах: PDF, Excel, JSON.

from abc import ABC, abstractmethod
from io import BytesIO

class ReportFormatter(ABC):
    @abstractmethod
    def format(self, data: list) -> bytes:
        pass
    
    @abstractmethod
    def content_type(self) -> str:
        pass

class PDFFormatter(ReportFormatter):
    def format(self, data: list) -> bytes:
        from reportlab.lib.pagesizes import letter
        from reportlab.pdfgen import canvas
        
        buffer = BytesIO()
        pdf = canvas.Canvas(buffer, pagesize=letter)
        
        y = 750
        for row in data:
            pdf.drawString(100, y, str(row))
            y -= 20
        
        pdf.save()
        return buffer.getvalue()
    
    def content_type(self) -> str:
        return "application/pdf"

class ExcelFormatter(ReportFormatter):
    def format(self, data: list) -> bytes:
        import openpyxl
        
        wb = openpyxl.Workbook()
        ws = wb.active
        
        for idx, row in enumerate(data, 1):
            ws.append(row)
        
        buffer = BytesIO()
        wb.save(buffer)
        return buffer.getvalue()
    
    def content_type(self) -> str:
        return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"

class ReportService:
    def __init__(self, formatter: ReportFormatter):
        self.formatter = formatter
    
    def generate(self, data: list):
        return {
            "content": self.formatter.format(data),
            "type": self.formatter.content_type()
        }

# Использование:
data = [{"name": "John", "sales": 1000}, {"name": "Jane", "sales": 1500}]

pdf_service = ReportService(PDFFormatter())
excel_service = ReportService(ExcelFormatter())

pdf_report = pdf_service.generate(data)
excel_report = excel_service.generate(data)

3. Database Adapter (Миграция с MongoDB на PostgreSQL)

Когда мигрировали с MongoDB на PostgreSQL, использовали полиморфизм для параллельной работы:

class DatabaseAdapter(ABC):
    @abstractmethod
    def find_one(self, collection: str, query: dict):
        pass
    
    @abstractmethod
    def insert_one(self, collection: str, document: dict):
        pass

class MongoDBAdapter(DatabaseAdapter):
    def __init__(self, client):
        self.client = client
    
    def find_one(self, collection: str, query: dict):
        db = self.client["myapp"]
        return db[collection].find_one(query)
    
    def insert_one(self, collection: str, document: dict):
        db = self.client["myapp"]
        return db[collection].insert_one(document)

class PostgreSQLAdapter(DatabaseAdapter):
    def __init__(self, session):
        self.session = session
    
    def find_one(self, collection: str, query: dict):
        # Преобразуем в SQL
        model = self.get_model(collection)
        return self.session.query(model).filter_by(**query).first()
    
    def insert_one(self, collection: str, document: dict):
        model = self.get_model(collection)
        obj = model(**document)
        self.session.add(obj)
        self.session.commit()
        return obj

class UserRepository:
    def __init__(self, adapter: DatabaseAdapter):
        self.adapter = adapter
    
    def get_user(self, user_id: int):
        return self.adapter.find_one("users", {"id": user_id})

4. Strategy Pattern в ML Pipeline

Для обработки данных разными алгоритмами:

from abc import ABC, abstractmethod

class DataProcessor(ABC):
    @abstractmethod
    def process(self, data: list) -> list:
        pass

class MinMaxNormalizer(DataProcessor):
    def process(self, data: list) -> list:
        min_val = min(data)
        max_val = max(data)
        return [(x - min_val) / (max_val - min_val) for x in data]

class StandardScaler(DataProcessor):
    def process(self, data: list) -> list:
        import statistics
        mean = statistics.mean(data)
        stdev = statistics.stdev(data)
        return [(x - mean) / stdev for x in data]

class MLPipeline:
    def __init__(self, preprocessor: DataProcessor, model):
        self.preprocessor = preprocessor
        self.model = model
    
    def train(self, raw_data: list, labels: list):
        processed = self.preprocessor.process(raw_data)
        self.model.fit(processed, labels)

Ключевые преимущества

1. DRY — код не дублируется
2. Open/Closed — открыто расширение, закрыто модификация
3. Тестируемость — легко создавать мокі
4. Гибкость — менять реализацию в конфиге
5. Масштабируемость — новые варианты без рефакторинга

Архитектура становится более понятной, когда каждый класс отвечает за одно.