Паттерн «Стратегия» в моей практике

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

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

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

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

☺ через запрос к PostgreSQL;
☺ через запрос к ClickHouse.

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

Именно здесь я на практике применил паттерн проектирования «Стратегия» и разобрался с ним.

В чём заключалась моя проблема

Можно было пойти самым простым путём и добавить обычное условие непосредственно в основной класс расчёта медиан:

if source == "postgres":
    prices = await fetch_prices_from_postgres(product_ids)
elif source == "clickhouse":
    prices = await fetch_prices_from_clickhouse(product_ids)
else:
    raise ValueError("Unknown source")

Когда у нас два источника, такой код ещё выглядит адекватно, но проблемы могут и будут появляться позже, когда начнут добавляться новые источники: CSV, JSON, API и т. д. Тогда основной класс расчёта начнёт постепенно обрастать условиями и знанием о каждом конкретном источнике данных. Он будет отвечать не только за расчёт медиан, но и за создание соединений, формирование запросов, особенности батчинга и закрытие клиентов разных баз данных. В результате нарушается принцип единственной ответственности: класс, который должен рассчитывать медианы, начинает одновременно управлять инфраструктурой получения цен.

Мне хотелось, чтобы основной алгоритм выглядел примерно так:

prices = await self.price_source.fetch_prices(product_ids)

То есть сервис просто просит источник вернуть цены, но не знает, каким образом тот это сделает.

Что такое паттерн «Стратегия»

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

Обе стратегии должны соблюдать одинаковый контракт:

async def fetch_prices(
    product_ids: list[str],
) -> dict[str, float]:
    ...

На вход поступает список идентификаторов товаров product_ids, а на выходе возвращается словарь с найденными ценами. Главный алгоритм не знает, работает ли внутри PostgreSQL, ClickHouse или что-то ещё. Для него важен только результат.

Общий контракт через Protocol

В Python для описания такого контракта удобно использовать Protocol:

from typing import Protocol


class PriceSource(Protocol):
    async def fetch_prices(
        self,
        product_ids: list[str],
    ) -> dict[str, float]:
        ...

Protocol описывает поведение, которое ожидается от объекта, и является хорошей альтернативой ABC при создании интерфейса.

Любой класс, в котором есть метод fetch_prices с подходящей сигнатурой, может использоваться как стратегия. Ему даже не обязательно явно наследоваться от PriceSource. Это хорошо сочетается с идеей структурной типизации в Python: если объект ведёт себя как нужный источник цен, значит его можно использовать как источник цен.

Стратегия получения цен из PostgreSQL

Первая реализация получает цены из основной PostgreSQL-базы:

import psycopg


class PostgresPriceSource:
    def __init__(
        self,
        conn_params: dict,
    ):
        self.conn_params = conn_params

    async def fetch_prices(
        self,
        product_ids: list[str],
    ) -> dict[str, float]:
        if not product_ids:
            return {}

        conn = await psycopg.AsyncConnection.connect(
            **self.conn_params,
        )

        try:
            async with conn.cursor() as cursor:
                await cursor.execute(
                    """
                    КАКОЙ-ТО SQL-СКРИПТ (секретик)
                    """,
                    {"product_ids": product_ids},
                )

                rows = await cursor.fetchall()
        finally:
            await conn.close()

        return {
            str(product_id): float(price)
            for product_id, price in rows
        }

Здесь стратегия отвечает за всё, что связано именно с PostgreSQL:

1. создание подключения; 2. выполнение SQL-запроса; 3. выбор цены со скидкой или обычной цены; 4. фильтрацию устаревших записей; 5. преобразование результата в общий формат.

После выполнения метода остальная система получает обычный словарь:

{
    "product-1": 119.90,
    "product-2": 240.00,
    "product-3": 87.50,
}

Стратегия получения цен из ClickHouse

Вторая стратегия решает ту же задачу, но использует ClickHouse:

import clickhouse_connect


class ClickHousePriceSource:
    def __init__(
        self,
        conn_params: dict,
        table_name: str,
    ):
        self.conn_params = conn_params
        self.table_name = table_name

    async def fetch_prices(
        self,
        product_ids: list[str],
    ) -> dict[str, float]:
        if not product_ids:
            return {}

        client = await clickhouse_connect.get_async_client(
            host=self.conn_params["host"],
            port=self.conn_params["port"],
            username=self.conn_params["user"],
            password=self.conn_params["password"],
            database=self.conn_params["database"],
        )

        query = f"""
            КАКОЙ-ТО SQL-СКРИПТ (секретик)
        """

        try:
            result = await client.query(
                query,
                parameters={"ids": product_ids},
            )
        finally:
            await client.close()

        return {
            str(product_id): float(price)
            for product_id, price in result.result_rows
        }

Здесь уже используется другая модель хранения данных.

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

Несмотря на то что внутри стратегий используются разные базы, разные клиенты и разные SQL-диалекты, снаружи контракт остаётся прежним:

dict[str, float]

Именно это и делает стратегии взаимозаменяемыми.

Выбор стратегии через фабрику

Чтобы основной класс не создавал конкретные реализации самостоятельно, я вынес выбор источника в отдельную функцию:

