Коммуникация по RESTful API: синхронизируем данные с маркетплейсами

В статье разбирается код на Ruby и в Ruby on Rails, в частности, на примере задачи по разработке web-сервиса синхронизации данных с внешними источниками. Погружение в программное решение начинается с разбора бизнес-задачи. Через освещение реальных API с маркетплейсов OZON и Яндекс.Маркет обосновываются способы принятия архитектурных решений и способы оптимизации кода. Эта статья также является авторской попыткой раскрыть принцы SOLID при реализации логики реального бэкенд приложения по переработке структурированных данных в условиях эксплуатации стороннего REST API.

I. Введение

1. О контексте бизнес-задачи

Мы обращаемся к теме продаж через маркетплейсы по причине её огромной популярности. Поскольку площадка продаж электронная, то и взаимодействие с ней подразумевает использование автоматизированных программных инструментов. В качестве таковых выступает либо личный кабинет клиента маркетплейса, либо стороннее программное решение, общающееся с маркетплейсом через REST API. Нас, как разработчиков, будет интересовать второй вариант в контексте данной статьи.

Попробуем подробнее описать контекст стоящей перед нами задачи по разработке сервиса, помогающего предпринимателям взаимодействовать с маркетплейсами. Вообще маркетплейсы выбираются в качестве площадки продаж по причине ожидаемого гарантированного трафика потенциальных покупателей. Однако доступ к услугам маркетплейсов предполагает некоторые расходы в процентах от продаж. Поэтому у предпринимателей к определенному моменту времени собирается пул вопросов, требующий независимой экспертной оценки. Например: насколько популярна товарная категория продавца на конкретном маркетплейсе; как расширить аналитику по продажам, предоставляемую через маркетплейс; как лучше распределять товарные запасы между складом от маркетплейса и своим собственным. Предполагается, что в данном случае было бы идеально собрать свою собственную независимую статистику с различных маркетплейсов перед принятием дальнейших стратегических решений по выбору идеальной торговой площадки для своей группы товаров. Вот здесь и возникает потребность в некоторой автоматизации, когда привлекают разработчиков к её воплощению.

На данном этапе разработчикам важно понимать, что перед ними не стоит задачи разрабатывать новый программный продукт с нуля, и в тоже время решение должно быть настолько автономным, чтобы заинтересовать предпринимателя в эксплуатации их сервиса на долгосрочной основе. Тут мы приходим к первой идее будущего программного сервиса, которая заключается в копировании структуры данных с маркетплейса в собственное решение. Эта структура будет базироваться на трех фундаментальных сущностях - Товар, Заказ, Транзакция. Под Товаром будем понимать совокупность данных, касающихся номенклатуры продаж. Под Заказом будем рассматривать совокупность данных, характеризующих перемещение товара от продавца к клиенту. И под Транзакцией будем рассматривать совокупность финансовых обязательств, возникающих на момент совершения сделки покупателя с продавцом.

Внимательный читатель может сделать замечание о том, что здесь не хватает ещё одной важной сущности – Склад, покрывающей ряд организационных моментов, связанных с хранением товаров и логистикой. Это справедливо в отношении производителей и глобальных импортёров. Однако на текущий момент мы рассматриваем розничных продавцов, для которых идеальная схема продаж – «с колёс». Поэтому для них эта сущность, очевидно, может быть избыточной, а следовательно, и мы последуем их логике.

2. О предмете автоматизации

Чтобы приблизиться к более детальному исследованию программного решения, которое будет разрабатываться для потенциальных заказчиков-предпринимателей, разберём ряд вопросов, над которыми надо подумать разработчикам перед тем, как приступать к кодингу.

Потенциальными потребителями автоматизированного решения со стороны заказчика будут лица, принимающие решения, менеджеры по продажам и маркетологи. Ещё предполагается, что время от времени к разработчикам будут обращаться представители от заказчика за технической поддержкой. На данном этапе стоит осознать, что у одного заказчика не будет средств на поддержание и развитие сложного автоматизированного решения. С самого начала разработки Web-сервиса стоит иметь ввиду, что им будет пользоваться условно неограниченное количество пользователей. При этом большинство из них уже должны признавать недостаточность для их бизнеса публичных средств от маркетплейсов и нуждаться в дополнительных программных инструментах. То есть разработчикам предстоит создать Web-приложение, обладающее своей собственной системой авторизации и некоторым количеством сервисов по обработке данных с маркетплейсов, а также рядом инструментов, обеспечивающих внутреннюю коммуникацию между разработчиками и пользователями (продавцами).

На данный момент мы определили предмет автоматизации и тип приложения. Далее перейдём к его архитектурным особенностям и инфраструктурным потребностям. Поскольку приложение должно обеспечить коммуникацию с маркетплейсами – поставщиками данных, и коммуникацию с пользователями – потребителями данных, то мы можем воспроизвести такую же модель Web-приложения, состоящего из нескольких относительно автономных частей. То есть требуется создать фоновое приложение, умеющее коммуницировать по REST API в глобальной сети с поставщиками данных, и визуальное приложение, умеющее отображать систематизированную информацию из собранных данных на дэшбордах. Это две активные части Web-приложения. В качестве пассивной части приложения, и фактически связующим звеном между двумя перечисленными, выступает хранилище данных.

Итак, у нас уже начинает складываться структура Web-приложения, которое должно состоять из фронтенда для коммуникации с пользователями и бэкенда, обслуживающего фронтенд данными из БД. И ещё из одного бэкенда, коммуницирующего с маркетплейсами, и из реляционной базы данных, хранящей историю данных клиента. Мы уже понимаем, что по мере увеличения нагрузки на арендуемые сервера приложение будет масштабироваться горизонтально посредством запуска большего количества параллельно работающих модулей над однотипными задачами. По мере увеличения нагрузки на СУБД её можно будет превратить в кластер, чтобы отделить аналитический блок от сервисного блока по сбору и актуализации данных о клиентах.

На этом вводную часть статьи мы завершаем. Дальнейшей темой нашего дискурса будет задача по загрузке данных с разных маркетплейсов в собственную базу данных. В рамках этой задачи обсудим механизм приёма данных, способ обработки данных с отделением бизнесовых вопросов от деталей технической реализации самого процесса загрузки, а также разберём вопрос трансформации данных перед их сохранением в базу данных.

II. REST API. Открываем двери на маркетплейсы

Для коммуникации с маркетплейсами организуем сервис-приложение, которое будет работать в фоновом режиме по расписанию. Далее сконцентрируемся на подзадаче по загрузке Товаров чтобы развернуть детали организуемой нами внутренней структуры приложения. Некоторая часть этой программной структуру также будет использована для загрузки данных о Заказах и Транзакциях. Сразу отметим, что хранение на собственных серверах «сырых» данных с маркетплейсов о каждом заказчике – дорогое мероприятие, поэтому имеет смысл предусмотреть первичную упаковку данных на этапе её загрузки с маркетплейсов в объёме, необходимом строго для будущей аналитической системы.

1. Собираемые данные

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

• Yandex. Информация о товарах в каталоге

• OZON. Получить список товаров по идентификаторам

