Зачем нужна интеграция систем?

Интеграция систем: зачем нужна и как это работает

Это фундаментальный вопрос архитектуры. Интеграция — это не просто техническая необходимость, это стратегический элемент современного бизнеса.

Основная проблема: системы работают изолированно

Представь ситуацию без интеграции:

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  CRM Система    │     │ Billing System  │     │ Analytics       │
│ (Клиенты)      │     │ (Платежи)       │     │ (Статистика)    │
└─────────────────┘     └─────────────────┘     └─────────────────┘
  Данные не синхронизированы
  Много ручной работы
  Ошибки и дублирование

Примеры проблем:

• Клиент обновился в CRM, но Billing система не знает об этом
• Платёж прошёл, но аналитика не видит этого
• Разные версии данных в разных системах (несогласованность)
• Ручной ввод данных → ошибки и трата времени

Цель интеграции: один источник истины

После интеграции:

    ┌─────────────────────────────────────────────────────┐
    │         Unified Data Platform                       │
    │    (Источник истины - один для всех)               │
    └─────────────────────────────────────────────────────┘
                         ↓
    ┌────────────┬──────────────┬──────────────┐
    ↓            ↓              ↓              ↓
  CRM      Billing        Analytics       Email
 Система   Система        Система         Service

Все системы видят одни и те же данные!

Архитектура: асинхронная интеграция через events

Основной подход в современных системах — event-driven architecture.

from dataclasses import dataclass
from enum import Enum
from uuid import UUID
from datetime import datetime

class EventType(Enum):
    USER_CREATED = "user.created"
    USER_UPDATED = "user.updated"
    PAYMENT_PROCESSED = "payment.processed"
    ORDER_COMPLETED = "order.completed"

@dataclass
class DomainEvent:
    """Событие, которое произошло в системе"""
    event_id: UUID
    event_type: EventType
    aggregate_id: UUID  # ID объекта, который изменился
    data: dict
    timestamp: datetime
    version: int

# ПРИМЕР: Клиент зарегистрировался
event = DomainEvent(
    event_id=uuid4(),
    event_type=EventType.USER_CREATED,
    aggregate_id=user_id,
    data={
        'email': 'user@example.com',
        'name': 'John Doe'
    },
    timestamp=datetime.now(UTC),
    version=1
)

# Это событие попадает в:
# 1. Event Store (БД) — история всех событий
# 2. Message Queue (RabbitMQ/Kafka) — для других систем

Практический пример: e-commerce система

import asyncio
from abc import ABC, abstractmethod

# ===== ОСНОВНАЯ СИСТЕМА: Orders =====
class OrderService:
    def __init__(self, event_bus: EventBus, order_repo):
        self.event_bus = event_bus
        self.order_repo = order_repo
    
    async def create_order(self, user_id: UUID, items: list) -> Order:
        order = Order(id=uuid4(), user_id=user_id, items=items)
        await self.order_repo.save(order)
        
        # Публикуем событие
        event = OrderCreatedEvent(
            order_id=order.id,
            user_id=user_id,
            total_amount=order.calculate_total()
        )
        await self.event_bus.publish(event)
        
        return order

# ===== ИНТЕГРАЦИЯ 1: Billing система =====
class BillingEventHandler:
    """Слушает события Order системы и обрабатывает биллинг"""
    
    def __init__(self, billing_service):
        self.billing_service = billing_service
    
    async def on_order_created(self, event: OrderCreatedEvent):
        """Когда заказ создан, создаём счёт для оплаты"""
        invoice = await self.billing_service.create_invoice(
            order_id=event.order_id,
            amount=event.total_amount,
            user_id=event.user_id
        )
        print(f"✓ Invoice {invoice.id} created for order {event.order_id}")

# ===== ИНТЕГРАЦИЯ 2: Analytics система =====
class AnalyticsEventHandler:
    """Слушает события и собирает аналитику"""
    
    def __init__(self, analytics_service):
        self.analytics_service = analytics_service
    
    async def on_order_created(self, event: OrderCreatedEvent):
        """Отслеживаем метрику: новый заказ"""
        await self.analytics_service.increment_counter(
            metric="orders_created",
            tags={"user_id": event.user_id}
        )
        await self.analytics_service.record_revenue(
            amount=event.total_amount
        )

# ===== ИНТЕГРАЦИЯ 3: Email сервис =====
class EmailEventHandler:
    """Слушает события и отправляет уведомления"""
    
    def __init__(self, email_service):
        self.email_service = email_service
    
    async def on_order_created(self, event: OrderCreatedEvent):
        """Отправляем email уведомление"""
        user = await get_user(event.user_id)
        await self.email_service.send(
            to=user.email,
            subject=f"Order #{event.order_id} confirmed",
            template="order_confirmation",
            data={"order_id": event.order_id, "amount": event.total_amount}
        )

