Subtask 2-14: Integration Steps and Best Practices - Complete Guide

Date: 2025-02-10

Overview

Это руководство содержит пошаговые инструкции по интеграции с API Lamoda, лучшие практики, рекомендации по тестированию и развертыванию в продакшене.

Подготовка к интеграции
Пошаговый процесс интеграции
Выбор API системы
Аутентификация и управление токенами
Архитектура интеграции
Лучшие практики разработки
Обработка ошибок и повторные попытки
Работа с Rate Limits
Тестирование интеграции
Развертывание в продакшен
Мониторинг и оповещения
Безопасность
Производительность и оптимизация
Типовые сценарии интеграции
Частые проблемы и решения
Полезные ресурсы

1. Подготовка к интеграции

1.1. Необходимые условия

Перед началом интеграции убедитесь, что у вас есть:

✅ Договор с Lamoda: Подписанный договор на продажу товаров
✅ Доступ к продавцу: Аккаунт в Lamoda Seller Academy (academy.lamoda.ru)
✅ API учетные данные:
- client_id - идентификатор клиента
- client_secret - секретный ключ
- Выдаются менеджером Lamoda по запросу
✅ Техническая спецификация: Понимание бизнес-модели (FBS, FBO, DBS)
✅ CRM/ERP система: Система для интеграции (1С, SAP, Magento, custom solution)

1.2. Выбор бизнес-модели

Определите, с какой моделью работы будете интегрироваться:

Модель	Описание	Используемые API
FBS	Fulfillment by Seller - хранение и сборка у продавца	Seller API + B2B API + REST API
FBO	Fulfillment by Operator - хранение на складе Lamoda	Seller API + B2B API
DBS	Delivery by Seller - доставка курьерами продавца	Seller API + B2B API
B2B FF	B2B Fulfillment - корпоративные заказы	B2B API
B2B FBS	B2B Fulfillment by Seller - корпоративные FBS	B2B API

Совет: Начните с одной модели, затем масштабируйте на другие.

1.3. Понимание архитектуры API

Lamoda предоставляет ТРИ API системы:

┌─────────────────────────────────────────────────────────────┐
│                    LAMODA API ECOSYSTEM                     │
├─────────────────────────────────────────────────────────────┤
│                                                               │
│  ┌─────────────────────┐  ┌──────────────────┐  ┌─────────┐ │
│  │   B2B Platform API  │  │ Seller JSON-RPC  │  │  Seller │ │
│  │       (REST)        │  │      API         │  │ REST API│ │
│  ├─────────────────────┤  ├──────────────────┤  ├─────────┤ │
│  │ • Orders            │  │ • Products       │  │ • FBS   │ │
│  │ • Shipments         │  │ • Prices         │  │   Return│ │
│  │ • Labels            │  │ • Stock          │  │   Boxes │ │
│  │ • Notifications     │  │ • Attributes     │  │ • Items │ │
│  │ • Delivery          │  │ • Images         │  │ •Feed-  │ │
│  │ • Addresses         │  │ • Dictionaries   │  │   back  │ │
│  │ • Webhooks          │  │ • Questions      │  │         │ │
│  └─────────────────────┘  └──────────────────┘  └─────────┘ │
│         ↓                           ↓                  ↓      │
│    OAuth2 (24h)               OAuth2 (15min)      Bearer (15min)│
│                                                               │
└─────────────────────────────────────────────────────────────┘

1.4. Планирование интеграции

Составьте план интеграции:

Список функций:
- Создание/обновление товаров
- Управление ценами
- Синхронизация остатков
- Получение заказов
- Обновление статусов заказов
- Печать этикеток
- Обработка возвратов
- Работа с вопросами покупателей
Объем данных:
- Количество товаров: _____
- Количество заказов в день: _____
- Количество обновлений цен в день: _____
- Количество обновлений остатков в день: _____
Частота синхронизации:
- Товары: раз в день / раз в неделю
- Цены: раз в час / раз в день
- Остатки: каждые 15 минут / раз в час
- Заказы: каждые 5 минут / каждые 15 минут

2. Пошаговый процесс интеграции

Шаг 1: Получение учетных данных API

Действие: Свяжитесь с менеджером Lamoda

Тема: Запрос доступа к API для продавца [Название компании]

Добрый день!

Прошу предоставить доступ к API Lamoda для интеграции нашей системы.

Компания: [Название]
ИНН: [Ваш ИНН]
Контакт: [ФИО, email, телефон]
Планируемая бизнес-модель: FBS / FBO / DBS
Необходимые функции:
- Создание и обновление товаров
- Синхронизация цен и остатков
- Получение заказов
- Обновление статусов

С уважением,
[Ваше имя]

Получите от менеджера:

client_id
client_secret
URL для тестовой среды (если доступна)

Сохраните в безопасном месте:

# environment variables
export LAMODA_CLIENT_ID="your_client_id"
export LAMODA_CLIENT_SECRET="your_client_secret"
export LAMODA_B2B_API_URL="https://api-b2b.lamoda.ru"
export LAMODA_SELLER_API_URL="https://public-api-seller.lamoda.ru"

Шаг 2: Настройка тестового окружения

Рекомендуемая структура проекта:

lamoda-integration/
├── config/
│   ├── __init__.py
│   ├── settings.py           # Конфигурация API
│   └── constants.py          # Константы бизнес-логики
├── services/
│   ├── __init__.py
│   ├── auth_service.py       # Управление токенами
│   ├── product_service.py    # Работа с товарами
│   ├── order_service.py      # Работа с заказами
│   └── stock_service.py      # Работа с остатками
├── models/
│   ├── __init__.py
│   ├── product.py            # Модели данных товаров
│   ├── order.py              # Модели данных заказов
│   └── api.py                # API модели (запросы/ответы)
├── utils/
│   ├── __init__.py
│   ├── http_client.py        # HTTP клиент с retry
│   ├── logger.py             # Логирование
│   └── validators.py         # Валидация данных
├── tests/
│   ├── __init__.py
│   ├── test_auth.py
│   ├── test_products.py
│   └── test_orders.py
├── .env                      # Секреты (не коммитить!)
├── .env.example              # Пример конфигурации
├── requirements.txt
└── README.md

Создайте .env файл:

# API Credentials
LAMODA_CLIENT_ID=your_client_id
LAMODA_CLIENT_SECRET=your_client_secret

# API URLs
LAMODA_B2B_API_URL=https://api-b2b.lamoda.ru
LAMODA_B2B_DEMO_API_URL=https://api-demo-b2b.lamoda.ru
LAMODA_SELLER_API_URL=https://public-api-seller.lamoda.ru