Мы бы хотели сохранять информацию из указанных методов в одну таблицу с привязкой к пользователю. В качестве описания товара было выбрано:

/app/services/yandex/products_downloader/importing_scheme.rb

module Yandex
  class ProductsDownloader
    module ImportingScheme
      def prepare_product(product, item)
        offer = item[:offer]
        mapping = item[:mapping]
        product.assign_attributes(
          {
            name:           offer.fetch(:name, ''),
            barcodes:       offer.fetch(:barcodes, []),
            price:          find_price(offer),
            status:         find_status(offer),
            schemes:        find_schemes(offer),
            images:         offer.fetch(:pictures, []),
            description:    offer.fetch(:description, nil),
            product_id:     mapping.fetch(:marketModelId, nil),
            skus:           find_skus(mapping),
            category_title: find_category_title(offer, mapping),
            scrub_status:   'success'
          }
        )
        product
      end
    end
  end
end

/app/services/ozon/products_downloader/importing_scheme.rb

module Ozon
  class ProductsDownloader
    module ImportingScheme
      def prepare_product(product, item)
        product.assign_attributes(
          {
            name:           item.fetch(:name, ''),
            offer_id:       item[:offer_id],
            barcodes:       find_barcodes(item),
            price:          find_price(item),
            status:         find_status(item),
            schemes:        find_schemes(item),
            images:         item[:images],
            skus:           find_skus(item),
            category_title: find_category_title(item),
            stock:          item.dig(:stocks, :present),
            scrub_status:   'success'
          }
        )
        product
      end

При разработке учётной системы товаров с маркетплейсов формирование такой сводной таблицы с характеристиками товара обязывает разработчика учесть несколько нюансов. Далее по тексту перечислим эти нюансы и подробно обсудим их детали.

2. Периодичность обновления информации о товарах

Предполагается, что продавец сможет управлять процессом актуализации базы товаров в приложении вручную по собственному усмотрению. Эта возможность предоставляется из соответствующего контроллера.

/app/controllers/api/v1/credentials_controller.rb

module Api
  module V1
    class CredentialsController < ApplicationController
      before_action :fetch_credential, only: %i[update archive descriptions]
 
      def update
        resp, status = Tasks::ReimportProducts.new(true, @mp_credential, I18n.t('messages.users_control')).call
        render json: resp, status:
      end

      private

      def fetch_credential
        @mp_credential = fetch_credentials.find_by(id: params[:id])
        if @mp_credential.nil?
          render json: {
            errors: [{
              code: 'error',
              title: I18n.t('errors.not_found', class_name: 'MarketplaceCredential')
            }]
          }, status: 404
        else
          @mp_credential
        end
      end

      def fetch_credentials
        current_user.marketplace_credentials
      end
    end
  end
end

Но со своей стороны мы также представляем жизненный цикл описания товара статичным в течении суток, после чего он может измениться. Этот правило положим в основу работы приложения по умолчанию через менеджера задач CRON.

/clock.rb

module Clockwork
  every(1.day, 'Import products', at: '23:00') do
    MarketplaceInteraction::ImportProductsJob.perform_later
  end
end

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

После введения на маркетплейсах режима загрузки архивных товаров такая опция перестала быть актуальной, но её можно использовать при отключении режима загрузки архивных данных.

/app/services/yandex/products_downloader.rb

module Yandex
  class ProductsDownloader
    attr_reader :mp_credential, :http_client, :parsed_ids

    private

    def tag_lost_products!
      products = Product.where(marketplace_credential_id: mp_credential.id)
      lost_offer_ids = products.pluck(:offer_id) - parsed_ids

      products.where(
        offer_id: lost_offer_ids
      ).update_all(scrub_status: 'gone')
    end
  end
end

/app/services/ozon/products_downloader.rb

module Ozon
  class ProductsDownloader
    attr_reader :mp_credential, :parsed_ids, :http_client_list, :http_client_info, :http_client_description

    private

    def tag_lost_products!
      products = Product.where(marketplace_credential_id: mp_credential.id)
      lost_product_ids = products.pluck(:product_id) - parsed_ids.keys

      products.where(
        product_id: lost_product_ids
      ).update_all(scrub_status: 'gone')
    end
  end
end

3. У товаров на разных маркетплейсах существуют различные способы их уникальной идентификации

На Яндекс.Маркет — это offerId. У OZON — это product_id или даже просто id, как «Номер задания на формирование документов».

Это является причиной разнесения процедуры обновления товаров от разных маркетплейсов в разные классы. Только в таком случае возможно сопоставлять поступившие данные с данными из базы по уникальному id от маркетплейса.

/app/services/ozon/products_downloader/importing_scheme.rb

module Ozon
  class ProductsDownloader
    module ImportingScheme
      def iterate(list_by_id, exists = [], updated_products = [], updated_fields = [])
        Product.where(
          marketplace_credential_id: mp_credential.id,
          product_id: list_by_id.keys
        ).find_each do |product|
          exists << product.product_id
          product = prepare_product(product, list_by_id[product.product_id])
          @parsed_ids[product.product_id] = 1
          # We can record the changes somewhere.
          # pp("product.changes=",product.changes) if product.changed?
          if product.changed?
            updated_products << product
            updated_fields += product.changes.keys
          end
        end
        [updated_products, updated_fields, exists]
      end
    end
  end
end

/app/services/yandex/products_downloader/importing_scheme.rb

module Yandex
  class ProductsDownloader
    module ImportingScheme
      def iterate(list_by_id, exists = [], updated_products = [], updated_fields = [])
        Product.where(
          marketplace_credential_id: mp_credential.id,
          offer_id: list_by_id.keys
        ).find_each do |product|
          exists << product.offer_id
          product = prepare_product(product, list_by_id[product.offer_id])
          # We can record the changes somewhere.
          # pp("product.changes=",product.changes) if product.changed?
          if product.changed? && imported?(product)
            updated_products << product
            updated_fields += product.changes.keys
          end
        end
        [updated_products, updated_fields, exists]
      end
    end
  end
end

Кстати, у Яндекс.Маркет нет однозначной характеристики, соответствующей нашему параметру product_id. Наиболее близким по семантике становится marketModelId — что в описании API определяется как «Идентификатор модели на Маркете, который может отсутствовать в ответе, если товар ещё не привязан к карточке».

4. Тип данных для идентификаторов

У OZON для offer_id, product_id, sku реализован самый очевидный формат — числовой. Однако offerId на Яндекс.Маркет уже строковый. На других маркетплейсах тоже часто используется строковый тип данных для перечисленных параметров. Поэтому у нас идентификаторы будем хранить только в строках. Эта информация является востребованной СУДБ.

/db/migrate/20240106161532_create_products.rb

