Написать мне
Статьи

Python Pipes: Прощайте, вложенные map() и filter()! Как писать чистый и читаемый код

2025-09-03 12:12 Синтаксис Python

Функции map и filter — удобные инструменты для работы с итерируемыми объектами в Python. Но как только вам нужно применить их вместе, код начинает напоминать матрешку из скобок и lambda-функций, читать которую — то еще удовольствие.

Вот классический пример, от которого у многих дергается глаз:

arr = [1, 2, 3, 4, 5, 6, 7, 8]

# Задача: выбрать четные числа и умножить их на 2
result = list(map(lambda x: x * 2, filter(lambda x: x % 2 == 0, arr)))

print(result)
# [4, 8, 12, 16]

Мы читаем этот код изнутри наружу. Сначала filter, потом map, потом list. Для простой операции — слишком много ментальных усилий. А что, если операций будет три? Четыре? Ад из скобок гарантирован.

А теперь представьте, что тот же самый код можно написать вот так:

from pipe import select, where

arr = [1, 2, 3, 4, 5, 6, 7, 8]

result = list(arr
              | where(lambda x: x % 2 == 0)
              | select(lambda x: x * 2)
             )

print(result)
# [4, 8, 12, 16]

Логика читается слева направо, сверху вниз. Взять arr, затем отфильтровать, затем преобразовать. Чисто, линейно и интуитивно понятно, как конвейер на заводе.

Именно эту элегантность и привносит в Python библиотека pipe.

Что такое Pipe и в чем его магия?

Pipe — это небольшая, но очень полезная библиотека, которая позволяет использовать оператор | (пайп) для построения цепочек обработки данных. Идея позаимствована из командной строки Unix/Linux, где вывод одной команды становится вводом для другой.

Это превращает вложенные вызовы функций в последовательный, линейный конвейер. Вместо того чтобы думать "что во что вложено", вы думаете "что происходит за чем".

Для установки достаточно одной команды:

pip install pipe

Давайте разберем основные "детали" этого конвейера, которые покроют 90% ваших задач по обработке данных.

Основы: where и select — фильтрация и преобразование

Два самых частых действия с любой коллекцией — это отфильтровать ненужное и преобразовать оставшееся. В pipe за это отвечают where и select.

where — фильтруем как в SQL

Метод where — это прямой аналог встроенной функции filter. Он пропускает дальше по конвейеру только те элементы, для которых переданная функция (условие) возвращает True.

from pipe import where

arr = [1, 2, 3, 4, 5, 6]

# Оставляем только четные числа
even_numbers = list(arr | where(lambda x: x % 2 == 0))

print(even_numbers)
# [2, 4, 6]

Название выбрано не случайно и будет знакомо всем, кто хоть раз писал SQL-запросы. Это делает код еще более семантически понятным.

select — преобразуем каждый элемент

Метод select — это аналог map. Он применяет указанную функцию к каждому элементу, который дошел до него по конвейеру, и передает дальше результат ее выполнения.

from pipe import select

arr = [1, 2, 3, 4, 5]

# Умножаем каждый элемент на 2
doubled = list(arr | select(lambda x: x * 2))

print(doubled)
# [2, 4, 6, 8, 10]

Опять же, название select — реверанс в сторону SQL, где оно используется для выбора и преобразования полей.

[!TIP] Зачем это нужно, если есть map и filter? Магия pipe раскрывается именно при комбинации операций. Вместо вложенности мы получаем элегантную цепочку. Код становится декларативным — мы описываем что мы хотим получить, а не как это сделать шаг за шагом. Это снижает когнитивную нагрузку и упрощает поддержку кода.

Работа с вложенными структурами: chain и traverse

Часто данные приходят к нам в виде списков списков или других запутанных структур. Pipe предлагает два мощных инструмента для их "разглаживания".

chain — разглаживаем на один уровень

Представьте, что у вас есть список, содержащий другие списки, и вам нужно получить один плоский список. Метод chain решает именно эту задачу. Он "распаковывает" каждый элемент-итератор на один уровень вложенности.

