2025, Nov 08 06:03
Как типизировать фабрику алгоритмов в Python: list против Protocol и перегрузки
Разбираем, почему list инвариантен, как работают ковариантность и контравариантность, два практичных решения: перегрузки с Literal и читающий Protocol в Python.
Когда вы создаёте обобщённый фреймворк, который связывает модели данных и алгоритмы, одна из первых точек трения — корректно аннотировать фабричный метод, выбирающий подходящий набор экземпляров алгоритмов по значению из конфигурации. Интерфейс у всех общий, но конкретные данные и алгоритмы зависят от задачи. Задумка проста: вернуть коллекцию алгоритмов, которые все работают с единым форматом данных. На практике правила типизации вокруг ковариантности/контравариантности делают это удивительно непростым.
Пример, который поднимает вопрос типизации
Следующий набросок показывает общий протокол для данных и алгоритмов, две конкретные структуры данных (payload) и три реализации алгоритмов. Фабрика ветвится по ключу, известному во время выполнения, и возвращает список алгоритмов, настроенных под конкретный payload. Единственный открытый вопрос — как аннотировать возвращаемое значение фабрики.
from dataclasses import dataclass
from typing import Protocol, TypeVar
# Протоколы
class CoreData(Protocol):
common: int
D = TypeVar("D", bound=CoreData)
class AlgoBase(Protocol[D]):
def update(self, data: D) -> None: ...
# Реализации данных
@dataclass
class PayloadOne:
common: int
extra: int
@dataclass
class PayloadTwo:
common: int
extra: str
# Реализации алгоритмов
class ProcOne:
def update(self, data: PayloadOne) -> None:
data.extra += data.common
class ProcTwoA:
def update(self, data: PayloadTwo) -> None:
data.extra *= data.common
class ProcTwoB:
def update(self, data: PayloadTwo) -> None:
data.extra += "2b"
# Фабрика, возвращающая разные списки в зависимости от ключа времени выполнения
class AlgoFactory:
def _build_one(self) -> list[AlgoBase[PayloadOne]]:
return [ProcOne()]
def _build_two(self) -> list[AlgoBase[PayloadTwo]]:
return [ProcTwoA(), ProcTwoB()]
def make(self, kind: int): # Какую аннотацию возвращаемого значения указать?
match kind:
case 1:
return self._build_one()
case 2:
return self._build_two()
case _:
raise ValueError(f"Unknown type of data {kind}")
Почему list[AlgoBase[CoreData]] выглядит уместно, но не работает
Цель — «коллекция алгоритмов, работающих с одним согласованным типом данных». Возникает соблазн объявить единый зонтичный тип вроде list[AlgoBase[CoreData]]. Но его нельзя принять — и на то есть причины. Параметр типа у list инвариантен: list[X] не является ни подтипом, ни супертипом list[Y], даже если X и Y связаны. При этом протокол алгоритма принимает свой параметр типа в методе update, а значит, он должен быть контравариантным по этому параметру. Однако даже если пометить протокол алгоритма как контравариантный, внешняя обёртка list сохраняет инвариантность всей конструкции. В итоге list[AlgoBase[PayloadOne]] не является подтипом list[AlgoBase[CoreData]], и статическая проверка справедливо отказывается сводить их к общему супертипу.
Есть ещё один момент, который стоит проговорить явно. Объединение внутри списка — это не то же самое, что объединение списков. Иными словами, list[A | B] не эквивалентно list[A] | list[B]. Для сигнатуры фабрики это важно: нельзя «схлопнуть» разные конкретные списки в один список над объединением и ожидать тех же результатов от типизации.
Два практичных способа типизировать фабрику
Первый вариант — использовать перегрузки, привязанные к Literal-значениям. Это распространённый и точный способ зафиксировать развилки фабрики. Там, где в вызове передаётся известный литерал, возвращаемый тип будет определён точно, а реализация возвращает объединение, покрывающее случаи.
from typing import overload, Literal, TypeVar, Protocol
class CoreData(Protocol):
common: int
D = TypeVar("D", bound=CoreData, contravariant=True)
class AlgoBase(Protocol[D]):
def update(self, data: D) -> None: ...
@dataclass
class PayloadOne:
common: int
extra: int
@dataclass
class PayloadTwo:
common: int
extra: str
class ProcOne:
def update(self, data: PayloadOne) -> None:
data.extra += data.common
class ProcTwoA:
def update(self, data: PayloadTwo) -> None:
data.extra *= data.common
class ProcTwoB:
def update(self, data: PayloadTwo) -> None:
data.extra += "2b"
GroupOne = list[AlgoBase[PayloadOne]]
GroupTwo = list[AlgoBase[PayloadTwo]]
class AlgoFactory:
def _build_one(self) -> GroupOne:
return [ProcOne()]
def _build_two(self) -> GroupTwo:
return [ProcTwoA(), ProcTwoB()]
@overload
def make(self, kind: Literal[1]) -> GroupOne: ...
@overload
def make(self, kind: Literal[2]) -> GroupTwo: ...
def make(self, kind: int) -> GroupOne | GroupTwo:
if kind == 1:
return self._build_one()
if kind == 2:
return self._build_two()
raise ValueError(f"Unknown type of data {kind}")
Второй вариант — вернуть неизменяемое, только для чтения представление через Protocol, который предоставляет лишь итерацию. Так мы обходим инвариантность list, переходя к ковариантному контейнеру вроде tuple и явно фиксируя, что потребителям достаточно уметь перебирать элементы и вызывать update. В протоколе объявляется только __iter__, а возвращать можно кортеж, собранный из списка, который строит каждая ветка.
from dataclasses import dataclass
from typing import Protocol, TypeVar, Iterator
class CoreData(Protocol):
common: int
D = TypeVar("D", bound=CoreData, contravariant=True)
class AlgoBase(Protocol[D]):
def update(self, data: D) -> None: ...
@dataclass
class PayloadOne:
common: int
extra: int
@dataclass
class PayloadTwo:
common: int
extra: str
class ProcOne:
def update(self, data: PayloadOne) -> None:
data.extra += data.common
class ProcTwoA:
def update(self, data: PayloadTwo) -> None:
data.extra *= data.common
class ProcTwoB:
def update(self, data: PayloadTwo) -> None:
data.extra += "2b"
class AlgoFactory:
def _build_one(self) -> list[AlgoBase[PayloadOne]]:
return [ProcOne()]
def _build_two(self) -> list[AlgoBase[PayloadTwo]]:
return [ProcTwoA(), ProcTwoB()]
class Batch(Protocol):
def __iter__(self) -> Iterator[AlgoBase[CoreData]]: ...
class AlgoFactory:
def _build_one(self) -> list[AlgoBase[PayloadOne]]:
return [ProcOne()]
def _build_two(self) -> list[AlgoBase[PayloadTwo]]:
return [ProcTwoA(), ProcTwoB()]
def make(self, kind: int) -> Batch:
match kind:
case 1:
return tuple(self._build_one())
case 2:
return tuple(self._build_two())
case _:
raise ValueError(f"Unknown type of data {kind}")
Ключевая мысль: клиентам не нужна мутация возвращаемой коллекции. Оставляя только итерацию и отдавая неизменяемую последовательность, вы обходите инвариантность list и держите API сфокусированным на реальном сценарии использования — пройтись по элементам и вызвать update. Что до протокола — в нём достаточно объявить лишь __iter__; больше ничего в теле протокола не требуется.
Почему это важно для реальных проектов
По мере масштабирования такого фреймворка число payload’ов и алгоритмов быстро растёт. Тип вроде list[AlgoBase[CoreData]] поначалу выглядит удобным для сопровождения, но незаметно подрывает те статические гарантии, которых вы ждёте от типизатора. Точная типизация через перегрузки даёт местам вызова конкретные типы, когда ключ известен, а «читающий» протокол ясно выражает намерение и предотвращает случайные изменения списка. Важно также помнить: инвариантность списков и правила распределения объединений легко применить неверно. Чёткая формулировка этих ограничений экономит время по мере эволюции кодовой базы.
Итоги
Если вызывающая сторона знает ключ фабрики на этапе компиляции, отдавайте предпочтение перегрузкам с Literal и возвращайте из реализации объединение конкретных типов списков. Это сохраняет точность типизации, не пытаясь навязать общий «зонтичный» тип, который не сойдётся. Если же от клиентов требуется только итерация, переходите к неизменяемому, ковариантному представлению и публикуйте протокол с единственным методом итерации; возврат кортежа алгоритмов органично вписывается в такой контракт. В обоих вариантах алгоритмы в каждой ветке работают с единым согласованным типом данных — именно это и было исходной целью дизайна.