# ===== EVENT BUS (Message Queue) =====
class EventBus:
    """Центральное место, где встречаются события"""
    
    def __init__(self, queue: RabbitMQ):
        self.queue = queue
        self.handlers = {}
    
    def subscribe(self, event_type: str, handler: callable):
        """Регистрируем обработчик для события"""
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)
    
    async def publish(self, event: DomainEvent):
        """Публикуем событие всем подписчикам"""
        await self.queue.publish(event.event_type.value, event.dict())
    
    async def handle_event(self, event_type: str, event_data: dict):
        """Вызываем все обработчики события"""
        if event_type in self.handlers:
            tasks = [
                handler(event_data) 
                for handler in self.handlers[event_type]
            ]
            await asyncio.gather(*tasks)

# ===== ИНИЦИАЛИЗАЦИЯ =====
async def main():
    # Создаём event bus
    event_bus = EventBus(queue=RabbitMQ())
    
    # Регистрируем обработчики из разных систем
    billing_handler = BillingEventHandler(billing_service)
    analytics_handler = AnalyticsEventHandler(analytics_service)
    email_handler = EmailEventHandler(email_service)
    
    event_bus.subscribe('order.created', billing_handler.on_order_created)
    event_bus.subscribe('order.created', analytics_handler.on_order_created)
    event_bus.subscribe('order.created', email_handler.on_order_created)
    
    # Когда создаём заказ:
    order_service = OrderService(event_bus, order_repo)
    order = await order_service.create_order(user_id, items)
    
    # Автоматически:
    # 1. Создан счёт в Billing системе
    # 2. Собрана статистика в Analytics
    # 3. Отправлено письмо пользователю
    # Всё произошло АСИНХРОННО, параллельно!

Различные типы интеграции

1. Синхронная интеграция (REST API)

# Order система вызывает Billing синхронно
class OrderServiceSync:
    def __init__(self, billing_client: BillingAPI):
        self.billing_client = billing_client
    
    async def create_order(self, user_id: UUID, items: list):
        order = Order(id=uuid4(), user_id=user_id, items=items)
        
        # Ждём ответа от Billing
        invoice = await self.billing_client.post(
            "/invoices",
            data={"order_id": order.id, "amount": order.total}
        )
        
        # Если Billing упал — заказ не создан!
        await self.order_repo.save(order)
        return order

Проблемы:

• Если Billing медленный → Orders медленный
• Если Billing упал → Orders упал
• Tight coupling (сильная связанность)

2. Асинхронная интеграция (Message Queue)

# Order система публикует событие, Billing слушает
class OrderServiceAsync:
    def __init__(self, event_bus: EventBus, order_repo):
        self.event_bus = event_bus
        self.order_repo = order_repo
    
    async def create_order(self, user_id: UUID, items: list):
        order = Order(id=uuid4(), user_id=user_id, items=items)
        await self.order_repo.save(order)
        
        # Публикуем событие
        await self.event_bus.publish(OrderCreatedEvent(order.id))
        
        # Возвращаемся сразу, не ждя Billing!
        return order

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

• Быстро (не ждём ответ)
• Loose coupling (слабая связанность)
• Если Billing упал, ордер всё равно создан
• Scalable (можно добавить больше обработчиков)

Реальные примеры интеграции

Пример 1: Платёжный шлюз Stripe

class PaymentService:
    async def process_payment(self, order_id: UUID, amount: Decimal):
        # Вызываем Stripe API
        charge = await stripe.Charge.create(
            amount=int(amount * 100),
            currency='usd',
            source='tok_visa'
        )
        
        # Публикуем событие
        if charge.status == 'succeeded':
            await self.event_bus.publish(
                PaymentProcessedEvent(
                    order_id=order_id,
                    charge_id=charge.id,
                    amount=amount
                )
            )
        
        return charge

# Теперь Orders, Billing, Analytics получат это событие

Пример 2: Интеграция с CRM (Salesforce)

class CRMIntegration:
    """Синхронизирует клиентов между нашей системой и Salesforce"""
    
    async def on_user_created(self, event: UserCreatedEvent):
        # Создаём контакт в Salesforce
        contact = await self.salesforce_client.contacts.create({
            'FirstName': event.user.first_name,
            'LastName': event.user.last_name,
            'Email': event.user.email,
            'ExternalId': event.user.id  # Связь с нашей системой
        })
        
        # Сохраняем mapping
        await self.mappings.save(
            our_id=event.user.id,
            salesforce_id=contact.id
        )

Когда интеграция критична

1. Когда много систем работают с одними данными — нужна синхронизация
2. Когда есть внешние зависимости (Stripe, Twilio, Slack) — нужна интеграция
3. Когда системы развиваются независимо — нужна слабая связанность через события
4. Когда нужна масштабируемость — асинхронная интеграция через очереди

Итоги

Интеграция систем нужна для:

1. Синхронизации данных — один источник истины
2. Автоматизации бизнес-процессов — меньше ручной работы
3. Масштабируемости — слабая связанность между компонентами
4. Надёжности — асинхронная обработка через очереди
5. Агилити — можно менять системы независимо друг от друга

Современный подход: event-driven архитектура через Message Queue (RabbitMQ/Kafka). Это позволяет системам быть независимыми, масштабируемыми и надёжными.