class CreateProducts < ActiveRecord::Migration[7.1]
  def up
    create_enum :product_scrub_status, ["unspecified", "success", "gone"]
    create_enum :product_status, ["preliminary", "on_moderation", "failed_moderation", "published", "unpublished", "archived"]

    execute <<-SQL
      DO $$
      BEGIN
        IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'monetary_amount') THEN
          CREATE TYPE monetary_amount AS
          (
            value     float,
            currency  VARCHAR(3)
          );
        END IF;
      END$$;
    SQL

    create_table :products do |t|
      t.references :marketplace_credential, type: :uuid, null: false, foreign_key: true
      t.string :offer_id, comment: 'client SKU'
      t.string :product_id, comment: 'marketplace object, articule or model'
      t.string :name, null: false
      t.text :description
      t.string :skus, array: true, comment: 'marketplace SKUs'
      t.string :images, array: true
      t.string :barcodes, array: true
      t.enum :status, enum_type: :product_status, default: "preliminary", null: false
      t.enum :scrub_status, enum_type: :product_scrub_status, default: "unspecified", null: false
      t.column :price, :monetary_amount
      t.integer :stock
      t.string :category_title
      t.string :schemes, array: true, comment: 'sales schemes'

      t.timestamps
    end
  end
end

5. О пакетной загрузке данных

Маркетплейсы имеют разные стратегии обслуживания клиентских приложений по API. Яндекс.Маркет быстро отдаёт необходимые данные, накладывая лишь ограничение на количество товаров в одном запросе в размере 200 позиций и разрешая до 600 запросов в минуту. Это значит, что в минуту можно загружать до 120_000 товарных позиций, чего с лихвой должно хватать для большинства продавцов на маркетплейсе без привлечения специальных алгоритмических техник по соблюдению rate limits при коммуникации с маркетплейсом. Для клиентов с большим количеством данных при установленных лимитах от маркетплейса можно позволить себе оставаться с алгоритмом синхронной обработки данных, и воспользоваться слиппером для пережидания ограничения на запрос к маркетплейсу.

/app/services/yandex/sleeper.rb

module Yandex
  module Sleeper
    # INPUT:
    # headers = {
    #   Date: Tue, 09 Jan 2024 09:14:10 GMT
    #   X-RateLimit-Resource-Until: Tue, 09 Jan 2024 09:15:00 GMT
    # }
    def do_sleep(headers, duration)
      dt =
        Time.parse(headers['X-RateLimit-Resource-Until']) -
        Time.parse(headers['Date']) +
        1
      if dt > duration
        Rails.logger.error I18n.t('errors.duration_of_rate_limit_has_been_changed',
                                  marketplace_name: 'Yandex')
      else
        sleep dt
      end
    rescue StandardError => e
      # We are checking the code. It's fixable
      ErrorLogger.push_trace e
    end
  end
end

При этом можно воспользоваться заголовочными параметрами X-RateLimit-Resource-Remaining и X-RateLimit-Resource-Until из последнего запроса к маркетплейсу.

/app/services/yandex/api/offer_mappings.rb

module Yandex
  class Api
    class OfferMappings < Api
      MAX_REQUESTS_PER_MINUTE = 600
      RATE_LIMIT_DURATION = 60

      include Yandex::Sleeper

      private

      # INPUT:
      # headers = {
      #   X-RateLimit-Resource-Remaining: 600
      # }
      def delay_if_limits(headers)
        if headers.fetch(
          'X-RateLimit-Resource-Remaining',
          MAX_REQUESTS_PER_MINUTE
        ).to_i < limiting_remaining_requests
          do_sleep(headers, RATE_LIMIT_DURATION)
        end
      end
    end
  end
end

У OZON нет лимитов на количество загрузок данных в единицу времени, но при этом в целом ситуация немного хуже по нескольким причинам. Первая, OZON задерживает время отклика на запрос. Вторая, OZON исключает возможность получать данные из одного метода с маркетплейса по ожидаемому перечню параметров. Только из отдельных запросов можно получить данные об описании продукта description и о наименовании категории товара category_title.

Что касается каталог с категориями товаров, то его можно скачать у OZON отдельно.

Дерево категорий и типов товаров (версия 2)

/app/services/ozon/load_categories.rb

Struct.new('CategoryOzon', :name, :id, :disabled)

module Ozon
  class LoadCategories
    def call
      @http_client = http_client
      list = http_client_call
      @imported_list = []
      scalp(
        list: list[:result],
        previous_category: Struct::CategoryOzon.new(name: '')
      )
      OzonCategory.import @imported_list,
                          on_duplicate_key_ignore: true,
                          on_duplicate_key_update: {
                            conflict_target: %i[
                              description_category_id
                              type_id
                            ],
                            columns: %i[
                              category_name
                              category_disabled
                              type_name
                              type_disabled
                            ]
                          }
    end
  end
end

В таком случае можно использовать дополнительные запросы к собственной базе во время загрузки описаний товаров.

/app/services/ozon/products_downloader/importing_scheme.rb

module Ozon
  class ProductsDownloader
    module ImportingScheme
      def find_category_title(item)
        return if item[:description_category_id].blank? && item[:type_id].blank?

        CashOzonCategory.get(
          item[:description_category_id] || 0,
          item[:type_id] || 0
        )
      end
    end
  end
end

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

/app/services/cash_ozon_category.rb

class CashOzonCategory
  class << self
    def o_cat
      @o_cat ||= Kredis.hash 'ozon_categories'
    end

    def get(category, type)
      o_cat["#{category}_#{type}"] || take_cat(category, type)
    end

    def take_cat(category, type)
      obj = OzonCategory.find_by(
        description_category_id: category,
        type_id: type
      )
      return if obj.nil?

      o_cat["#{category}_#{type}"] = "#{obj.category_name}/#{obj.type_name}"
    end
  end
end

Загружать на ежедневной основе каталог категорий нет необходимости, и какого-то явного графика в обязательном осуществлении обновления каталога категорий тоже не прослеживается. Поэтому вопрос обновления каталога товаров для OZON оставим до момента появления проблемы с прекращением заполнения позиции с категорией товара в каталоге продавца внутри нашей системы.

Теперь вернёмся к текстовому описанию продукта description, который можно загружать строго под одну товарную единицу, как разрешает OZON.

Получить описание товара

15 тыс. товарных позиций грузится около 5 минут. Это очень долго. Поэтому мы выносим этот процесс в фоновую задачу, отделённую от процедуры загрузки описаний товаров.

/app/services/ozon/products_downloader/import_desriptions.rb

module Ozon
  class ProductsDownloader
    module ImportDesriptions
      def downloading_product_desriptions
        Product.where(
          marketplace_credential_id: @mp_credential.id,
          product_id: @parsed_ids.keys
        ).find_in_batches(batch_size: 100) do |products|
          updated_products = []
          products.each do |product|
            product.description = load_description(product.product_id)
            updated_products << product if product.changed?
          end
          if updated_products.any?
            Product.import(updated_products,
                           on_duplicate_key_ignore: true,
                           on_duplicate_key_update: {
                             conflict_target: %i[
                               marketplace_credential_id
                               product_id
                               offer_id
                             ],
                             columns: %i[
                               description
                             ]
                           })
          end
        end
      end
    end
  end
end

Импорт описаний товаров тоже пусть запускается из CRON задачи:

/clock.rb

module Clockwork
  every(1.day, 'Import product descriptions (for Ozon)', at: '1:00') do
    MarketplaceInteraction::ImportProductDescriptionsJob.perform_later
  end