# Integration Settings
TOKEN_REFRESH_BUFFER=300      # Refresh token 5min before expiry
DEFAULT_TIMEOUT=30            # Request timeout in seconds
MAX_RETRIES=3                 # Max retry attempts
RETRY_DELAY=1                 # Initial retry delay in seconds

# Business Logic
BUSINESS_MODEL=FBS            # FBS, FBO, DBS, etc.
DEFAULT_WAREHOUSE=            # Склад по умолчанию

Шаг 3: Реализация аутентификации

Шаблон кода для управления токенами:

import os
import time
import requests
from datetime import datetime, timedelta
from typing import Optional
import logging

logger = logging.getLogger(__name__)

class TokenManager:
    """Управление OAuth2 токенами Lamoda API"""

    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.b2b_token: Optional[dict] = None
        self.seller_token: Optional[dict] = None
        self.token_buffer = int(os.getenv('TOKEN_REFRESH_BUFFER', 300))  # 5 min

    def get_b2b_token(self) -> str:
        """Получить актуальный токен B2B Platform API"""
        if self._should_refresh_token(self.b2b_token):
            self._refresh_b2b_token()
        return self.b2b_token['access_token']

    def get_seller_token(self) -> str:
        """Получить актуальный токен Seller API"""
        if self._should_refresh_token(self.seller_token):
            self._refresh_seller_token()
        return self.seller_token['access_token']

    def _should_refresh_token(self, token_data: Optional[dict]) -> bool:
        """Проверить, нужно ли обновить токен"""
        if not token_data:
            return True

        expires_at = token_data.get('expires_at')
        if not expires_at:
            return True

        # Обновляем за N секунд до истечения
        return time.time() > (expires_at - self.token_buffer)

    def _refresh_b2b_token(self):
        """Обновить токен B2B Platform API"""
        url = f"{os.getenv('LAMODA_B2B_API_URL')}/auth/token"
        params = {
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret
        }

        response = requests.get(url, params=params, timeout=30)
        response.raise_for_status()

        token_data = response.json()
        # B2B токен действует 24 часа
        self.b2b_token = {
            'access_token': token_data['access_token'],
            'expires_at': time.time() + (24 * 60 * 60) - 60,  # 24h - 1min
            'received_at': time.time()
        }

        logger.info(f"B2B Token refreshed, expires at {datetime.fromtimestamp(self.b2b_token['expires_at'])}")

    def _refresh_seller_token(self):
        """Обновить токен Seller API (JSON-RPC)"""
        url = f"{os.getenv('LAMODA_SELLER_API_URL')}/jsonrpc"

        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "v1.tokens.create",
            "params": {
                "clientId": self.client_id,
                "clientSecret": self.client_secret
            }
        }

        response = requests.post(url, json=payload, timeout=30)
        response.raise_for_status()

        result = response.json().get('result', {})
        # Seller токен действует 15 минут
        self.seller_token = {
            'access_token': result['token'],
            'expires_at': time.time() + (15 * 60) - 60,  # 15min - 1min
            'received_at': time.time()
        }

        logger.info(f"Seller Token refreshed, expires at {datetime.fromtimestamp(self.seller_token['expires_at'])}")

    def revoke_tokens(self):
        """Отозвать все токены"""
        self.b2b_token = None
        self.seller_token = None
        logger.info("All tokens revoked")

Шаг 4: Создание базового HTTP клиента

import requests
from typing import Optional, Dict, Any
import time
import logging
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

logger = logging.getLogger(__name__)

class LamodaAPIClient:
    """Базовый HTTP клиент для Lamoda API с retry логикой"""

    def __init__(self, token_manager: TokenManager):
        self.token_manager = token_manager
        self.timeout = int(os.getenv('DEFAULT_TIMEOUT', 30))
        self.session = requests.Session()

    def _get_headers(self, api_type: str) -> Dict[str, str]:
        """Получить заголовки с токеном"""
        if api_type == 'b2b':
            token = self.token_manager.get_b2b_token()
        elif api_type == 'seller':
            token = self.token_manager.get_seller_token()
        else:
            raise ValueError(f"Unknown API type: {api_type}")

        return {
            'Authorization': f'Bearer {token}',
            'Content-Type': 'application/json'
        }

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry=retry_if_exception_type((requests.exceptions.RequestException,))
    )
    def request(
        self,
        method: str,
        url: str,
        api_type: str = 'b2b',
        **kwargs
    ) -> Dict[str, Any]:
        """Выполнить запрос с retry логикой"""

        headers = self._get_headers(api_type)
        kwargs.setdefault('headers', {})
        kwargs['headers'].update(headers)
        kwargs.setdefault('timeout', self.timeout)

        try:
            response = self.session.request(method, url, **kwargs)
            response.raise_for_status()
            return response.json()

        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                logger.error(f"Authentication failed for {url}")
                # Попытка обновить токен и повторить
                self.token_manager.revoke_tokens()
                raise

            elif e.response.status_code == 429:
                retry_after = e.response.headers.get('Retry-After', 60)
                logger.warning(f"Rate limited, waiting {retry_after}s")
                time.sleep(int(retry_after))
                # Повтор будет выполнен декоратором @retry
                raise

            elif e.response.status_code >= 500:
                logger.error(f"Server error: {e.response.status_code}")
                raise

            else:
                logger.error(f"Request failed: {e.response.status_code} - {e.response.text}")
                raise

        except requests.exceptions.RequestException as e:
            logger.error(f"Network error: {str(e)}")
            raise

Шаг 5: Реализация базовых сервисов

Пример сервиса товаров:

class ProductService:
    """Сервис для работы с товарами"""

    def __init__(self, api_client: LamodaAPIClient):
        self.api_client = api_client
        self.base_url = os.getenv('LAMODA_SELLER_API_URL')

    def get_nomenclatures(self, limit: int = 1000, offset: int = 0) -> Dict[str, Any]:
        """Получить список номенклатуры"""
        url = f"{self.base_url}/jsonrpc"

        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "v1.nomenclatures.list",
            "params": {
                "limit": limit,
                "offset": offset
            }
        }

        return self.api_client.request('POST', url, api_type='seller', json=payload)

    def create_product(self, product_data: Dict[str, Any]) -> Dict[str, Any]:
        """Создать новый товар"""
        url = f"{self.base_url}/jsonrpc"

        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "v1.nomenclatures.store",
            "params": product_data
        }

        return self.api_client.request('POST', url, api_type='seller', json=payload)

    def update_prices(self, prices: list) -> Dict[str, Any]:
        """Массовое обновление цен"""
        url = f"{self.base_url}/jsonrpc"

        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "v1.nomenclatures.set-prices",
            "params": {
                "prices": prices  # [{"sku": "...", "price": 100}, ...]
            }
        }

        return self.api_client.request('POST', url, api_type='seller', json=payload)

