$\lambda$FP в Python


fedor1113

Функционалность. «Теория суха…» — за что мы её и любим

«All race conditions,
deadlock conditions,
and concurrent update problems
are due to mutable variables.»
― Robert C. Martin, Clean Architecture
*на год моложе Fortran-а*

Чистота

«Разорвал я лучистые нити,
Обручавшие мне красоту; —
Братья, сестры, скажите, скажите,
Где мне вновь обрести чистоту?»
— Николай Гумилёв, 1906

Чистота


Чистой функцией называют функцию в ныне стандартном математическом понимании этого слова — это определённый закон, ставящий вещи в соответствие. Таким образом, мы возвращаемся к истокам, очищаем наше понимание функции от лишних сложностей: функция, получая на вход один и тот же аргумент всегда, *всегда* должна возвращать одно и то же значение! Это всё. Больше она ничего при этом делать не должна. Пойдём дальше и скажем, что она обязана больше ничего и не делать! Это просто правило соответствия '$x \mapsto y$'. Получается такое простое определение:

Чистая функция
Функция, которая при получении одних и тех же входных данных всегда возвращает одни и те же выходные, не имея при этом побочных эффектов (то есть никак больше не влияющая на окружающий мир и не имеющая внутреннего состояния, меняющего выходные данные).

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

Что касается возвращения к истокам, тут уместно вспомнить, что изначально в терминологии языка Pascal ещё было формальное различие между функциями (под чем понимались функционально чистые функции) и процедурами (изменяющими внешнее состояние), которое современные языки потеряли.

In [1]:
from typing import Any

lst = ['a', 'b']
print(lst)


def change(lst: list[Any]) -> None:
    """Remove the last item from a list."""

    lst.pop()
    # x Unpure! >.<


change(lst)
print(lst)
['a', 'b']
['a']
In [2]:
from typing import Any

lst = ['a', 'b']
print(lst)


def change(lst: list[Any]) -> list[Any]:
    """Remove the last item from a list"""

    return lst[:-1]
    # ✓


change(lst)
print(lst)
['a', 'b']
['a', 'b']

Чистая или нет?

In [209]:
x = 1


def function():
    global x
    x += 1

    return x


function()

Чистая или нет?

In [215]:
y = [1, 2, 3]


def function():
    z = y.append(4)
    return z


function()

scope_resolution_1.png

Ссылочная прозрачность

Функции (вообще всё ФП) обладает ссылочной прозрачностью. Если любой вызов функции заменить её возвращаемым значением, результат от этого поменяться не должен: $$f\left(x\right) = y,$$ а значит $f\left(x\right)$ можно заменять на $y$.

All you need is $\lambda$

Чистая функция вообще не просто функция обладающая ссылочной прозрачностью, но функция, которая может быть выражена в лямбда-исчислении: $$ x \mapsto x $$ $$\lambda x \mathrel{.} x$$

In [1]:
lambda x: x
Out[1]:
<function __main__.<lambda>(x)>

Поиграем роль интерпретатора

$$ \left( \mathop{\lambda} a b c \mathbin{.} c b a \right) z z \left( \mathop{\lambda} w v \mathbin{.} w \right) $$
  1. $ \left( \mathop{\lambda} a b c \mathbin{.} c b a \right) z z \left( \mathop{\lambda} w v \mathbin{.} w \right) $
  2. $ \left( \mathop{\lambda} a \mathbin{.} \mathop{\lambda} b \mathbin{.} \mathop{\lambda} c \mathbin{.} c b a \right) \left(z\right) z \left( \mathop{\lambda} w \mathbin{.} \mathop{\lambda} v \mathbin{.} w \right) $ (каррирование x3)
  3. $ \left( \mathop{\lambda} b \mathbin{.} \mathop{\lambda} c \mathbin{.} c b z \right) \left(z\right) \left( \mathop{\lambda} w \mathbin{.} \mathop{\lambda} v \mathbin{.} w \right) $
    • $ \left( \mathop{\lambda} \left[ a \mathrel{:=} z \right] \mathbin{.} \mathop{\lambda} b \mathbin{.} \mathop{\lambda} c \mathbin{.} c b a \right) z \left( \mathop{\lambda} w \mathbin{.} \mathop{\lambda} v \mathbin{.} w \right) $ ($\beta$-редукция)
  4. $ \left( \mathop{\lambda} c \mathbin{.} c z z \right) \left( \mathop{\lambda} w \mathbin{.} \mathop{\lambda} v \mathbin{.} w \right) $
    • $ \left( \mathop{\lambda} \left[ b \mathrel{:=} z \right] \mathbin{.} \mathop{\lambda} c \mathbin{.} c b z \right) \left( \mathop{\lambda} w \mathbin{.} \mathop{\lambda} v \mathbin{.} w \right) $ ($\beta$-редукция)
  5. $ \left( \mathop{\lambda} w \mathbin{.} \mathop{\lambda} v \mathbin{.} w \right) \left(z\right) z $
    • $ \left( \mathop{\lambda} \left[ c \mathrel{:=} \left( \mathop{\lambda} w \mathbin{.} \mathop{\lambda} v \mathbin{.} w \right) \right] \mathbin{.} c z z \right) $ ($\beta$-редукция)
  6. $ \left( \mathop{\lambda} v \mathbin{.} z \right) z $
    • $ \left( \mathop{\lambda} \left[ w \mathrel{:=} z \right] \mathbin{.} \mathop{\lambda} v \mathbin{.} w \right) z $ ($\beta$-редукция)
  7. $z$
    • $\left( \mathop{\lambda} \left[ v \mathrel{:=} z \right] \mathbin{.} z \right)$ ($\beta$-редукция --- аппликация постоянной функции)