def build_price_source(
    source: str | PriceSource | None,
) -> PriceSource:
    if source is None:
        return PostgresPriceSource(
            conn_params=POSTGRES_CONN_PARAMS,
        )

    if not isinstance(source, str):
        return source

    if source == "postgres":
        return PostgresPriceSource(
            conn_params=POSTGRES_CONN_PARAMS,
        )

    if source == "clickhouse":
        return ClickHousePriceSource(
            conn_params=CLICKHOUSE_CONN_PARAMS,
            table_name="price_history.products",
        )

    raise ValueError(
        "Unknown price source: "
        "expected 'postgres', 'clickhouse' "
        "or an object implementing PriceSource"
    )

Здесь уже используется не только «Стратегия», но и небольшая фабрика, которая отвечает за создание нужной реализации.

И тут небольшая деталь в различиях между этими паттернами:

☺ стратегия описывает взаимозаменяемые алгоритмы;
☺ фабрика выбирает и создаёт конкретную стратегию.

*P. S. Фабрику я постараюсь разобрать в отдельном посте.*

По умолчанию используется PostgreSQL, но источник можно изменить через конфигурацию:

price_source = build_price_source("clickhouse")

Также можно передать уже готовый объект:

custom_source = SomeExternalApiPriceSource()
price_source = build_price_source(custom_source)

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

Использование стратегии в основном алгоритме

В конструкторе нашего основного алгоритма медиан источник создаётся один раз во время инициализации:

class MedianCalculator:
    def __init__(
        self,
        price_source: str | PriceSource | None = None,
    ):
        self.price_source = build_price_source(
            price_source,
        )

Дальше основной алгоритм работает только через общий интерфейс:

class MedianCalculator:
    async def process_batch(
        self,
        product_ids: list[str],
    ) -> dict[str, float]:
        prices = await self.price_source.fetch_prices(
            product_ids,
        )

        return prices

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

Получается примерно такая схема:

PostgreSQL ─────┐
                │
                ▼
         PriceSource
                │
                ▼
       MedianCalculator
                │
                ▼
    дедупликация и медиана
                │
                ▼
        сохранение результата
                ▲
                │
ClickHouse ─────┘

Главный класс знает только про абстракцию PriceSource, а конкретная реализация подставляется снаружи.

Почему я не стал использовать обычный if внутри основного класса MedianCalculator?

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

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

Если бы работа с источниками находилась внутри MedianCalculator, этому классу пришлось бы знать:

1. как подключаться к PostgreSQL; 2. как подключаться к ClickHouse; 3. какой SQL используется в каждой базе; 4. как закрывать соединения; 5. как выбирать последнее значение цены; 6. как обрабатывать особенности каждого драйвера.

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

С использованием стратегии обязанности разделились естественным образом:

MedianCalculator
    отвечает за расчёт медиан

PostgresPriceSource
    отвечает за получение цен из PostgreSQL

ClickHousePriceSource
    отвечает за получение цен из ClickHouse

build_price_source
    отвечает за выбор реализации

Тестирование

Ещё одно важное преимущество, на мой взгляд, — возможность тестировать алгоритм без настоящих баз данных.

Для этого достаточно создать тестовую стратегию:

class FakePriceSource:
    async def fetch_prices(
        self,
        product_ids: list[str],
    ) -> dict[str, float]:
        return {
            product_id: 100.0
            for product_id in product_ids
        }

После этого её можно передать в калькулятор:

calculator = MedianCalculator(
    price_source=FakePriceSource(),
)

Основной алгоритм будет работать так же, как с PostgreSQL или ClickHouse, но тест не потребует:

☺ поднимать контейнер с базой;
☺ создавать таблицы;
☺ заполнять данные;
☺ ждать выполнения реальных запросов.

Можно проверить именно расчётную логику:

async def test_median_calculation():
    calculator = MedianCalculator(
        price_source=FakePriceSource(),
    )

    result = await calculator.process_batch(
        ["1", "2", "3"],
    )

    assert result == {
        "1": 100.0,
        "2": 100.0,
        "3": 100.0,
    }

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

Когда «Стратегия» действительно полезна?

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

Но стратегия хорошо подходит, когда:

☺ существует несколько вариантов одного алгоритма;
☺ реализации должны быть взаимозаменяемыми;
☺ детали реализации не должны проникать в бизнес-логику;
☺ нужно удобно подменять зависимости в тестах;
☺ ожидается появление новых вариантов в будущем.

В моём случае эти условия совпали почти полностью.

Что в итоге получилось?

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

Он получает источник цен через конструктор:

self.price_source = build_price_source(
    price_source,
)

А во время расчёта вызывает единый метод:

prices = await self.price_source.fetch_prices(
    product_ids,
)

За этой строкой может находиться PostgreSQL, ClickHouse, тестовая заглушка или будущий внешний сервис. Для алгоритма расчёта медиан это больше не имеет значения.

Наверное, именно за это мне и нравится паттерн «Стратегия». Он не делает сам алгоритм умнее и не решает бизнес-задачу вместо разработчика. Он просто помогает провести понятную границу между тем, что система делает, и тем, каким способом она это делает.

В моём случае система рассчитывает медианные цены, а PostgreSQL и ClickHouse являются лишь разными реализациями одного интерфейса.

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