Шаг 6: Тестирование интеграции

Тестовый скрипт:

def test_integration():
    """Базовый тест интеграции"""

    # 1. Инициализация
    token_manager = TokenManager(
        client_id=os.getenv('LAMODA_CLIENT_ID'),
        client_secret=os.getenv('LAMODA_CLIENT_SECRET')
    )

    api_client = LamodaAPIClient(token_manager)
    product_service = ProductService(api_client)

    # 2. Тест аутентификации
    print("Testing authentication...")
    try:
        b2b_token = token_manager.get_b2b_token()
        seller_token = token_manager.get_seller_token()
        print(f"✓ B2B Token: {b2b_token[:20]}...")
        print(f"✓ Seller Token: {seller_token[:20]}...")
    except Exception as e:
        print(f"✗ Authentication failed: {e}")
        return

    # 3. Тест получения товаров
    print("\nTesting products API...")
    try:
        products = product_service.get_nomenclatures(limit=10)
        print(f"✓ Retrieved {len(products.get('result', {}).get('items', []))} products")
    except Exception as e:
        print(f"✗ Products API failed: {e}")

    # 4. Тест получения заказов
    print("\nTesting orders API...")
    try:
        order_service = OrderService(api_client)
        orders = order_service.get_orders(limit=10)
        print(f"✓ Retrieved {len(orders.get('orders', []))} orders")
    except Exception as e:
        print(f"✗ Orders API failed: {e}")

    print("\n✓ Integration test completed!")

if __name__ == '__main__':
    test_integration()

Шаг 7: Развертывание в продакшен

Проверочный список перед релизом:

3. Выбор API системы

Сравнительная таблица API систем

Характеристика	B2B Platform API	Seller JSON-RPC API	Seller REST API
Тип	REST	JSON-RPC 2.0	REST
Базовый URL	https://api-b2b.lamoda.ru	https://public-api-seller.lamoda.ru/jsonrpc	https://public-api-seller.lamoda.ru/api
Токен	OAuth2 (24h)	OAuth2 (15min)	Bearer (15min)
Эндпоинтов	51	24 метода	10
Главная задача	Заказы, отправления, логистика	Товары, цены, остатки	Возвраты FBS, отзывы
Сложность	Средняя	Низкая	Низкая
Для новичков	⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐

4. Аутентификация и управление токенами

Стратегия обновления токенов

# B2B Platform API Token
# - Время жизни: 24 часа
# - Рекомендуемое обновление: каждые 23 часа
# - Buffer: 1 час до истечения

# Seller API Token (JSON-RPC и REST)
# - Время жизни: 15 минут
# - Рекомендуемое обновление: каждые 12-13 минут
# - Buffer: 2-3 минуты до истечения

Лучшие практики управления токенами

Храните токены в памяти, не в БД
Обновляйте проактивно (до истечения срока)
Логируйте события обновления (для мониторинга)
Обрабатывайте 401 ошибки (повторная попытка после обновления)
Используйте buffer (5-10% от времени жизни токена)

Пример реализации с кешированием

from functools import lru_cache
import threading

class CachedTokenManager:
    """Менеджер токенов с кешированием и потокобезопасностью"""

    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self._lock = threading.Lock()

    @lru_cache(maxsize=1)
    def get_b2b_token_cached(self) -> str:
        """Получить токен с кешированием"""
        with self._lock:
            # Token refresh logic here
            return self._fetch_b2b_token()

    def invalidate_cache(self):
        """Сбросить кеш (например, при 401 ошибке)"""
        self.get_b2b_token_cached.cache_clear()

5. Архитектура интеграции

Структура данных

# Пример структуры для очереди задач
class IntegrationTask:
    task_id: str
    task_type: str  # 'product_update', 'price_update', 'order_sync', etc.
    payload: Dict[str, Any]
    priority: int  # 0-9, 9 = highest
    retry_count: int = 0
    max_retries: int = 3
    created_at: datetime
    scheduled_at: datetime

6. Лучшие практики разработки

6.1. Работа с товарами

Создание товаров:

# ✅ ПРАВИЛЬНО: Проверка перед созданием
def create_or_update_product(product_data: Dict):
    # Сначала проверим, существует ли товар
    existing = get_product_by_sku(product_data['sku'])
    if existing:
        return update_product(product_data['sku'], product_data)
    else:
        return create_product(product_data)

# ❌ НЕПРАВИЛЬНО: Всегда создаем новый
def bad_create_product(product_data: Dict):
    return create_product(product_data)  # Дубликаты!

Массовое обновление:

# ✅ ПРАВИЛЬНО: Пакетная обработка
def bulk_update_products(products: List[Dict], batch_size: int = 100):
    results = []
    for i in range(0, len(products), batch_size):
        batch = products[i:i + batch_size]
        result = update_products_batch(batch)
        results.extend(result)
        time.sleep(1)  # Пауза между пачками
    return results

# ❌ НЕПРАВИЛЬНО: По одному
def bad_bulk_update(products: List[Dict]):
    for product in products:
        update_product(product)  # Очень медленно!

6.2. Работа с заказами

Получение заказов:

# ✅ ПРАВИЛЬНО: Инкрементальная синхронизация
def sync_orders(since: datetime):
    params = {
        'createdAtFrom': since.isoformat(),
        'limit': 1000
    }
    return get_orders(params)

# ❌ НЕПРАВИЛЬНО: Всегда получаем все заказы
def bad_sync_orders():
    return get_orders()  # Millions of orders!

Обновление статусов:

# ✅ ПРАВИЛЬНО: Проверка перед обновлением
def update_order_status(order_id: str, new_status: str):
    order = get_order(order_id)
    if can_transition_to(order['status'], new_status):
        return set_order_status(order_id, new_status)
    else:
        logger.warning(f"Invalid transition: {order['status']} → {new_status}")
        return None

# ❌ НЕПРАВИЛЬНО: Обновляем без проверки
def bad_update_status(order_id: str, new_status: str):
    return set_order_status(order_id, new_status)

6.3. Управление ценами

# ✅ ПРАВИЛЬНО: Валидация цен
def update_product_prices(price_updates: List[Dict]):
    validated = []
    for update in price_updates:
        if validate_price(update['price']):
            validated.append(update)
        else:
            logger.warning(f"Invalid price for {update['sku']}: {update['price']}")
    return bulk_update_prices(validated)