end

Поскольку OZON считает данные с описаниями товаров несущественными для API приложения, то имеет смысл в предоставлении продавцам самостоятельного решения - осуществлять или нет загрузку описаний товаров с OZON в нашу платформу. Поэтому выносим выключатель загрузчика описаний товаров в эндпоинт приложения.

/app/controllers/api/v1/credentials_controller.rb

module Api
  module V1
    class CredentialsController < ApplicationController
      before_action :fetch_credential, only: %i[update archive descriptions]

      def descriptions
        value = Handles::ProductsDownloader.to_bool(params[:value])
        @mp_credential.autoload_descriptions.value = value if value.in? [true, false]
        render json: {
          marketplace_credential: {
            id: @mp_credential.id,
            autoload_descriptions: @mp_credential.autoload_descriptions.value
          }
        }
      end

      private

      def fetch_credential
        @mp_credential = fetch_credentials.find_by(id: params[:id])
        if @mp_credential.nil?
          render json: {
            errors: [{
              code: 'error',
              title: I18n.t('errors.not_found', class_name: 'MarketplaceCredential')
            }]
          }, status: 404
        else
          @mp_credential
        end
      end
    end
  end
end

6. Привязка списка продуктов к учётным данным пользователя на маркетплейсе

В начале этой главы было упомянуто, что списки товаров должны быть связаны с пользователями сервиса. На самом деле Яндекс.Маркет подсказывает, что у пользователя на маркетплейсе может быть до нескольких бизнесов. Собственно эти бизнесы и являются промежуточным звеном между пользователями и товарами. То есть схема базы данных будет выглядеть следующим образом:

В этом случае коммуникация приложения с маркетплейсами будет устанавливаться через регистрационные данные пользователя marketplace_credentials. Поэтому, чтобы работать с маркетплейсом по REST API, Яндекс.Маркет в личном кабинете пользователя выдаёт token, а OZON - api_key.

Дальнейшая коммуникация с «рабочей средой» маркетплейса осуществляется посредством добавления в заголовочную часть запроса авторизационных данных:

Что такое рабочая среда

/app/services/ozon/api.rb

module Ozon
  class Api < BaseApi
    def headers
      super.merge(
        {
          'Api-Key' => @api_key,
          'Client-Id' => @client_id,
          'x-o3-app-name' => ENV.fetch('OZON_APP_ID')
        }
      )
    end
  end
end

Авторизация для запросов магазина к Маркету

/app/services/yandex/api.rb

module Yandex
  class Api < BaseApi
    def headers
      super.merge(
        {
          'Authorization' => "OAuth oauth_token=\"#{@token}\", oauth_client_id=\"#{ENV.fetch('YANDEX_APP_ID')}\""
        }
      )
    end
  end
end

При длительной эксплуатации ключи доступа могут устаревать по разным причинам, поэтому мы должны предусмотреть автоматическую проверку регистрационных данных на их актуальность. В случае OZON такую проверку можно осуществить через любой запрос к существующему методу API.

В случае Яндекс.Маркет проверка регистрационных данных может быть осуществлена через конкретный URL, который в своём адресе может содержать какой-нибудь дополнительный Id. Например, запрос за описаниями товаров содержит в себе business_id:

https://api.partner.market.yandex.ru/businesses/{businessId}/offer-mappings

Актуальность этого Id можно проверить только через реальный запрос

Информация о товарах в каталоге

чтобы по реальному HTTP-ответу получить статус ошибки или подтверждение об успешности выполнения запроса

Типы ошибок и что с ними делать

В нашем случае можно использовать универсальный механизм для обоих маркетплейсов по обработке статусов HTTP-запросов

/app/services/base_api.rb

class BaseApi
  EXACT_ERROR_RAISER = {
    420 => ->(status, msg, mp_id) { raise MarketplaceAggregator::ApiLimitError.new(status, msg, mp_id) },
    401 => ->(status, msg, mp_id) { raise MarketplaceAggregator::ApiAccessDeniedError.new(status, msg, mp_id) },
    403 => ->(status, msg, mp_id) { raise MarketplaceAggregator::ApiAccessDeniedError.new(status, msg, mp_id) }
  }.freeze

  RANGE_ERROR_RAISER = {
    (400...500) => ->(status, msg, mp_id) { raise MarketplaceAggregator::ApiBadRequestError.new(status, msg, mp_id) },
    (500..) => ->(status, msg, mp_id) { raise MarketplaceAggregator::ApiError.new(status, msg, mp_id) }
  }.freeze

  def call(method:, raise_an_error: false, params: {}, body: {})
    # Let's define an HTTP method for a specific endpoint
    @raise_an_error = raise_an_error
    send(method, params, body)
  end

  private

  def get(params = {}, _ = {})
    api_call do
      connection.get do |req|
        req.url url
        req.params = params
        req.headers = headers
      end
    end
  end

  def post(params = {}, body = {})
    api_call do
      connection.post do |req|
        req.url url
        req.params = params
        req.headers = headers
        req.body = body.to_json
      end
    end
  end

  def api_call
    response = yield
    @status = response.status
    @response_content_type = content_type(response.headers)
    body = response_parse(response.body)
    raise_error(
      error_message(body)
    )

    [@status, response.headers, body]
  end

  def content_type(headers)
    case headers.[]('content-type')
    when %r{application/json}
      :json
    when %r{text/html}
      :html
    else
      :any
    end
  end

  def response_parse(body)
    if response_content_type == :json
      begin
        JSON.parse body, symbolize_names: true
      rescue JSON::ParserError => e
        ErrorLogger.push e
        {}
      end
    else
      body
    end
  end

  def raise_error(message)
    return unless @raise_an_error

    EXACT_ERROR_RAISER[status]&.call(status, message, @marketplace_credential.id)

    RANGE_ERROR_RAISER.each do |k, v|
      v.call(status, message, @marketplace_credential.id) if k.include?(status)
    end
  end
end

Недостаток проверки регистрационных данных через реальный запрос заключается в сокращении количества допустимых запросов в единицу времени (то есть в расходовании rate limits).

По идее, в разрешении данной проблемы должен помочь HTTP-метод HEAD, вида:

irb(main):01> response = Faraday.head(
    'https://api.partner.market.yandex.ru/businesses/:business_id/offer-mappings.json',
    { limit: 1 },
    {
      'Authorization' => "OAuth oauth_token=\"y0_token\", oauth_client_id=\"#{ENV.fetch('YANDEX_APP_ID')}\"",
      'Content-Type' => 'application/json'
    }
  )

На который мы бы получили статус ответа, вида:

irb(main):02> response.status
=> 200

Но мы, к сожалению, получаем только статус 405:

irb(main):02> response.status
=> 405

что означает, https://developer.mozilla.org/ru/docs/Web/HTTP/Status - Method Not Allowed

Поэтому был найден альтернативный способ проверки регистрационных данных через curl-запрос с параметром -I

/app/services/yandex/check_credentials.rb

