Что такое инструменты функционального программирования в collections?

Инструменты функционального программирования в collections

collections — это модуль Python, который предоставляет специализированные контейнеры данных, а также взаимодействует с функциональным подходом к программированию через итерирование, трансформацию и комбинирование коллекций. Хотя инструменты функционального программирования (map, filter, reduce) находятся в других модулях, collections тесно интегрируется с функциональным стилем.

Основные типы данных в collections

from collections import (
    namedtuple,      # Неизменяемые кортежи с именованными полями
    defaultdict,     # Словарь с дефолтными значениями
    Counter,         # Счётчик элементов
    deque,           # Двусторонняя очередь
    OrderedDict      # Упорядоченный словарь (Python 3.7+ встроен)
)

1. namedtuple

from collections import namedtuple
from functools import reduce

# Создать именованный кортеж
Point = namedtuple('Point', ['x', 'y'])
p1 = Point(1, 2)
p2 = Point(3, 4)
p3 = Point(5, 6)

print(p1.x, p1.y)  # 1 2
print(p1[0])       # 1 (доступ как к кортежу)

# Функциональный подход: применить функцию к каждому point
points = [p1, p2, p3]

# Map: преобразовать в расстояния
def distance_from_origin(point):
    return (point.x ** 2 + point.y ** 2) ** 0.5

distances = list(map(distance_from_origin, points))
print(distances)  # [2.236..., 5.0, 7.810...]

# Filter: оставить точки с x > 2
filtered = list(filter(lambda p: p.x > 2, points))
print(filtered)  # [Point(x=3, y=4), Point(x=5, y=6)]

# Reduce: найти сумму всех координат
total_sum = reduce(lambda acc, p: acc + p.x + p.y, points, 0)
print(total_sum)  # 21

2. defaultdict с функциональным подходом

from collections import defaultdict
from functools import reduce

# Группировка данных функционально
students = [
    ('Alice', 'Math'),
    ('Bob', 'Math'),
    ('Charlie', 'English'),
    ('David', 'English'),
    ('Eve', 'Math')
]

# Способ 1: defaultdict с list
groups = defaultdict(list)
for name, subject in students:
    groups[subject].append(name)

print(dict(groups))
# {'Math': ['Alice', 'Bob', 'Eve'], 'English': ['Charlie', 'David']}

# Способ 2: Функциональный подход с reduce
def group_by(data, key_func):
    return reduce(
        lambda acc, item: (
            acc.update({key_func(item): acc.get(key_func(item), []) + [item]}) or acc
        ),
        data,
        {}
    )

result = group_by(students, lambda x: x[1])
print(result)

3. Counter для анализа данных

from collections import Counter
from functools import reduce

# Подсчёт элементов
text = "hello world hello python"
words = text.split()

word_counts = Counter(words)
print(word_counts)  # Counter({'hello': 2, 'world': 1, 'python': 1})

# Функциональный подход: фильтрация по частоте
frequent_words = list(filter(lambda x: x[1] > 1, word_counts.items()))
print(frequent_words)  # [('hello', 2)]

# Map: получить только слова
words_only = list(map(lambda x: x[0], word_counts.items()))
print(words_only)  # ['hello', 'world', 'python']

# Reduce: найти максимальную частоту
max_frequency = reduce(
    lambda max_count, item: max(max_count, item[1]),
    word_counts.items(),
    0
)
print(max_frequency)  # 2

# most_common: встроенный метод
print(word_counts.most_common(2))  # [('hello', 2), ('world', 1)]

4. deque для функционального обхода

from collections import deque
from functools import reduce

# Двусторонняя очередь
dq = deque([1, 2, 3, 4, 5])

# Функциональное преобразование
squared = list(map(lambda x: x ** 2, dq))
print(squared)  # [1, 4, 9, 16, 25]

# Filter: чётные числа
evens = list(filter(lambda x: x % 2 == 0, dq))
print(evens)  # [2, 4]

# Reduce: произведение
product = reduce(lambda a, b: a * b, dq)
print(product)  # 120