def validate_price(price: float) -> bool:
    return price > 0 and price < 1000000

6.4. Работа с остатками

# ✅ ПРАВИЛЬНО: Дельта-обновления
def update_stock_delta(sku: str, quantity_delta: int):
    current = get_stock(sku)
    new_quantity = current['quantity'] + quantity_delta
    return set_stock(sku, new_quantity)

# ❌ НЕПРАВИЛЬНО: Полная перезапись
def bad_update_stock(sku: str, new_quantity: int):
    return set_stock(sku, new_quantity)  # Race conditions!

7. Обработка ошибок и повторные попытки

Иерархия исключений

class LamodaAPIError(Exception):
    """Базовый класс для всех ошибок Lamoda API"""
    pass

class AuthenticationError(LamodaAPIError):
    """Ошибки аутентификации (401)"""
    pass

class RateLimitError(LamodaAPIError):
    """Превышен rate limit (429)"""
    def __init__(self, message: str, retry_after: int = None):
        super().__init__(message)
        self.retry_after = retry_after

class ValidationError(LamodaAPIError):
    """Ошибки валидации данных (400)"""
    def __init__(self, message: str, fields: Dict[str, str] = None):
        super().__init__(message)
        self.fields = fields or {}

class NotFoundError(LamodaAPIError):
    """Ресурс не найден (404)"""
    pass

class ServerError(LamodaAPIError):
    """Внутренняя ошибка сервера (5xx)"""
    pass

Retry стратегии

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log
)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((RateLimitError, ServerError)),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def api_call_with_retry(func, *args, **kwargs):
    """Вызов API с автоматическим retry"""
    try:
        return func(*args, **kwargs)
    except RateLimitError as e:
        if e.retry_after:
            time.sleep(e.retry_after)
        raise
    except ServerError:
        # Серверная ошибка, пробуем снова
        raise

Обработка конкретных сценариев

Сценарий 1: Токен истек (401)