module Yandex
  class CheckCredentials < BaseCheckCredentials
    def bash_command
      <<~`BASH`
        curl -I \
        -H 'Authorization: OAuth oauth_token=#{@mp_credential.credentials['token']}, \
        oauth_client_id=#{ENV.fetch('YANDEX_APP_ID')}' \
        -X POST https://api.partner.market.yandex.ru/businesses/\
        #{@mp_credential.credentials['business_id']}/offer-mappings.json?limit=1
      BASH
    end
  end
end

В HTTP-ответе находится только текст со статусом и заголовками. Проверка HTTP-ответа регулярным выражением помогает выяснить реальный статус на запрос с введёнными регистрационными данными:

/app/services/base_check_credentials.rb

class BaseCheckCredentials
  def call
    curl = bash_command

    case message = curl.split("\r\n").first
    when %r{(HTTP/1.1 200)|(HTTP/2 200)}
      { ok: true }
    when %r{(HTTP/1.1 403)|(HTTP/2 403)|(HTTP/1.1 401)|(HTTP/2 401)|(<html>)}
      # find out details from e.message of POST request
      redo_post_request
    else
      { errors: message }
    end
  end
end

Таким образом мы повысили управляемость техническим состоянием сервиса по доставке данных с маркетплейсов.

7. Как достать данные клиента о товарах на маркетплейсе

Разобранный выше (раздел 5) способ загрузки данных с маркетплейса Ozon не заканчивается на освещённых проблемах отдельной загрузки описаний и категорий товаров. Метод выдачи списка товаров «Получить список товаров по идентификаторам» работает только при явном указании перечня идентификаторов для запрашиваемых товаров. На выбор предлагается три категории идентификаторов – offer_id, product_id, sku. Вопрос: откуда их взять? Как узнать от маркетплейса, какие товары вообще размещены пользователем на маркетплейсе?

Ответом на поставленный вопрос становится метод «Список товаров», который в пакетном режиме выдаёт списки offer_id и product_id товаров, размещённых продавцом на маркетплейсе.

В этом методе API также предоставляется возможность указания фильтров по visibility товара. Для нас актуальны два значения — ALL и ARCHIVED.

В итоге полный цикл загрузки характеристик товаров продавца с маркетплейса сводится к следующим трём этапам.

/app/services/ozon/products_downloader/downloading_scheme.rb

Первый этап. Загружаем все товары, кроме архивных.

      def downloading_unarchived_products
        @archive = false
        benchmarking(
          -> { "import: :mp_credential[#{@mp_credential.id}] — actual[#{@parsed_ids.size}] — Done" }
        ) { circle_downloader }
        @total = @parsed_ids.size
      end

Второй этап. Загружаем архивные товары.

      def downloading_archived_products
        if @mp_credential.autoload_archives.value
          @archive = true
          benchmarking(
            -> { "import: :mp_credential[#{@mp_credential.id}] — archived[#{@parsed_ids.size - @total}] — Done" }
          ) { circle_downloader }
        end
      end

Третий этап. Загружаем описания товаров по списку товаров из собственной БД.

      def downloading_desriptions
        if @mp_credential.autoload_descriptions.value
          benchmarking(
            -> { "import: :mp_credential[#{@mp_credential.id}] — desriptions[#{@parsed_ids.size}] — Done" }
          ) { downloading_product_desriptions }
        end
      en

Первые два этапа базируются на одном и том же алгоритме, в котором выполняется цикл:

• формируется запрос к странице со списком идентификаторов товаров

• загружается пакет со списками product_id товаров

• верифицируется успешность загрузки пакета данных с product_id

• вызывается процедура пакетной загрузки характеристик товаров кроме описаний товаров с последующим импортом данных в локальную базу

• достаётся токен следующей страницы для загрузки очередного пакета с product_id

• прерывается цикл, если токен пустой

/app/services/ozon/products_downloader/downloading_scheme.rb

      def circle_downloader
        page_tokens = {}
        loop do
          status, _, body = @http_client_list.call(
            body: {
              filter: {
                visibility: (@archive ? 'ARCHIVED' : 'ALL')
              },
              limit: PAGE_LIMIT
            }.merge(page_tokens)
          )
          break_if_http_error(status) if status != 200

          items = (body&.dig(:result, :items) || []).map { |elem| elem[:product_id] }
          break if items.blank?

          download_product_info_list(items)

          page_token = body&.dig(:result, :last_id)
          if page_token.blank?
            break
          else
            page_tokens = { last_id: page_token }
          end
        end
      end

Для сравнения Яндекс.Маркет работает в два этапа. Первый этап — загрузка всех товаров, кроме архивных. Второй этап — загрузка архивных данных. А цикл загрузки товаров выглядит следующим образом:

• формируется запрос к странице с характеристиками товаров

• загружается пакет с характеристиками товаров

• верифицируется успешность загрузки пакета данных с характеристиками товаров

• вызывается процедура импорта пакета данных с характеристиками товаров в локальную базу данных

• достаётся токен следующей страницы для загрузки очередного пакета с характеристиками товаров

• прерывается цикл, если токен пустой

/app/services/yandex/products_downloader/downloading_scheme.rb

      def circle_downloader
        page_tokens = {}
        loop do
          status, _, body = @http_client.call(
            params: LIMITS.merge(page_tokens),
            body: { archived: @archive }
          )
          if status == 200
            items = body&.dig(:result, :offerMappings) || []
            break if items.blank?

            import_payload(items)
          else # any other status anyway
            # To be safe, but we shouldn't get here.
            # This is possible if the status is < 400 and the status is != 200.
            raise MarketplaceAggregator::ApiError.new(
              status,
              I18n.t('errors.something_went_wrong'),
              mp_credential.id
            )
          end

          page_token = body&.dig(:result, :paging, :nextPageToken)
          if page_token.blank?
            break
          else
            page_tokens = { page_token: }
          end
        end
      end

III. Оптимизация кода. Повышение надёжности

1. Откуда берётся бизнес логика

Загрузка данных с маркетплейсов осуществляется с помощью HTTP-клиента. В нашем случае используется Faraday.

/app/services/connection.rb

class Connection
  def self.start
    @conn ||= Faraday.new do |config|
      config.request :retry, RETRY_OPTIONS
      config.adapter Faraday.default_adapter
    end
  end
end

Предполагается, что в целях обеспечения контакта с маркетплейсом во время обслуживания одной сессии или работы одной фоновой задачи может быть выполнено до нескольких обращений к маркетплейсу. При этом достаточно одного HTTP-клиента, поэтому при объявлении нового HTTP-клиента используется оператор ||= присвоения по условию отсутствия предварительного подключения к нему. Поскольку диапазон задач, для которых будет использован данный клиент, может быть различным, то мы минимизируем на данном этапе его конфигурацию. Конфигурация будет отличаться, например, при GET и POST запросах.

/app/services/base_api.rb

class BaseApi
  attr_accessor :connection

  def initialize(mp_credential, options = {})
    @connection = Connection.start
    @marketplace_credential = mp_credential
    @raise_an_error = false
  end

  private

  def get(params = {}, _ = {})
    api_call do
      connection.get do |req|
        req.url url
        req.params = params
        req.headers = headers
      end
    end
  end

  def post(params = {}, body = {})
    api_call do
      connection.post do |req|
        req.url url
        req.params = params
        req.headers = headers
        req.body = body.to_json
      end
    end
  end