Но зачем?!!

The Zen of Python, by Tim Peters

Beautiful is better than ugly.
Explicit is better than implicit.
Simple is better than complex.
Complex is better than complicated.
...
  • Декларативный подход
  • Формальная верификация (за счёт прослеживания инвариантов программы)
  • Модульность (разделяй-и-властвуй)
  • Композициональность кода как по Фреге (возможность повторного использования) (за счёт высокого уровня абстракции)
  • Простота отладки и тестирования (за счёт втч всего вышеперечисленного)
  • Легко хорошо распараллелить (в теории, а иногда и на практике)
    • (можно жить без GIL — см. PyPy STM)

А при чём здесь собственно Python?

Применяем теорию на практике: Ничего не меняйте!

...

....

Применяем теорию на практике: Ничего не меняйте!

Ложь! На самом деле без локальной мутации объектов внутри функции на Питоне очень не просто бы было программировать. Главное, чтобы эта мутация оставалась лишь внутри области видимости функции и более ничего не затрагивала.

Примеры кода

Функциональный ящик инструментов

In [3]:
#from typing import Iterator, Union
from collections.abc import Iterator, Iterable
from typing import Any, TypeVar
# import itertools

# StandardNumericType = Union[int, float]
# StandardNumericType = int | float
Number = TypeVar('Number', int, float)
T = TypeVar('T')


def count(
    start: Number = 0,
    step: Number = 1
) -> Iterator[Number]:
    """Reinvent itertools.count."""

    number: Number = start
    while True:
        yield number
        number += step


def cut_off_iterator_at(iterator: Iterable[T], val: T) -> Iterable[T]:
    """Return a new iterator from a given one
    that cuts it off at reaching value val (without it).
    """

    for item in iterator:
        if item == val:
            break

        yield item


def sum_of_ints_up_to(upper_bound: int) -> int:
    """Sum integers until reaching the upper_bound
    (not including it).
    """

    summa: int = 0
    successive_integers: Iterable[int] = count(0)
    # for number in successive_integers:
    #     if int(number) == upper_bound:
    #         break
    #     summa += int(number)
    summa = sum(cut_off_iterator_at(successive_integers,
                                    val=upper_bound))

    return summa


sum_of_ints_up_to(100)
Out[3]:
4950
In [4]:
def naive_is_prime(n: int) -> bool:
    """Check if n is prime (directly by brute force)
    """

    if n < 2:  # 0 and 1 are not prime
        return False

    if n == 2:  # 2 is prime
        return True

    if n % 2 == 0:  # exclude multiples of two
        return False

    for k in range(3, 1+int(n**0.5), 2):
        # check if n is divisible by some odd number up to sqrt(n)
        if n % k == 0:
            return False

    return True


naive_is_prime(29)
Out[4]:
True
In [5]:
# from typing import Callable, Iterator, Iterable, TypeVar, Union
from collections.abc import Callable, Iterator, Iterable
from typing import TypeVar
from functools import partial

# StandardNumericType = int | float
T = TypeVar('T')


def within_approximation(number1: float,
                         number2: float,
                         eps: float = 10e-5) -> bool:
    """Check if number1 equals number2 within given approximation."""

    return abs(number2 - number1) <= eps


def apply_iteratively(function: Callable[[T], T],
                      initial_value: T) -> Iterator[T]:
    """Apply function iteratively with initial_value,
    yield each result in the sequence of applications.
    """

    yield initial_value
    # for val in apply_iteratively(function, function(initial_value)):
    #     yield val
    yield from apply_iteratively(function, function(initial_value))