from pipe import chain

nested_list = [[1, 2], [3, 4], [5, 6]]

flat_list = list(nested_list | chain)

print(flat_list)
# [1, 2, 3, 4, 5, 6]

[!WARNING] Ограничение chain chain работает только на один уровень вглубь. Если у вас более сложная вложенность, он не справится.

deeply_nested = [[1, 2, [3]], [4, 5]]
print(list(deeply_nested | chain))
# [1, 2, [3], 4, 5]  <- [3] остался вложенным

traverse — рекурсивное разглаживание до основания

А вот traverse — это тяжелая артиллерия. Он рекурсивно обходит всю структуру данных и извлекает все неитерируемые элементы, создавая из них один абсолютно плоский список.

from pipe import traverse

deeply_nested = [[1, 2, [3, [4, 5]]], [6, 7]]

super_flat_list = list(deeply_nested | traverse)

print(super_flat_list)
# [1, 2, 3, 4, 5, 6, 7]

traverse особенно полезен, когда вы работаете с JSON-ответами от API, где уровень вложенности может быть непредсказуемым. Давайте объединим его с select, чтобы извлечь все цены из списка словарей:

from pipe import select, traverse

products = [
    {"name": "Яблоко", "prices": [100, 110]},
    {"name": "Апельсин", "prices": 80},
    {"name": "Банан", "prices": [50, 55, [60]]},
]

all_prices = list(products
                  | select(lambda product: product["prices"])
                  | traverse
                 )

print(all_prices)
# [100, 110, 80, 50, 55, 60]

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

Группировка и агрегация: groupby

Часто возникает задача не просто отфильтровать или преобразовать данные, а сгруппировать их по какому-то признаку. Например, разделить товары по категориям или студентов по оценкам. Для этого в pipe есть метод groupby, работающий по аналогии с itertools.groupby.

Он принимает функцию, которая для каждого элемента вычисляет "ключ группировки". На выходе groupby создает итератор, где каждый элемент — это кортеж (ключ, <объекты_группы>).

Давайте сгруппируем числа на четные и нечетные:

from pipe import groupby, select

arr = [1, 2, 3, 4, 5, 6]

# Шаг 1: Группируем
grouped = arr | groupby(lambda x: "Even" if x % 2 == 0 else "Odd")
# На этом этапе grouped - это итератор. Если его превратить в список,
# он будет выглядеть примерно так:
# [('Odd', <iterator>), ('Even', <iterator>), ('Odd', <iterator>), ...]
# Важно: groupby группирует только ПОСЛЕДОВАТЕЛЬНЫЕ одинаковые ключи.
# Для корректной работы данные лучше предварительно отсортировать,
# если это необходимо. В нашем простом случае порядок не важен.

# Шаг 2: Преобразуем группы в удобный формат
result = list(grouped | select(lambda group: {group[0]: list(group[1])}))

print(result)
# [{'Odd': [1]}, {'Even': [2]}, {'Odd': [3]}, {'Even': [4]}, {'Odd': [5]}, {'Even': [6]}]

[!NOTE] Особенность groupby Как и его тезка из itertools, pipe.groupby группирует только последовательные элементы с одинаковым ключом. Если вам нужна группировка в стиле SQL, где все элементы с одним ключом попадают в одну группу независимо от их положения, сначала отсортируйте данные по тому же ключу. list(sorted(data, key=...) | groupby(...))

А теперь магия: мы можем применять пайпы прямо внутри пайпов! Давайте сгруппируем числа, а затем в каждой группе оставим только те, что больше 3.

from pipe import groupby, select, where

arr = [1, 2, 3, 4, 5, 6]

result = list(arr
    # Группируем на четные/нечетные
    | groupby(lambda x: "Even" if x % 2 == 0 else "Odd")
    # Преобразуем каждую группу
    | select(lambda group: {
          group[0]: list(
              # А вот и вложенный пайп!
              group[1] | where(lambda x: x > 3)
          )
      })
    # Отфильтруем пустые группы, если они образовались
    | where(lambda d: list(d.values())[0])
)