end

Общий каркас настройки HTTP-клиента на этом заканчивается. Далее перейдём к реализации дополнительных параметров подключения к разным маркетплейсам из отдельных классов.

Для установления контакта с маркетплейсом используются параметры url, params, headers, body. Два из них — params и body — являются входными хэшами при вызове соответствующих методов .get и .post при обращении к Web-сервису. Два других параметра — url и headers можно настроить автоматически из отдельных классов для каждого маркетплейса. Headers содержат регистрационные данные на маркетплейсах, и они уже упоминались выше в разделе I.6 — /app/services/ozon/api.rb, /app/services/yandex/api.rb.

Url от API-методов также предполагается объявлять в отдельных классах в целях соблюдения «Принципа единственной ответственности». Так повышается Cohesion внутри кода, обслуживающего HTTP-клиента, и понижается Coupling между HTTP-клиентом и другими частями кода по переработке различных данных, загруженных с маркетплейсов. /Cohesion и Coupling: отличия/

В нашем случае имеется один API-метод на загрузку продуктов с Яндекс.Маркет:

/app/services/yandex/api/offer_mappings.rb

module Yandex
  class Api
    class OfferMappings < Api
      def url
        "#{URL}/businesses/#{@business_id}/offer-mappings.json"
      end
    end
  end
end

и три API-метода на загрузку продуктов с OZON:

/app/services/ozon/api/product_list.rb

module Ozon
  class Api
    class ProductList < Api
      def url
        "#{URL}/v2/product/list"
      end
    end
  end
end

/app/services/ozon/api/product_info_list.rb

module Ozon
  class Api
    class ProductInfoList < Api
      def url
        "#{URL}/v2/product/info/list"
      end
    end
  end
end

/app/services/ozon/api/product_info_description.rb

module Ozon
  class Api
    class ProductInfoDescription < Api
      def url
        "#{URL}/v1/product/info/description"
      end
    end
  end
end

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

/app/services/ozon/products_downloader.rb

module Ozon
  class ProductsDownloader
    def initialize(mp_credential)
      @mp_credential = mp_credential
      @http_client_list = Ozon::Api::ProductList.new(mp_credential)
      @http_client_info = Ozon::Api::ProductInfoList.new(mp_credential)
      @http_client_description = Ozon::Api::ProductInfoDescription.new(mp_credential)
      @parsed_ids = {}
    end

    def call
      return false if mp_credential.credentials.nil?

      download_products
      tag_lost_products!
      true
    end
  end
end

/app/services/yandex/products_downloader.rb

module Yandex
  class ProductsDownloader
    def initialize(mp_credential)
      @mp_credential = mp_credential
      @http_client = Yandex::Api::OfferMappings.new(mp_credential)
      @parsed_ids = []
    end

    def call
      return false if mp_credential.credentials.nil?

      download_products
      tag_lost_products!
      true
    end
  end
end

Процедуры импорта товаров на маркетплейсах запускаются из единого источника процессов - фоновая задача по импорту данных с маркетплейсов

/app/jobs/products/import_job.rb

module Products
  class ImportJob < ApplicationJob
    include MaBenchmarking

    queue_as do
      is_client_queue = arguments.first
      if is_client_queue
        :client_grabber_products
      else
        :marketplace_grabber_products
      end
    end

    DOWNLOADER_CLASS = 'ProductsDownloader'

    def perform(is_client_queue, mp_credential_id)
      @is_client_queue = is_client_queue
      @mp_credential = MarketplaceCredential.find_by(id: mp_credential_id)
      return if irrelevant?

      @mp_credential.update(last_sync_at_products: Time.current) if downloadable?
    end

    private

    def downloadable?
      import(@mp_credential.marketplace.to_constant_with(DOWNLOADER_CLASS).new(@mp_credential))
      true
      #... 
    end

    def import(downloader)
      benchmarking(
        -> { "import: :mp_credential[#{@mp_credential.id}] — [#{downloader.parsed_ids.size}] - OK" }
      ) { downloader.call }
    end
  end
end

Здесь важно обратить внимание на использование принципа инверсии зависимостей, когда по имени маркетплейса выбирается класс, в котором реализованы детали процедуры импорта - downloader.call /Принципы SOLID на примере ruby♦️/

Другим не менее значимым алгоритмическим подходом, реализованным в приведённом коде, является разнесение задачи импорта по разным очередям в зависимости от источника инициации процесса импорта. Это может быть клиентский запрос на немедленную загрузку данных, например, при регистрации нового пользователя в нашем Web-приложении. Такие запросы отправляются в очередь :client_grabber_products.

Вызов клиентского запроса на импорт данных осуществляется из соответствующего эндпоинта POST http://localhost:3000/api/v1/credentials

/app/controllers/api/v1/credentials_controller.rb

module Api
  module V1
    class CredentialsController < ApplicationController
      def create
        @mp_credential = fetch_credentials.new(credential_params)
        @mp_credential.fix_credentials!
        Operations::CheckCredentials.new(@mp_credential).call
        if @mp_credential.is_valid
          @mp_credential.save!
          Products::ImportJob.perform_later(true, @mp_credential.id)
          render json: {
            marketplace_credential: @mp_credential
          }
        else
          render json: {
            errors: [{
              code: 'error',
              title: I18n.t('errors.credentials_are_invalid'),
              detail: @mp_credential.credentials['errors']
            }]
          }, status: 400
        end
      end
    end
  end
end

В другом случае инициатором запроса на импорт данных может быть CRON-задача, о чём говорится в разделе I.2. Тогда такие запросы отправляются в очередь :marketplace_grabber_products.

Здесь стоит подчеркнуть, что в автоматической режиме на повторный импорт данных на ежедневной основе отправляется вся клиентская база:

/app/business_logic/tasks/import_products.rb

module Tasks
  class ImportProducts
    def call
      MarketplaceCredential.find_each(batch_size: 100) do |mp_credential|
        # 1. correcting client mistakes
        mp_credential.fix_credentials!
        # 2. verifying the validity of credentials
        Operations::CheckCredentials.new(mp_credential).call
        # 3. refresh the record
        mp_credential.save! if mp_credential.changed?
        # 4. cause direct data import
        Products::ImportJob.perform_later(false, mp_credential.id) if mp_credential.is_valid
      end
    end
  end
end

При этом алгоритм должен пройтись по всем клиентским данным, используемым для авторизации. Очевидно, что нам нужно перепроверить актуальность авторизационных данных перед попыткой запустить процедуру импорта товаров для последующего выявления изменений в базе. По идее, результатом проверки должны быть сделаны рассылки уведомлений тем пользователям, чьи регистрационные данные устарели и когда требуется их ручное обновление из кабинета маркетплейса. Отвечать за логику поведения данного программного блока должен администратор, который по мере накопления опыта взаимодействия с пользователями-продавцами должен иметь возможность оперативно менять данный функционал. В силу высокой зависимости данного участка кода Web-приложения от операционных намерений его менеджеров мы и вынесли этот код в отдельный раздел под названием app/business_logic.