def heron_method(appr: float, num: float) -> float:
    """Calculate an iteration of Heron's method's sequence
    to compute the square root.
    """
    return (appr+num/appr) / 2


def limit(
    term_def: Callable[[float], float],
    initial_val: float,
    ε: float = 10e-5
):
    """Find limit of a converging series with error ε."""

    def head_tail(ε: float,
                  a0: float,
                  iterable: Iterator[float]) -> float:
        b = next(iterable)
        if within_approximation(a0, b, ε):
            return b

        return head_tail(ε, b, iterable)

    functional_sequence: \
        Iterator[float] = apply_iteratively(
            term_def,
            initial_val
        )

    return head_tail(
        ε,
        next(functional_sequence),
        functional_sequence
    )


limit(partial(heron_method, num=5),
      2, 10e-8) == 5 ** 0.5
Out[5]:
True
In [6]:
def group_by_iter(n: int, iterable: Iterator[T]) -> Iterator[tuple[T, ...]]:
    row: tuple[T, ...] = tuple(next(iterable) for _ in range(n))
    while row:
        yield row
        row = tuple(next(iterable) for _ in range(n))
In [7]:
import bisect
from collections.abc import Iterable, Mapping
from typing import Any


class StaticMapping(Mapping):
    def __init__(self, iterable: Iterable[tuple[Any, Any]]) -> None:
        self._data = tuple(iterable)
        self._keys = tuple(sorted(key for key, _ in self._data))

    def __getitem__(self, key):
        ix = bisect.bisect_left(self._keys, key)
        if (ix != len(self._keys)
                and self._keys[ix] == key):
            return self._data[ix][1]
        raise KeyError('{0!r} not found'.format(key))

    def __iter__(self) -> Iterator[Any]:
        return iter(self._keys)

    def __len__(self) -> int:
        return len(self._keys)

Аннотация переменных (mypy)

Генераторы

«The wilderness yieldeth food for them and for their children.»
―The Bible (King James Version)

In [8]:
def simple_count(start=0):
    """Generate successive integers from start"""

    while True:
        start += 1
        yield start  # кнопка "Пауза"
        # return = yield
        # + удаление локальной
        # области видимости
        # со всеми переменными
        # yield = return
        # + удержание состояния и
        # локальных переменных
        # функции в памяти


numbers = simple_count()
print(numbers)
print(numbers.__next__())
print(next(numbers))
print(next(numbers))
<generator object simple_count at 0x7f5c29481460>
1
2
3
In [9]:
def stop(): yield


s = stop()
next(s)
next(s)
---------------------------------------------------------------------------
StopIteration                             Traceback (most recent call last)
Input In [9], in <cell line: 6>()
      4 s = stop()
      5 next(s)
----> 6 next(s)

StopIteration: 

Ленивые вычисления (отложенные)

«Глупо думать про лень негативно
И надменно о ней отзываться:
Лень умеет мечтать так активно,
Что мечты начинают сбываться.»
—Игорь Губерман
  • способы создания итераторов
    • iter(range(...))
    • iter(iterable)
    • генераторные выражения (вида (... for item in iterable ...))
    • пользовательские генераторы (yield и yield from iterable)
  • ветвление if-else
  • логические операторы or, and...
In [ ]:
iterable: Iterable[Any] = []

for item in iter(iterable):
    # ...
    pass

for item in iterable:
    # ...
    pass

Проблемы в раю