# Ротация
dq.rotate(2)  # Сдвиг вправо на 2
print(list(dq))  # [4, 5, 1, 2, 3]

5. ChainMap для функциональной работы со словарями

from collections import ChainMap

# Объединение словарей
dict1 = {'a': 1, 'b': 2}
dict2 = {'c': 3, 'd': 4}
dict3 = {'e': 5}

chained = ChainMap(dict1, dict2, dict3)

# Функциональное преобразование
transformed = {k: v * 2 for k, v in chained.items()}
print(transformed)  # {'a': 2, 'b': 4, 'c': 6, 'd': 8, 'e': 10}

# Map на keys и values
keys = list(map(str.upper, chained.keys()))
print(keys)  # ['A', 'B', 'C', 'D', 'E']

6. Комбинирование с functools

from collections import Counter, defaultdict
from functools import reduce, partial

# Анализ текста: найти top-3 буквы
text = "the quick brown fox"
letters = Counter(text.replace(' ', ''))

# Через reduce
top_letters = dict(reduce(
    lambda acc, item: acc + [item] if len(acc) < 3 else acc,
    sorted(letters.items(), key=lambda x: -x[1]),
    []
))
print(top_letters)  # Самые частые буквы

# Через встроенный метод
print(dict(letters.most_common(3)))

7. Пайплайн обработки данных

from collections import namedtuple, Counter, defaultdict
from functools import reduce

# Определение данных
Transaction = namedtuple('Transaction', ['id', 'amount', 'category'])

transactions = [
    Transaction(1, 100, 'food'),
    Transaction(2, 50, 'transport'),
    Transaction(3, 200, 'food'),
    Transaction(4, 75, 'entertainment'),
    Transaction(5, 150, 'food'),
]

# Pipeline: фильтрация → группировка → агрегирование

# 1. Filter: только суммы > 75
expensive = list(filter(lambda t: t.amount > 75, transactions))

# 2. Map: преобразовать в (категория, сумма)
by_category = list(map(lambda t: (t.category, t.amount), expensive))

# 3. Reduce: сумма по категориям
totals = reduce(
    lambda acc, item: {
        **acc,
        item[0]: acc.get(item[0], 0) + item[1]
    },
    by_category,
    {}
)

print(totals)  # {'food': 350, 'entertainment': 75}

# Альтернатива с defaultdict
totals_dd = defaultdict(int)
for category, amount in by_category:
    totals_dd[category] += amount
print(dict(totals_dd))

8. Практический пример: обработка логов

from collections import Counter, namedtuple
from functools import reduce

LogEntry = namedtuple('LogEntry', ['timestamp', 'level', 'message'])

logs = [
    LogEntry('10:00', 'INFO', 'User logged in'),
    LogEntry('10:05', 'ERROR', 'Database connection failed'),
    LogEntry('10:10', 'INFO', 'Cache cleared'),
    LogEntry('10:15', 'ERROR', 'Invalid request'),
    LogEntry('10:20', 'INFO', 'Report generated'),
]

# 1. Подсчитать ошибки
errors = list(filter(lambda log: log.level == 'ERROR', logs))
print(f"Total errors: {len(errors)}")

# 2. Группировать по уровню
from collections import defaultdict

grouped = defaultdict(list)
for log in logs:
    grouped[log.level].append(log.message)

print(dict(grouped))

# 3. Map: извлечь только сообщения ошибок
error_messages = list(map(lambda log: log.message, errors))
print(error_messages)

# 4. Reduce: построить статистику
stats = reduce(
    lambda acc, log: {**acc, log.level: acc.get(log.level, 0) + 1},
    logs,
    {}
)
print(stats)  # {'INFO': 3, 'ERROR': 2}

Преимущества функционального подхода с collections

1. Иммутабельность — namedtuple неизменяемы
2. Composability — легко комбинировать функции
3. Readability — декларативный стиль
4. Testability — чистые функции легче тестировать
5. Performance — оптимизация за счёт ленивых вычислений

Вывод: collections + функциональное программирование = мощный инструмент для обработки данных в Python, позволяющий писать чистый, выразительный и надёжный код.