Поскольку задача масс-импорта данных предполагает продолжительную загрузку, то здесь также организуется отдельная фоновая задача в отдельной ветке :marketplace_grabber_products.

/app/jobs/marketplace_interaction/import_products_job.rb

module MarketplaceInteraction
  class ImportProductsJob < ApplicationJob
    queue_as :marketplace_grabber_products

    def perform
      Tasks::ImportProducts.new.call
    end
  end
end

Аналогично организован код масс-импорта описаний по товарам с OZON:

/app/business_logic/tasks/import_product_descriptions.rb, /app/jobs/marketplace_interaction/import_product_descriptions_job.rb

2. Обработка ошибок

Существует несколько способов работы с программными ошибками. В данном приложении активно эксплуатируются четыре способа - перехват ошибок; генерация и перехват ошибок (Railway Oriented Programming); генерация и перехват ошибок, чтобы перезапустить задачу; генерация ошибок.

Под способом «перехвата ошибок» понимается допущение в будущем появления ошибок на этапе разработки. В задачу перехватчиков входит нормальное прерывание выполнения кода с логированием места и причины ошибки.

/app/jobs/products/import_job.rb

module Products
  class ImportJob < ApplicationJob
    def downloadable?
      import(@mp_credential.marketplace.to_constant_with(DOWNLOADER_CLASS).new(@mp_credential))
      true
    rescue NameError => e
      # We are checking the code. It's fixable
      ErrorLogger.push_trace e
      false
      #...
    end
  end
end

Здесь перехватывается ошибка неизвестного наименования класса для нового маркетплейса. Причина может быть в непосредственном имени или в неправильном размещении файла.

Второй способ, связанный с «генерированием и перехватом ошибок», эксплуатируется в двух случаях.

Первый случай — это генерация исключения с целью идентификации основания отказа в исполнении функции по эндпоинту и перехват исключения для возврата на запрос по эндпоинту форматированного ответа с указанием причины отказа

/app/controllers/application_controller.rb

class ApplicationController < ActionController::API
  rescue_from(MarketplaceAggregator::UnauthorizedError) do
    render json: {
      errors: [{
        code: 'error',
        title: I18n.t('errors.unauthorized')
      }]
    }, status: 401
  end

  rescue_from(MarketplaceAggregator::ForbiddenError) do
    render json: {
      errors: [{
        code: 'error',
        title: I18n.t('errors.forbidden')
      }]
    }, status: 403
  end

  def current_user
    raise MarketplaceAggregator::UnauthorizedError unless fetch_user_uuid

    @current_user ||= Client.find_by(id: fetch_user_uuid)
    raise MarketplaceAggregator::ForbiddenError unless @current_user

    @current_user
  end

  private

  def fetch_user_uuid
    request.headers['HTTP_USER']
  end
end

Такой способ генерации исключений до и после проверки заголовка запроса потребовался для соблюдения строгой привязки HTTP-запроса к зарегистрированному пользователю в базе данных.

В этом случае также следует учесть, что для запроса используется нестандартный заголовок запроса, поэтому Ruby on Rails добавляет к нему префикс HTTP_ . /https://github.com/rails/rails/blob/main/actionpack/lib/action_dispatch/http/headers.rb/

Таким образом, заголовок у запроса к эндпоинту выглядит как USER:registered_user_id, а внутри приложения следует использовать обращение за заголовком, как request.headers['HTTP_USER'].

Второй случай «генерирования и перехвата ошибок» предназначен для реализации Railway-ориентированного программирования при возвращении маркетплейсами статусов, отличных от значения 200.

Генерация исключения происходит в классе HTTP-клиента. Пример реализации был рассмотрен в разделе I.6 /app/services/base_api.rb .

Перехват исключения для этого случая осуществляется в методе downloadable? у класса с реализацией фоновой задачи импорта данных.

/app/jobs/products/import_job.rb

module Products
  class ImportJob < ApplicationJob
     def downloadable?
      import(@mp_credential.marketplace.to_constant_with(DOWNLOADER_CLASS).new(@mp_credential))
      true
    rescue NameError => e
      # We are checking the code. It's fixable
      ErrorLogger.push_trace e
      false
    rescue MarketplaceAggregator::ApiAccessDeniedError => e
      ErrorLogger.push e
      # that's all
      false
    rescue MarketplaceAggregator::ApiBadRequestError => e
      ErrorLogger.push e
      # that's all
      false
    rescue MarketplaceAggregator::ApiLimitError => e
      # restart the task taking into account restrictions on limits
      Tasks::ReimportProducts.new(@is_client_queue, @mp_credential, e).call
      false
    rescue MarketplaceAggregator::ApiError => e
      ErrorLogger.push e
      # TODO: restart task after an hour (for example)
      false
    end
  end
end

В большинстве случаев, как видно из кода, прерывание процедуры импорта не предполагает автоматизированного способа решения проблемы.

Кроме Второго способа обработки исключений здесь мы также видим Третий способ «генерирования и перехвата ошибок с целью перезапуска задачи по импорту данных» на строке с rescue MarketplaceAggregator::ApiLimitError. Это тот редкий случай, когда правила маркетплейса позволяют очевидным способом автоматически разрешить проблему последнего отказа в отклике по достижению rate limits.

/app/business_logic/tasks/reimport_products.rb

module Tasks
  class ReimportProducts
    def call
      # 1. check the history of ReimportProducts calls
      # It should be less than three times
      analyse do
        # 2. cause direct data import
        Products::ImportJob.set(wait_until: 1.minute.from_now)
                           .perform_later(@is_client_queue, @mp_credential.id)
      end
    end

  end
end

Здесь стоит рассказать об одном нюансе, учтённом в настройках HTTP-клиента при обработке статусов HTTP-ответов от маркетплейсов. Известно, что ответ по REST API может приходить не только от Web-приложения, но и от Web-сервера. Это ошибки 500-й серии. Некоторые из них являются некритическими для удержания контакта на незначительном промежутке времени посредством повторения запроса без отключения соединения с реципиентом. Пример такой настройки представлен в коде HTTP-клиента:

/app/services/connection.rb

class Connection
  RETRY_OPTIONS = {
    max: 5,                     # Retry a failed request up to 5 times
    interval: 2,                # First retry after 2s
    backoff_factor: 2,          # Double the delay for each subsequent retry
    retry_statuses: [500, 502, 503, 504]
    # Retry only when we get a 500, 502, 503 or 504 response
  }.freeze

  def self.start
    @conn ||= Faraday.new do |config|
      config.request :retry, RETRY_OPTIONS
      config.adapter Faraday.default_adapter
    end
  end
end

В заключении этого раздела разберём Четвёртый способ обработки ошибок, основанный на генерации ошибок без последующей их программной обработки. Это ситуация называется «проектированием по контракту», когда в родительском классе заявляется пул методов, обязательных к реализации в дочерних классах. Чтобы не забывать прописывать реализации обязательных методов и создаются искусственные условия, генерирующие критические прерывания программы. Ещё такой подход в программировании называют Принципом Барбары-Лисков. Пример его применения можно видеть в /app/services/base_check_credentials.rb.