«I have never considered Python to be heavily influenced by functional languages,
no matter what people say or think. I was much more familiar
with imperative languages such as C and Algol 68
and although I had made functions first-class objects,
I didn't view Python as a functional programming language.
[…]In general, because Python's extremely dynamic nature,
it is impossible to do the kind of compile-time optimization
known from functional languages like Haskell or ML. And that's fine.»
—Гвидо ван Россум, https://python-history.blogspot.com/2009/04/origins-of-pythons-functional-features.html
  • Модуль functools скорее является дополнением к стандартному Python, прилаженным к нему потом. И так со всем FP. Объекты Python не так хороши для FP в реальности и совершенно для него не оптимизированы. Их можно для него использовать, но нельзя сказать, что они специально на это рассчитаны. Вообще структуры Python не оптимизированы для FP.
  • Итераторы имеют изменяющееся состояние. Могут быть проитерированы лишь однажды. Потом заканчиваются.
  • Нет встроенных связанных списков :(
  • нет TCO (tail call optimization — оптимизации хвостовой рекурсии) sys.setrecursionlimit(1000) (размер стека в любом случае ограничен.)
In [10]:
import sys


def call_me(me, counter=0, limit=2971):
    if counter == limit:
        return counter
    return me(me, counter + 1)


print(call_me(call_me), sys.getrecursionlimit())
2971 3000
In [11]:
import itertools
from typing import Any
from collections.abc import Iterable


def limits(iterable: Iterable[Any]) -> tuple[Any, Any]:
    """Find minimum and maximum for an iterable
    and return a tuple with them.
    """

    tee1, tee2 = itertools.tee(iterable, 2)

    return min(tee1), max(tee2)


print(*itertools.islice(itertools.count(5), 0, 15 + 1))
seq = itertools.islice(itertools.count(5), 0, 15 + 1)
print()
print(*limits(seq))
next(seq)
5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

5 20
---------------------------------------------------------------------------
StopIteration                             Traceback (most recent call last)
Input In [11], in <cell line: 20>()
     18 print()
     19 print(*limits(seq))
---> 20 next(seq)

StopIteration: 

Software_and_hardware-Alan_Kay.jpg

Задача. Сделайте код более функциональным

In [12]:
import random
from collections.abc import Iterator
from typing import cast

some_data: list[float] = [random.uniform(0, 10) for _ in range(20)]
iterable: Iterator[float] = iter(some_data)

begin: float = next(cast(Iterator[float], iterable))
res: float = 0
for end in iterable:
    res = max(res, sum((begin, end)))
    begin = end

print(res)
19.411006963835693
In [13]:
# import random
from collections.abc import Iterator
from typing import Any, TypeVar, Optional, cast

T = TypeVar('T')
Pair = tuple[Optional[T], Optional[T]]


def break_into_pairs(iterator: Iterator[T]) -> Iterator[Pair]:
    """Accrete an iterator's data to form another iterator
    of successive pairs of values.
    For example, for an iterator returning s1, s2, s3, ...
    we will have (s1, s2), (s2, s3), (s3, ...), (..., ...), ...
    """

    first: T = next(iterator)
    for second in iterator:
        yield first, second
        first = second


# some_data: list[float] = [random.uniform(0, 10) for _ in range(20)]
iterable: Iterator[float] = iter(some_data)
paired_iterable: Iterator[Pair] = break_into_pairs(
    cast(Iterator[float], iterable)
)
res: float = max(map(sum, paired_iterable))

print(res)
19.411006963835693
In [14]:
# import random
from collections.abc import Sequence
from typing import Any, TypeVar, Optional, cast

T = TypeVar('T')
Pair = tuple[Optional[T], Optional[T]]


def break_seq_into_pairs(seq: Sequence[T]) -> Iterator[Pair]:
    """For example, for a sequence of values s1, s2, s3, ...
    we will return an iterator of (s1, s2), (s2, s3), (s3, ...),  ...
    """

    return zip(seq[0::2], seq[1::2])


# some_data: list[float] = [random.uniform(0, 10) for _ in range(20)]
paired_iterable = break_seq_into_pairs(some_data)
# res: float = max(map(sum, paired_iterable))
res = max(map(sum, paired_iterable))

print(res)
19.411006963835693
In [15]:
from itertools import zip_longest
from collections.abc import Sequence
from typing import Any, TypeVar, Optional, cast

T = TypeVar('T')


def break_seq_into_tuples(seq: Sequence[T],
                          tuple_len: int) -> Iterator[tuple[T, ...]]:
    """Return an iterator of all tuples of n consecutive elements
    for a given sequence.
    """

    return zip_longest(*(seq[k::tuple_len] for k in range(tuple_len)))


max(map(sum, break_seq_into_tuples(some_data, 2)))
Out[15]:
19.411006963835693

Функциональный подход vs императивный подход на простейшем примере

In [16]:
# imperative procedural 'von Neumann' style

dim = 3
v1 = (1, 2, 3)
v2 = (6, 5, 4)

scalar_prod = 0
for k in range(dim):
    scalar_prod += v1[k] * v2[k]

scalar_prod
Out[16]:
28
In [17]:
#  a more pythonic style
# (and, incidentally (or is it??) more functional)
# but still imperative

dim = 3
v1 = (1, 2, 3)
v2 = (6, 5, 4)

scalar_prod = 0
for pr1, pr2 in zip(v1, v2):
    scalar_prod += pr1 * pr2

scalar_prod
Out[17]:
28
In [19]:
# functional declarative style

from math import prod

dim = 3
v1 = (1, 2, 3)
v2 = (6, 5, 4)

scalar_prod = sum(map(prod, zip(v1, v2)))

scalar_prod
Out[19]:
28

Ссылки

Литература

Её практически нет. ._.

Awesome Functional Python ссылается на две "книги" (одна из которых состоит из всего ~37 страниц, а вторая (хоть и из ~398) мне показалась чтивом весьма сумбурным и специфическим).