def handle_auth_error(func):
    """Декоратор для обработки 401 ошибок"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except AuthenticationError:
            # Обновляем токен и пробуем снова
            logger.info("Token expired, refreshing...")
            token_manager.revoke_tokens()
            return func(*args, **kwargs)
    return wrapper

Сценарий 2: Rate Limit (429)

def handle_rate_limit_error(response):
    """Обработка 429 ошибки"""
    retry_after = int(response.headers.get('Retry-After', 60))
    logger.warning(f"Rate limited, waiting {retry_after}s")
    time.sleep(retry_after)
    # Retry будет выполнен декоратором @retry

Сценарий 3: Ошибка валидации (400)

def handle_validation_error(response):
    """Обработка 400 ошибки"""
    error_data = response.json()
    errors = error_data.get('errors', [])

    for error in errors:
        field = error.get('field')
        message = error.get('message')
        logger.error(f"Validation error for {field}: {message}")

    raise ValidationError(
        "Data validation failed",
        fields={e['field']: e['message'] for e in errors}
    )

8. Работа с Rate Limits

Консервативная стратегия (начните с нее)

class RateLimiter:
    """Простой rate limiter"""

    def __init__(self, requests_per_second: float = 1.0):
        self.min_interval = 1.0 / requests_per_second
        self.last_request_time = 0

    def wait_if_needed(self):
        """Подождать, если необходимо"""
        current_time = time.time()
        time_since_last_request = current_time - self.last_request_time

        if time_since_last_request < self.min_interval:
            sleep_time = self.min_interval - time_since_last_request
            time.sleep(sleep_time)

        self.last_request_time = time.time()

# Использование
rate_limiter = RateLimiter(requests_per_second=1.0)

def make_api_request():
    rate_limiter.wait_if_needed()
    return api_client.request(...)

Адаптивный Rate Limiting

class AdaptiveRateLimiter:
    """Адаптивный rate limiter с реакцией на 429 ошибки"""

    def __init__(self, initial_rps: float = 1.0):
        self.current_rps = initial_rps
        self.min_rps = 0.1
        self.max_rps = 10.0
        self.last_request_time = 0

    def wait_if_needed(self):
        """Подождать перед запросом"""
        interval = 1.0 / self.current_rps
        current_time = time.time()

        time_since_last = current_time - self.last_request_time
        if time_since_last < interval:
            time.sleep(interval - time_since_last)

        self.last_request_time = time.time()

    def report_error(self):
        """Уменьшить скорость при ошибке"""
        self.current_rps = max(self.min_rps, self.current_rps * 0.5)
        logger.warning(f"Reduced rate to {self.current_rps} RPS")

    def report_success(self):
        """Постепенно увеличить скорость"""
        self.current_rps = min(self.max_rps, self.current_rps * 1.01)

Использование нескольких токенов для параллельности

class ParallelAPIClient:
    """API клиент с несколькими токенами для параллельных запросов"""

    def __init__(self, num_tokens: int = 3):
        self.token_managers = [
            TokenManager(client_id, client_secret)
            for _ in range(num_tokens)
        ]
        self.current_index = 0

    def get_next_client(self) -> LamodaAPIClient:
        """Получить следующий клиент по кругу"""
        self.current_index = (self.current_index + 1) % len(self.token_managers)
        return LamodaAPIClient(self.token_managers[self.current_index])

    def parallel_request(self, urls: List[str]) -> List[Dict]:
        """Параллельные запросы с разными токенами"""
        from concurrent.futures import ThreadPoolExecutor

        with ThreadPoolExecutor(max_workers=len(self.token_managers)) as executor:
            futures = []
            for url in urls:
                client = self.get_next_client()
                future = executor.submit(client.request, 'GET', url)
                futures.append(future)

            return [f.result() for f in futures]

9. Тестирование интеграции

Стратегия тестирования

Уровни тестирования:

Unit Tests: Тестирование отдельных функций
Integration Tests: Тестирование с mock API
End-to-End Tests: Тестирование с реальным API (демо-среда)
Load Tests: Тестирование производительности

Unit Tests (с pytest)

import pytest
from unittest.mock import Mock, patch
from services.product_service import ProductService

@pytest.fixture
def mock_api_client():
    return Mock()

@pytest.fixture
def product_service(mock_api_client):
    return ProductService(mock_api_client)

def test_get_nomenclatures(product_service, mock_api_client):
    """Test getting nomenclatures list"""
    mock_response = {
        'result': {
            'items': [{'sku': 'TEST001'}],
            'total': 1
        }
    }
    mock_api_client.request.return_value = mock_response

    result = product_service.get_nomenclatures(limit=10)

    assert result['result']['total'] == 1
    assert len(result['result']['items']) == 1
    mock_api_client.request.assert_called_once()

def test_create_product_validation(product_service):
    """Test product creation validation"""
    invalid_product = {
        # Missing required fields
        'name': 'Test Product'
    }

    with pytest.raises(ValidationError):
        product_service.create_product(invalid_product)

Integration Tests (с pytest и responses)

import pytest
import responses
from services.auth_service import TokenManager

@responses.activate
def test_token_refresh():
    """Test token refresh"""
    responses.add(
        responses.GET,
        'https://api-b2b.lamoda.ru/auth/token',
        json={'access_token': 'test_token', 'expires_in': 86400},
        status=200
    )

    token_manager = TokenManager('client_id', 'client_secret')
    token = token_manager.get_b2b_token()

    assert token == 'test_token'

Load Tests (с locust)

from locust import HttpUser, task, between

class LamodaAPIUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        """Login before starting tasks"""
        self.token = self.get_token()

    def get_token(self):
        response = self.client.post("/auth/token", json={
            'grant_type': 'client_credentials',
            'client_id': 'test_id',
            'client_secret': 'test_secret'
        })
        return response.json()['access_token']

    @task(3)
    def get_products(self):
        self.client.get(
            "/api/v1/nomenclatures",
            headers={'Authorization': f'Bearer {self.token}'}
        )

    @task(1)
    def get_orders(self):
        self.client.get(
            "/api/v1/orders",
            headers={'Authorization': f'Bearer {self.token}'}
        )

Запуск load теста:

locust -f locustfile.py --host=https://api-b2b.lamoda.ru

План тестирования

Тип теста	Цель	Частота
Unit Tests	Логика функций	Каждый коммит
Integration Tests	Взаимодействие компонентов	Каждый коммит
E2E Tests	Проверка реального API	Перед релизом
Load Tests	Производительность	Раз в неделю / перед релизом

10. Развертывание в продакшен

Контрольный checklist

Перед развертыванием:

Environment Variables

# .env.production
ENVIRONMENT=production
DEBUG=false

# API
LAMODA_CLIENT_ID=${LAMODA_CLIENT_ID}
LAMODA_CLIENT_SECRET=${LAMODA_CLIENT_SECRET}
LAMODA_B2B_API_URL=https://api-b2b.lamoda.ru
LAMODA_SELLER_API_URL=https://public-api-seller.lamoda.ru

# Rate Limiting
MAX_REQUESTS_PER_SECOND=2
MAX_RETRIES=3
REQUEST_TIMEOUT=30

# Monitoring
SENTRY_DSN=${SENTRY_DSN}
LOG_LEVEL=INFO

# Business
BUSINESS_MODEL=FBS
WAREHOUSE_CODE=WH001

Docker конфигурация

FROM python:3.11-slim

WORKDIR /app

# Установка зависимостей
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Код приложения
COPY . .

# Environment variables
ENV PYTHONUNBUFFERED=1
ENV ENVIRONMENT=production

# Запуск
CMD ["python", "-m", "services.main"]

# docker-compose.yml
version: '3.8'

services:
  integration-service:
    build: .
    environment:
      - LAMODA_CLIENT_ID=${LAMODA_CLIENT_ID}
      - LAMODA_CLIENT_SECRET=${LAMODA_CLIENT_SECRET}
    env_file:
      - .env.production
    restart: unless-stopped
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

  redis:
    image: redis:7-alpine
    restart: unless-stopped

  worker:
    build: .
    command: python -m services.worker
    environment:
      - LAMODA_CLIENT_ID=${LAMODA_CLIENT_ID}
      - LAMODA_CLIENT_SECRET=${LAMODA_CLIENT_SECRET}
    env_file:
      - .env.production
    restart: unless-stopped
    depends_on:
      - redis

План отката (Rollback)

#!/bin/bash
# rollback.sh

echo "Rolling back to previous version..."

# 1. Stop current service
docker-compose down

# 2. Checkout previous version
git checkout HEAD~1

# 3. Rebuild and restart
docker-compose up -d

# 4. Verify
sleep 10
docker-compose ps

echo "Rollback completed!"

11. Мониторинг и оповещения

Ключевые метрики

1. Метрики API:

Requests per second (RPS)
Response time (p50, p95, p99)
Error rate (4xx, 5xx)
Rate limit hits (429)
Token refresh count

2. Бизнес метрики:

Orders synchronized
Products updated
Stock updates processed
Returns processed

Реализация мониторинга

from prometheus_client import Counter, Histogram, Gauge
import time

# Метрики
api_requests_total = Counter(
    'lamoda_api_requests_total',
    'Total API requests',
    ['method', 'endpoint', 'status']
)

api_request_duration = Histogram(
    'lamoda_api_request_duration_seconds',
    'API request duration',
    ['method', 'endpoint']
)

api_errors_total = Counter(
    'lamoda_api_errors_total',
    'Total API errors',
    ['error_type']
)

active_tokens = Gauge(
    'lamoda_active_tokens',
    'Number of active API tokens'
)

# Декоратор для метрик
def monitor_api_call(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        status = 'success'

        try:
            result = func(*args, **kwargs)
            return result
        except Exception as e:
            status = 'error'
            api_errors_total.labels(error_type=type(e).__name__).inc()
            raise
        finally:
            duration = time.time() - start_time
            api_request_duration.labels(
                method=func.__name__,
                endpoint=kwargs.get('url', 'unknown')
            ).observe(duration)
            api_requests_total.labels(
                method=func.__name__,
                endpoint=kwargs.get('url', 'unknown'),
                status=status
            ).inc()

    return wrapper

Алерты (Prometheus AlertManager)

# alerts.yml
groups:
  - name: lamoda_api
    interval: 30s
    rules:
      - alert: HighErrorRate
        expr: rate(lamoda_api_errors_total[5m]) > 0.1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "High error rate detected"
          description: "Error rate is {{ $value }} errors/sec"

      - alert: RateLimitHit
        expr: rate(lamoda_api_requests_total{status="429"}[1m]) > 0
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "Rate limit hit"
          description: "We're hitting rate limits"

      - alert: SlowResponseTime
        expr: histogram_quantile(0.95, rate(lamoda_api_request_duration_seconds_bucket[5m])) > 5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Slow API response times"
          description: "p95 latency is {{ $value }} seconds"

Логирование

import logging
import sys
from datetime import datetime

# Настройка логирования
def setup_logging(log_level: str = 'INFO'):
    """Настройка структурированного логирования"""

    log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    logging.basicConfig(
        level=getattr(logging, log_level),
        format=log_format,
        handlers=[
            logging.StreamHandler(sys.stdout),
            logging.FileHandler(f'lamoda_integration_{datetime.now():%Y%m%d}.log')
        ]
    )

    # Отдельный логгер для API вызовов
    api_logger = logging.getLogger('lamoda.api')
    api_handler = logging.FileHandler('api_calls.log')
    api_handler.setFormatter(logging.Formatter(log_format))
    api_logger.addHandler(api_handler)

    return logging.getLogger(__name__)

# Использование
logger = setup_logging('INFO')

def log_api_request(method: str, url: str, params: dict, response):
    """Логирование API запроса"""
    logger.info(f"API Request: {method} {url}")
    logger.debug(f"Request params: {params}")
    logger.debug(f"Response status: {response.status_code}")
    logger.debug(f"Response body: {response.text[:500]}")  # First 500 chars

12. Безопасность

Защита учетных данных

# ✅ ПРАВИЛЬНО: Environment variables
import os
from dotenv import load_dotenv

load_dotenv()  # Загружаем из .env файла

client_id = os.getenv('LAMODA_CLIENT_ID')
client_secret = os.getenv('LAMODA_CLIENT_SECRET')

# ❌ НЕПРАВИЛЬНО: Хардкод в коде
client_id = "my_hardcoded_client_id"  # SECURITY RISK!

Хранение секретов

Варианты хранения:

Environment Variables (для простых случаев)
HashiCorp Vault (для продакшена)
AWS Secrets Manager (если на AWS)
Azure Key Vault (если на Azure)

Пример с Vault

import hvac

class VaultSecretManager:
    """Менеджер секретов с HashiCorp Vault"""

    def __init__(self, vault_url: str, token: str):
        self.client = hvac.Client(url=vault_url, token=token)

    def get_lamoda_credentials(self) -> tuple:
        """Получить учетные данные Lamoda из Vault"""
        response = self.client.secrets.kv.v2.read_secret_version(
            path='lamoda/prod'
        )

        credentials = response['data']['data']
        return (
            credentials['client_id'],
            credentials['client_secret']
        )

Безопасность в коде

Никогда не логируйте токены:

# ✅ ПРАВИЛЬНО
logger.info(f"Token received: {token[:10]}...")  # Только первые 10 символов

# ❌ НЕПРАВИЛЬНО
logger.info(f"Token: {token}")  # Логируем полный токен!

Валидация входных данных:

def validate_sku(sku: str) -> bool:
    """Валидация SKU"""
    if not sku or len(sku) > 100:
        return False
    if not all(c.isalnum() or c in '-_' for c in sku):
        return False
    return True

Используйте HTTPS только:

# ✅ ПРАВИЛЬНО
api_url = "https://api-b2b.lamoda.ru"

# ❌ НЕПРАВИЛЬНО
api_url = "http://api-b2b.lamoda.ru"  # Незащищенное соединение!

13. Производительность и оптимизация

Кеширование

from functools import lru_cache
import redis
import json

class CacheManager:
    """Менеджер кеширования с Redis"""

    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)

    def get(self, key: str) -> Any:
        """Получить из кеша"""
        value = self.redis.get(key)
        if value:
            return json.loads(value)
        return None

    def set(self, key: str, value: Any, ttl: int = 300):
        """Записать в кеш"""
        self.redis.setex(key, ttl, json.dumps(value))

    def invalidate(self, pattern: str):
        """Инвалидация по паттерну"""
        keys = self.redis.keys(pattern)
        if keys:
            self.redis.delete(*keys)

# Использование
cache = CacheManager(os.getenv('REDIS_URL'))

def get_categories():
    cache_key = 'lamoda:categories'

    cached = cache.get(cache_key)
    if cached:
        return cached

    # Если нет в кеше, загружаем из API
    categories = fetch_categories_from_api()
    cache.set(cache_key, categories, ttl=3600)  # 1 hour
    return categories

Оптимизация пакетных операций

def bulk_update_with_chunks(items: List[Dict], chunk_size: int = 100):
    """Пакетное обновление с чанками"""
    results = []

    for i in range(0, len(items), chunk_size):
        chunk = items[i:i + chunk_size]

        try:
            result = api_client.update_chunk(chunk)
            results.extend(result)
        except Exception as e:
            logger.error(f"Failed to update chunk {i}-{i+len(chunk)}: {e}")
            # Retry logic here
            continue

        # Небольшая пауза между чанками
        time.sleep(0.5)

    return results

Асинхронная обработка

import asyncio
import aiohttp

class AsyncLamodaClient:
    """Асинхронный API клиент"""

    async def fetch_orders_async(self, order_ids: List[str]):
        """Параллельное получение заказов"""
        async with aiohttp.ClientSession() as session:
            tasks = [
                self.fetch_order(session, order_id)
                for order_id in order_ids
            ]
            return await asyncio.gather(*tasks)

    async def fetch_order(self, session: aiohttp.ClientSession, order_id: str):
        """Получить один заказ"""
        url = f"{self.base_url}/api/v1/orders/{order_id}"
        headers = {"Authorization": f"Bearer {self.token}"}

        async with session.get(url, headers=headers) as response:
            return await response.json()

14. Типовые сценарии интеграции

Сценарий 1: Первичная загрузка каталога

Задача: Загрузить 10,000 товаров в Lamoda

def initial_catalog_upload(products_file: str):
    """Первоначальная загрузка каталога"""

    # 1. Загружаем товары из файла
    products = load_products_from_csv(products_file)

    # 2. Получаем справочники
    brands = get_brands()
    categories = get_categories()
    attributes = get_attributes()

    # 3. Валидируем и нормализуем данные
    validated_products = []
    for product in products:
        normalized = normalize_product(product, brands, categories, attributes)
        if validate_product(normalized):
            validated_products.append(normalized)

    # 4. Загружаем батчами по 100 товаров
    results = []
    for i in range(0, len(validated_products), 100):
        batch = validated_products[i:i+100]
        result = create_products_batch(batch)
        results.extend(result)
        logger.info(f"Uploaded {i+len(batch)}/{len(validated_products)} products")
        time.sleep(2)  # Пауза между батчами

    # 5. Отчет
    successful = sum(1 for r in results if r['success'])
    failed = len(results) - successful

    logger.info(f"Upload completed: {successful} successful, {failed} failed")

    return results

Сценарий 2: Ежедневная синхронизация заказов

Задача: Каждые 15 минут получать новые заказы

from apscheduler.schedulers.blocking import BlockingScheduler

def sync_orders_job():
    """Job для синхронизации заказов"""
    logger.info("Starting order synchronization...")

    try:
        # Получаем время последней синхронизации
        last_sync = get_last_sync_timestamp()

        # Загружаем новые заказы
        new_orders = fetch_orders_since(last_sync)

        if not new_orders:
            logger.info("No new orders")
            return

        # Обрабатываем каждый заказ
        for order in new_orders:
            try:
                process_order(order)
            except Exception as e:
                logger.error(f"Failed to process order {order['orderNr']}: {e}")
                continue

        # Обновляем время последней синхронизации
        update_last_sync_timestamp()

        logger.info(f"Synchronized {len(new_orders)} orders")

    except Exception as e:
        logger.error(f"Order sync failed: {e}")
        raise

# Настройка планировщика
scheduler = BlockingScheduler()
scheduler.add_job(
    sync_orders_job,
    'interval',
    minutes=15,
    id='sync_orders'
)

scheduler.start()

Сценарий 3: Обновление цен

Задача: Обновить цены на 5,000 товаров

def update_prices_strategy_1():
    """Стратегия 1: Полное обновление (для малых объемов)"""
    products = get_all_products()

    price_updates = []
    for product in products:
        new_price = calculate_price(product)
        price_updates.append({
            'sku': product['sku'],
            'price': new_price
        })

    # Один батч на все обновления
    return bulk_update_prices(price_updates)

def update_prices_strategy_2():
    """Стратегия 2: Инкрементальное обновление (для больших объемов)"""
    # Получаем только измененные цены
    price_changes = get_price_changes_from_erp()

    # Батчами по 100
    for i in range(0, len(price_changes), 100):
        batch = price_changes[i:i+100]
        bulk_update_prices(batch)
        time.sleep(1)

Сценарий 4: Обработка возвратов FBS

Задача: Обработать поступившие возвраты

def process_fbs_returns():
    """Обработка возвратов FBS"""

    # 1. Получаем новые возвраты
    return_boxes = get_fbs_return_boxes(status='IN_PROGRESS')

    for box in return_boxes:
        try:
            # 2. Получаем товары в коробке
            items = get_fbs_return_items(box['boxId'])

            # 3. Проверяем товары
            for item in items:
                if should_accept_return(item):
                    accept_return_item(item['itemId'])
                else:
                    reject_return_item(item['itemId'], reason='Damage')

            # 4. Закрываем коробку
            close_return_box(box['boxId'])

        except Exception as e:
            logger.error(f"Failed to process return box {box['boxId']}: {e}")

15. Частые проблемы и решения

Проблема 1: "401 Unauthorized"

Причины:

Токен истек
Неверный client_id/client_secret
Токен отозван

Решение:

# Автоматическое обновление токена
def handle_401_error():
    """Обработка 401 ошибки"""
    # Инвалидируем токен
    token_manager.revoke_tokens()

    # Пробуем снова с новым токеном
    return retry_request()

Профилактика:

Обновляйте токены проактивно
Проверяйте срок действия токена
Логируйте события обновления

Проблема 2: "429 Too Many Requests"

Причины:

Превышен rate limit
Слишком много параллельных запросов

Решение:

# Экспоненциальный backoff
from tenacity import retry, wait_exponential

@retry(wait=wait_exponential(multiplier=1, min=4, max=60))
def api_call_with_backoff():
    # API call here
    pass

Профилактика:

Используйте rate limiting
Применяйте exponential backoff
Кешируйте запросы

Проблема 3: "400 Bad Request - Validation Error"

Причины:

Неверный формат данных
Обязательные поля отсутствуют
Неверные значения enum

Решение:

# Детальная валидация
def validate_product_data(product: dict) -> tuple:
    """Валидация данных товара"""
    errors = []

    # Проверка обязательных полей
    required_fields = ['sku', 'name', 'brand', 'category']
    for field in required_fields:
        if field not in product:
            errors.append(f"Missing required field: {field}")

    # Проверка формата SKU
    if 'sku' in product and not validate_sku(product['sku']):
        errors.append(f"Invalid SKU format: {product['sku']}")

    # Проверка диапазона значений
    if 'price' in product and product['price'] <= 0:
        errors.append("Price must be positive")

    return len(errors) == 0, errors

Проблема 4: Медленная работа API

Причины:

Слишком много последовательных запросов
Большие объемы данных
Неоптимизированные запросы

Решение:

# Оптимизация через параллельность
from concurrent.futures import ThreadPoolExecutor

def parallel_fetch_orders(order_ids: list):
    """Параллельное получение заказов"""
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(fetch_order, order_id)
            for order_id in order_ids
        ]
        return [f.result() for f in futures]

Проблема 5: "Order status transition not allowed"

Причины:

Попытка изменить статус на недопустимый
Заказ уже в финальном статусе

Решение:

# Проверка допустимости перехода
ALLOWED_TRANSITIONS = {
    'New': ['Processing', 'Canceled'],
    'Processing': ['ReadyForShipment', 'Canceled'],
    'ReadyForShipment': ['Shipped', 'Canceled'],
    'Shipped': ['Delivered', 'NotDelivered'],
    'Delivered': ['Returned'],  # Финальный статус
    'Canceled': [],  # Финальный статус
    'Returned': []   # Финальный статус
}

def can_transition_to(current_status: str, new_status: str) -> bool:
    """Проверка допустимости перехода статуса"""
    allowed = ALLOWED_TRANSITIONS.get(current_status, [])
    return new_status in allowed

16. Полезные ресурсы

Официальная документация

Lamoda Seller Academy: https://academy.lamoda.ru/
- Основная документация для продавцов
- Гайды по интеграции
- FAQ
B2B Platform API OpenAPI Spec: https://b2b-guide.lamoda.ru/specification
- Полная спецификация B2B Platform API
- Примеры запросов/ответов
GitHub - Lamoda PHP SDK: https://github.com/lamoda/lamoda-b2b-platform.php-sdk
- Официальный PHP SDK
- Примеры использования

Сообщество и поддержка

Email поддержки: api-integration@lamoda.ru
Время ответа: ~4 часа
GitHub Issues: https://github.com/lamoda/lamoda-b2b-platform.php-sdk/issues

Инструменты разработчика

Postman: Для тестирования API запросов
Swagger UI: https://api.sellercenter.lamoda.ru/docs/
curl: Для быстрой проверки
jq: Для обработки JSON ответов

Библиотеки

Python:

# requirements.txt
requests>=2.31.0
tenacity>=8.2.3  # Retry логика
prometheus-client>=0.19.0  # Мониторинг
redis>=5.0.0  # Кеширование
python-dotenv>=1.0.0  # Environment variables
aiohttp>=3.9.0  # Асинхронные запросы

Пример установки:

pip install -r requirements.txt

Резюме

Ключевые принципы успешной интеграции

Начните с малого: Тестируйте на демо-среде
Используйте retry логику: Обрабатывайте временные ошибки
Логируйте всё: Детальные логи помогут при отладке
Мониторьте метрики: Отслеживайте производительность и ошибки
Обновляйте токены проактивно: Избегайте 401 ошибок
Используйте кеширование: Снижайте нагрузку на API
Тестируйте перед релизом: Unit, Integration, E2E тесты
Планируйте масштабирование: Используйте очереди для больших объемов

Чеклист готовности к продакшену

Контакты и поддержка

Техническая поддержка Lamoda:

Email: api-integration@lamoda.ru
Документация: https://academy.lamoda.ru/
GitHub: https://github.com/lamoda

При проблемах:

Проверьте логи
Проверьте статус токена
Проверьте rate limits
Сверьтесь с документацией
Свяжитесь с поддержкой

Дата создания: 2025-02-10 Версия: 1.0 Статус: Полная документация

Date: 2025-02-10​

Overview​

Table of Contents​

1. Подготовка к интеграции​

1.1. Необходимые условия​

1.2. Выбор бизнес-модели​

1.3. Понимание архитектуры API​

1.4. Планирование интеграции​

2. Пошаговый процесс интеграции​

Шаг 1: Получение учетных данных API​

Шаг 2: Настройка тестового окружения​

Шаг 3: Реализация аутентификации​

Шаг 4: Создание базового HTTP клиента​

Шаг 5: Реализация базовых сервисов​

Шаг 6: Тестирование интеграции​

Шаг 7: Развертывание в продакшен​

3. Выбор API системы​

Сравнительная таблица API систем​

Рекомендации по выбору​

4. Аутентификация и управление токенами​

Стратегия обновления токенов​

Лучшие практики управления токенами​

Пример реализации с кешированием​

5. Архитектура интеграции​

Рекомендуемые паттерны интеграции​

5.1. Синхронная интеграция (для малых объемов)​

5.2. Асинхронная интеграция (через очередь)​

5.3. Гибридный подход (рекомендуется)​

Структура данных​

6. Лучшие практики разработки​

6.1. Работа с товарами​

6.2. Работа с заказами​

6.3. Управление ценами​

6.4. Работа с остатками​

7. Обработка ошибок и повторные попытки​

Иерархия исключений​

Retry стратегии​

Обработка конкретных сценариев​

8. Работа с Rate Limits​

Консервативная стратегия (начните с нее)​

Адаптивный Rate Limiting​

Использование нескольких токенов для параллельности​

9. Тестирование интеграции​

Стратегия тестирования​

Unit Tests (с pytest)​

Integration Tests (с pytest и responses)​

Load Tests (с locust)​

План тестирования​

10. Развертывание в продакшен​

Контрольный checklist​

Environment Variables​

Docker конфигурация​

План отката (Rollback)​

11. Мониторинг и оповещения​

Ключевые метрики​

Реализация мониторинга​

Алерты (Prometheus AlertManager)​

Логирование​

12. Безопасность​

Защита учетных данных​

Хранение секретов​

Пример с Vault​

Безопасность в коде​

13. Производительность и оптимизация​

Кеширование​

Оптимизация пакетных операций​

Асинхронная обработка​

14. Типовые сценарии интеграции​

Сценарий 1: Первичная загрузка каталога​

Сценарий 2: Ежедневная синхронизация заказов​

Сценарий 3: Обновление цен​

Сценарий 4: Обработка возвратов FBS​

15. Частые проблемы и решения​

Проблема 1: "401 Unauthorized"​

Проблема 2: "429 Too Many Requests"​

Проблема 3: "400 Bad Request - Validation Error"​

Проблема 4: Медленная работа API​

Проблема 5: "Order status transition not allowed"​

16. Полезные ресурсы​

Официальная документация​

Date: 2025-02-10

Overview

Table of Contents

1. Подготовка к интеграции

1.1. Необходимые условия

1.2. Выбор бизнес-модели

1.3. Понимание архитектуры API

1.4. Планирование интеграции

2. Пошаговый процесс интеграции

Шаг 1: Получение учетных данных API

Шаг 2: Настройка тестового окружения

Шаг 3: Реализация аутентификации

Шаг 4: Создание базового HTTP клиента

Шаг 5: Реализация базовых сервисов

Шаг 6: Тестирование интеграции

Шаг 7: Развертывание в продакшен

3. Выбор API системы

Сравнительная таблица API систем

Рекомендации по выбору

4. Аутентификация и управление токенами

Стратегия обновления токенов

Лучшие практики управления токенами

Пример реализации с кешированием

5. Архитектура интеграции

Рекомендуемые паттерны интеграции

5.1. Синхронная интеграция (для малых объемов)

5.2. Асинхронная интеграция (через очередь)

5.3. Гибридный подход (рекомендуется)

Структура данных

6. Лучшие практики разработки

6.1. Работа с товарами

6.2. Работа с заказами

6.3. Управление ценами

6.4. Работа с остатками

7. Обработка ошибок и повторные попытки

Иерархия исключений

Retry стратегии

Обработка конкретных сценариев

8. Работа с Rate Limits

Консервативная стратегия (начните с нее)

Адаптивный Rate Limiting

Использование нескольких токенов для параллельности

9. Тестирование интеграции

Стратегия тестирования

Unit Tests (с pytest)

Integration Tests (с pytest и responses)

Load Tests (с locust)

План тестирования

10. Развертывание в продакшен

Контрольный checklist

Environment Variables

Docker конфигурация

План отката (Rollback)

11. Мониторинг и оповещения

Ключевые метрики

Реализация мониторинга

Алерты (Prometheus AlertManager)

Логирование

12. Безопасность

Защита учетных данных

Хранение секретов

Пример с Vault

Безопасность в коде

13. Производительность и оптимизация

Кеширование

Оптимизация пакетных операций

Асинхронная обработка

14. Типовые сценарии интеграции

Сценарий 1: Первичная загрузка каталога

Сценарий 2: Ежедневная синхронизация заказов

Сценарий 3: Обновление цен

Сценарий 4: Обработка возвратов FBS

15. Частые проблемы и решения

Проблема 1: "401 Unauthorized"

Проблема 2: "429 Too Many Requests"

Проблема 3: "400 Bad Request - Validation Error"

Проблема 4: Медленная работа API

Проблема 5: "Order status transition not allowed"

16. Полезные ресурсы

Официальная документация