print(result)
# [{'Even': [4, 6]}, {'Odd': [5]}]

Такая композиция операций выглядит невероятно чисто и логично по сравнению с тем, как бы мы писали это с помощью стандартных циклов или comprehensions.

dedup — умное удаление дубликатов

Все знают, что для получения уникальных элементов из списка можно использовать set(): list(set(my_list)). Но у этого подхода есть два недостатка:

  1. Он не сохраняет исходный порядок элементов.
  2. Он не позволяет задать сложный критерий уникальности (например, считать объекты уникальными по значению одного из полей).

Метод dedup из pipe решает обе проблемы.

В простейшем виде он просто удаляет дубликаты, сохраняя порядок первого вхождения:

from pipe import dedup

arr = [1, 2, 1, 3, 4, 2, 2, 5, 3]

unique_ordered = list(arr | dedup)

print(unique_ordered)
# [1, 2, 3, 4, 5]

Но вся его сила — в необязательном параметре key. Он принимает функцию, которая вычисляет значение для проверки на уникальность. dedup пропустит дальше только первый элемент для каждого уникального значения, которое вернет эта key-функция.

Например, оставим только по одному числу из двух групп: "меньше 5" и "больше или равно 5".

from pipe import dedup

arr = [1, 8, 2, 9, 3, 6, 4, 7]

# Ключ: True для x < 5, False для x >= 5
result = list(arr | dedup(key=lambda x: x < 5))

print(result)
# [1, 8]

Как это сработало?

  1. Приходит 1. key(1) возвращает True. Такого ключа еще не было. Пропускаем 1.
  2. Приходит 8. key(8) возвращает False. Такого ключа еще не было. Пропускаем 8.
  3. Приходит 2. key(2) возвращает True. Этот ключ (True) уже встречался. Игнорируем 2.
  4. Приходит 9. key(9) возвращает False. Этот ключ (False) уже встречался. Игнорируем 9.
  5. ...и так далее.

Финальный аккорд: собираем все вместе

Давайте решим практическую задачу. У нас есть список словарей с данными о продуктах. Нам нужно:

  1. Убрать дубликаты по названию продукта, оставив первое вхождение.
  2. Из полученного списка извлечь количество (count).
  3. Оставить только те количества, которые являются целыми числами (отбросить None и прочий мусор).
from pipe import select, where, dedup

data = [
    {"name": "Яблоко", "count": 20},
    {"name": "Апельсин", "count": 40},
    {"name": "Грейпфрут", "count": None},
    {"name": "Яблоко", "count": 50},  # Дубликат по 'name'
    {"name": "Апельсин", "count": 70}, # Дубликат по 'name'
]

result = list(data
    # 1. Удаляем дубликаты по ключу 'name'
    | dedup(key=lambda fruit: fruit["name"])
    # 2. Извлекаем значение поля 'count'
    | select(lambda fruit: fruit["count"])
    # 3. Фильтруем, оставляя только целые числа
    | where(lambda count: isinstance(count, int))
)

print(result)
# [20, 40]

Попробуйте мысленно написать то же самое без pipe. Скорее всего, это будет либо вложенный list comprehension, который сложно читать, либо цикл for с дополнительным set для отслеживания увиденных имен. Вариант с pipe — самый декларативный и читаемый.

Вердикт: когда стоит использовать Pipe?

Библиотека pipe — это не замена стандартным инструментам Python, а их элегантное дополнение. Она будет полезна в задачах, связанных с обработкой и трансформацией данных, когда вам нужно выстроить конвейер из нескольких последовательных шагов.

Используйте pipe, когда:

  • Вы ловите себя на написании вложенных map, filter и list comprehensions.
  • Вам важна линейность и читаемость кода, описывающего трансформацию данных.
  • Вы выполняете цепочку из 2-3 и более операций над итерируемым объектом.

Не стоит использовать ее для единственной операции (list(arr | select(f)) — это просто более длинный [f(x) for x in arr]).

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