class BaseCheckCredentials
  
  private

  # curl command can invoke POST method with -I/--head option
  # https://curl.se/docs/manpage.html
  def bash_command
    raise NotImplementedError, "#{self.class}.#{__method__}: #{I18n.t('errors.marketplace_has_not_been_selected')}"
  end

  def http_client
    raise NotImplementedError, "#{self.class}.#{__method__}: #{I18n.t('errors.marketplace_has_not_been_selected')}"
  end
end

В наследуемых классах заявленные методы превращаются в следующее содержание

/app/services/yandex/check_credentials.rb

module Yandex
  class CheckCredentials < BaseCheckCredentials

    private

    def bash_command
      <<~`BASH`
        curl -I \
        -H 'Authorization: OAuth oauth_token=#{@mp_credential.credentials['token']}, \
        oauth_client_id=#{ENV.fetch('YANDEX_APP_ID')}' \
        -X POST https://api.partner.market.yandex.ru/businesses/\
        #{@mp_credential.credentials['business_id']}/offer-mappings.json?limit=1
      BASH
    end

    def http_client
      Yandex::Api::OfferMappings.new(@mp_credential)
    end
  end
end

/app/services/ozon/check_credentials.rb

module Ozon
  class CheckCredentials < BaseCheckCredentials

    private

    def bash_command
      <<~`BASH`
        curl -I \
        -H 'Api-Key: #{@mp_credential.credentials['api_key']}' \
        -H 'Client-Id: #{@mp_credential.credentials['client_id']}' \
        -H 'x-o3-app-name: #{ENV.fetch('OZON_APP_ID')}' \
        -X POST https://api-seller.ozon.ru/v2/product/list
      BASH
    end

    def http_client
      Ozon::Api::ProductList.new(@mp_credential)
    end
  end
end

3. Выстраивание задач по загрузке данных в очередь

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

У нас в качестве адаптера очередей используется Sidekiq, о чём и указано в конфигурационном файле /config/application.rb.

Поскольку Sidekiq является Ruby процессом, то он остаётся обладателем только одного ядра процессора на инстанс. Чтобы увеличить количество запущенных инстансов, нужно увеличить количество конфигурационных файлов в папке /config.

Для текущего приложения принято решение выделить два инстанса. Один на клиентские запросы по подключению новых аккаунтов и обслуживанию запросов на повторную, внеплановую перезагрузку данных с маркетплейсов /config/sidekiq_live.yml .

:timeout: 10
:concurrency: <%= ENV.fetch("SIDEKIQ_CONCURRENCY") { 15 } %>
:max_retries: 0
:queues:
  - [client_grabber_products, 5]
  - [client_grabber_orders, 3]
  - [client_grabber_transactions, 2]
  - [default, 1]
  - [client_grabber_product_descriptions, 4]

Второй инстанс выделяется на выполнение регулярной плановой загрузки данных по расписанию и их автоматическую догрузку в случае предусмотренных вариантов такой отложенной обработки /config/sidekiq_scheduled.yml .

:timeout: 10
:concurrency: <%= ENV.fetch("SIDEKIQ_CONCURRENCY") { 15 } %>
:max_retries: 0
:queues:
  - [marketplace_grabber_products, 5]
  - [marketplace_grabber_orders, 3]
  - [marketplace_grabber_transactions, 3]
  - [marketplace_grabber_product_descriptions, 4]

Для каждого инстанса выделяем некоторое число параллельных потоков concurrency, которое позволим одномоментно открыть для увеличения пропускной способности Sidekiq. Мы выбрали значение 15.

Далее распределим выделенное количество потоков между очередями. Здесь нужно учитывать, что чем больше потоков находится в параллельной работе, тем больше они друг другу мешают. Идеального распределения потоков между очередями не существует. В любом случае конфигурацию придётся настраивать под реальные объёмы обрабатываемых данных. Страница административной панели Sidekiq будет нам в помощь.

Теперь оценим процессы, происходящие внутри одного потока Sidekiq.

После ознакомления в данной статье с подробным описанием всех процессов импорта данных с маркетплейсов можно составить представления о расходуемых ресурсах. Одним существенным ресурсом является время загрузки данных по запросу о товарах. Для Яндекс.Маркет тормозящим фактором в имеющейся реализации может стать процесс «засыпания» на очень больших списках товаров в целях соблюдения наложенных маркетплейсом rate limits. На OZON замедляющим выполнение задачи фактором при загрузке данных может становиться загрузка описаний товаров. Для обоих маркетплейсов с течением продолжительной эксплуатации активно торгующим продавцом может оказаться избыточной операция загрузки архивных данных. Просто их может оказаться слишком много. На все эти случаи у нас уже интегрирован в код рукописный бенчмаркинг /lib/ma_benchmarking.rb .

module MaBenchmarking
  def benchmarking(log_string)
    back_time = Time.now
    yield
    Rails.logger.info format("(in %.3f sec) #{log_string.call}", Time.now - back_time)
  end
end

Он логирует события выгрузки данных. На основе логов мы можем проводить периодический анализ выгружаемых данных для последующего принятия организационных мер.

Следующим расходным ресурсом является количество обращений из параллельных потоков к базе данных на запись и обновление данных. На текущий момент в программе реализована наилучшая практика на чтение и запись данных из/в базу пакетами. Так для импорта используется гем activerecord-import. Предполагается соблюдать заданную практику при дальнейшем развитии программного решения.

Последний по упоминанию расходный ресурс, на который стоит обращать внимание, является объём выделяемой памяти на инстанс Sidekiq. Им мы можем управлять через изменение размеров батчей в пакетах данных на загрузку с маркетплейса и на обработку базы данных.

IV. Заключение

В статье детально разобран код Web-сервиса, эксплуатирующего REST API по импорту данных о товарах с маркетплейсов Яндекс.Маркет и OZON. На реальных участках кода показаны примеры реализации принципов SOLID. Проектируемое приложение не предполагает превращение Web-сервиса в большой монолит. Здесь намеренно отсутствует модуль авторизации пользователей. Этот функционал предполагает реализацию в отдельном сервисе. Архитектуру проектируемого приложения можно назвать сервисной. Описанные в документации к Web-сервису эндпоинты предназначены для внутреннего использования внутри приложения.

Представленный Web-сервис построен на фреймворке Ruby on Rails. В качестве СУБД используется PostgreSQL. Для хранения данных активно используются нативные типы данных Postgres. Выполнение фоновых задач вынесено в Sidekiq. В проекте используется системный менеджер задач CRON из гема clockwork. Для кэширования в Redis используется гем Kredis. С полностью рабочим проектом можно ознакомиться в репозитории https://github.com/rubygitflow/marketplace_aggregator.

Проект открыт для вкладов от контрибьютеров.

Приветствуется сохранение выбранного стиля в настройках Rubocop. Одобряется глубокое тестирование в районе 100% с сохранением покрытия кода тестами в Rspec.