Как говорится, спроси пять программистов, что такое функциональное программирование, получишь шесть разных ответов. В целом это программирование через функции в их математическом понимании, то есть когда функция принимает что-то на вход и что-то возвращает на выходе, не меняя глобального состояния.

В своей команде — команде разработки инструментов для разработчиков под KasperskyOS — мы создаем разные интересные консольные утилиты, эмулятор, обеспечиваем интеграцию с IDE и так далее. И для этого мы используем разные языки — C++, C, TypeScript; но больше всего пишем на Python.



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

Чистые функции

Начну с того, что попроще, — с чистых функций, а потом перейду к более сложным вещам.

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

def distance(p1: Point, p2: Point) -> float:
    return ((p1.x - p2.x) ** 2 + (p1.y - p2.y) ** 2) ** 0.5

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

Что можно считать побочным эффектом? Это практически бесконечный список действий:

• Изменение внешних переменных
• Логирование
• Обращения к базе данных
• Выполнение сетевых запросов
• Вывод текста (в консоль, в файл…)
• Получение пользовательского ввода
• Импорт модуля и др.

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

Безотносительно функционального подхода к разработке у чистых функций есть два больших преимущества.

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

def do_something(x):
    ...
with Pool(N) as p:
    print(p.map(do_something, data_stream))

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

@functools.cache
def do_something(a, b):
    ...

Функции высшего порядка

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

@retry(3)
def upload(data):
    ...
upload_with_retries = retry(3)(upload)

В данном случае декоратор рассматривает функцию как обычный объект — принимает ее на вход, добавляет какое-то поведение и возвращает на выходе.

def retry(retries: int):
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            result = None
            for attempt in range(retries):
                try:
                    result = func(*args, **kwargs)
                    return result
                except Exception as e:
                    if attempt + 1 == retries:
                        raise e
            return result
        return wrapper
    return decorator

Декоратор также добавляет интересный концепт, а именно специализацию функции через добавление к ней данных. В ООП мы привыкли, что у нас есть данные с функциями — классы. Здесь же получается наоборот — функция, в которую мы добавили данные (в нашем случае — максимальное количество перезапусков).

Частичное применение

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

import sys
from functools import partial

def some_func(logger: Callable):
    logger('Hello')
    logger('World')

def log(date, level, message): ...

warning_log = partial(log, now(), 'WARNING')
some_func(warning_log)

Каррирование

Схожее и очень важное в функциональном мире понятие: каррирование (Curring) — это превращение функции от нескольких переменных в серию функций от одной переменной. Во многих чисто функциональных языках функции от нескольких аргументов все по умолчанию реализуют такой подход.

curry(f(a, b, c)) => f(a)(b)(c)

Фактически это синтаксический сахар вокруг partial, который доступен в пакетах toolz и funcy.

Попробуем на том же примере применить каррирование. Для этого фиксированные аргументы нужно будет добавить немного по-другому.

from toolz import curry

def some_func(logger: Callable):
    logger('Hello')
    logger('World')

def log(date, level, message): ...

curried_log = curry(log)
warning_log = curried_log(some_date)('WARNING')

some_func(warning_log)

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

Композиция функций

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

from toolz import compose

def str_to_json(column): ...
def json_to_data(json): ...
def data_to_user(data): ...

parse_column = compose(str_to_json, json_to_data, data_to_user)
users = (parse_column(column) for column in columns)

К композиции я вернусь чуть позже.

Ленивые вычисления

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

Приведу пример, который позволяет понять некоторые преимущества этого подхода.

def decode(line: str):
    ...

def get_lines(filename):
    with open(filename) as f:
        for line in f:
            yield line
...
pattern = r"^Abnormal program \(.*\) termination\.\.\.$"
decoded = (decode(line) for line in get_lines("file.log"))

if any(line for line in decoded if re.search(regexp_line)):
    print("Found!")

Здесь мы читаем файл, но явно нигде не указываем, в какой момент хотим перестать получать строки из него. Также есть декоратор, который декларативно описывает, что собой представляют декодированные строки. Все это отправляется в функцию any, поддерживающую ленивое исполнение, — она будет получать значения только до тех пор, пока они ей нужны.

Если бы ленивого исполнения не было, пришлось бы использовать if или циклы с условиями выхода. А ленивые вычисления помогают не только писать более лаконичный код, но и по умолчанию быть более производительными, ведь если мы где-то забудем условия выхода, код будет медленнее.

Рекурсия как основной способ имплементации алгоритмов

В чисто функциональных языках рекурсия — очень важный инструмент. Многие из них не поддерживают других способов создания итераций (while, for и так далее). Рекурсия позволяет избежать побочных эффектов, потому что предоставляет возможность сделать изолированные итерации без изменения состояния. То есть используя обычный цикл, пожелав использовать в каждой итерации результаты предыдущей или просто какие-то переменные, мы вольны брать что-то извне. А рекурсия заставляет четко определить те аргументы, которые мы хотим передать в функцию, и то, как мы будем с ними работать.
Кроме того, рекурсия — это инструмент математического языка. Но мы сейчас не об этом.

Продемонстрирую рекурсию на примере. Если бы на Python мы захотели написать сервер так, как на Haskell, он бы выглядел так:

def run_connection(client):
...

def main_loop(server_socket):
    client, _ = server_socket.accept()
    run_connection(client)
    main_loop(server_socket)

def main():
…
    main_loop(server_socket)

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

На практике такой сервер долго не протянет — где-то на тысячном клиенте мы взорвем стек, потому что в Python нет оптимизации хвостовой рекурсии (tail call optimization). Некоторые языки под капотом превращают рекурсию в цикл, но не Python.

TCO-оптимизация (tail call optimization)

Для tail call optimization есть готовые решения. Например, в библиотеке fnpy есть специальный декоратор, который позволит реализовать ту же самую рекурсию, но с измененным синтаксисом: мы должны возвращать true, когда мы хотим продолжить итерации, и false, если хотим их остановить.

from fn import recur

@recur.tco
def main_loop(server_socket):
    client, _ = server_socket.accept()
    run_connection(client)
    return True [server_socket]

Реализация у этого декоратора очень простая:

class tco(object):
    def __init__(self, func):
        self.func = func
    def __call__(self, *args, **kwargs):
        action = self
        while True:
            result = action.func(*args, **kwargs)
            if not result[0]: return result[1]
            act, args = result[:2]
            if callable(act): action = act
            kwargs = result[2] if len(result) > 2 else {}

Под капотом простой цикл, который зависит от того, какие значения мы возвращаем.
Надо отметить, что есть и более экстремальные декораторы, которые меняют байт-код на лету, получая более естественный TCO (https://medium.com/@tianlihua1024/python-tco-tail-call-optimization-12dce31d7dca).

А встроенную TCO в Python ждать не стоит. На этот счет уже были баталии, и Гвидо ван Россум написал статью, почему считает, что такая оптимизация не нужна. По его мнению, из-за нее одни вызовы будут иметь нормальный стектрейс, в другие — нет, и это усложнит дебаг. Если интересно, вот его статья: http://neopythonic.blogspot.com/2009/04/final-words-on-tail-calls.html.

В примере выше можно просто использовать цикл, и сильно ситуация не ухудшится.

def main_loop(server_socket):
    while True:
        client, _ = server_socket.accept()
        run_connection(client)

Я встречал рекомендации писать рекурсии, но на последнем шаге менять их на циклы (то есть вроде реализовали и отладили алгоритм на рекурсии, но потом сделали ее циклом).

Несмотря все на это, рекурсия — хороший и полезный инструмент. Множество алгоритмов в рекурсивном варианте читаются проще. Например, поиск в глубину:

Через цикл:

def iterative_dfs(visited, graph, node) :
    stack = [node] 
    while stack: 
        vertex = stack.pop() 
        if vertex not in visited: 
            visited.add(vertex) 
            for neighbour in graph[vertex]: 
                stack.append(neighbour)

Через рекурсию:

defrecursive_dfs(visited, graph, node): 
    if node not in visited: 
        visited.add(node) 
        for neighbour in graph[node]: 
            recursive_dfs(visited, graph, neighbour)

Особый акцент на обработке последовательностей и декларативное программирование

Если немного напрячь воображение, то может показаться, что все вокруг — это последовательности. Когда парсим файл, у нас есть списки символов, строк и так далее. В игре есть списки игровых объектов, действий и тому подобное:

• Парсинг файла — список строк -> список токенов -> список символов
• Сервер — список соединений -> список запросов -> список ответов
• Игра — список игровых объектов -> список действий -> список событий

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

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

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

Лично мне больше нравится вариант с comprehensions:

map(lambda x: int(is_prime(x)), range(2, 10)) -> 1, 1, 0, 1, 0, 1, 0, 0
filter(lambda x: is_prime(x), range(2, 10)) -> 2, 3, 5, 7

(is_prime(x) for x in range(2, 10)) -> 1, 1, 0, 1, 0, 1, 0, 0
(x for x in range(2, 10) if is_prime(x)) -> 2, 3, 5, 7

Подробнее об этом — в статье The fate of reduce() in Python 3000 — by Guido van van Rossum (https://www.artima.com/weblogs/viewpost.jsp?thread=98196).

Композиция в работе с последовательностями

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

file_rows = get_file_rows('filght.csv')
parsed_filtered = []
for row in file_rows:
    pos = PlanePos.from_row(row)
    if pos.in_polygon(restrictedArea):
        parsed_filtered.append(pos)
sorted_data = sorted(parsed_filtered, key=lambda pos: (pos.lat, pos.time))
violations = []
for pos in sorted_data:
    violations.append(f"{pos.time}: {pos.lat}-{pos.lon}")

Особых проблем в этом коде нет. Все более-менее читается. Но мы попробуем его улучшить. Желая применить идеи функционального программирования в Python, можно сделать что-то такое:

file_rows = get_file_rows('flight.csv')

violations = list(
    map(lambda pos: f"{pos.time}: {pos.lat}-{pos.lon}",
        sorted(
            filter(lambda pos: pos.in_polygon(restrictedArea),
                map(PlanePos.from_row, file_rows)),
            key=lambda pos: (pos.lat, pos.time)
        )
    )
)

Здесь декларативно описан тот же алгоритм. А результат получен одним выражением.

Главная проблема этого кода — плохая читаемость. Первая операция над входным элементом находится не в начале или конце, а где-то в середине выражения, и это может смутить. Пробуем применить comprehensions.

file_rows = get_file_rows('flight.csv')
violations = [f"{pos.time}: {pos.lat}-{pos.lon}" for pos in
    sorted((pos for pos in (PlanePos.from_row(row) for row in file_rows)
        if pos.in_polygon(restrictedArea)),
    key=lambda pos: (pos.lat, pos.time))]

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

file_rows = get_file_rows('filght.csv')
parsed = (PlanePos.from_row(row) for row in file_rows)
in_polygon = (pos for pos in parsed if pos.in_polygon(restrictedArea))
sorted_by_lat_time = sorted(in_polygon, key=lambda pos: (pos.lat, pos.time))
violations = [f"{pos.time}: {pos.lat}-{pos.lon}" for pos in sorted_by_lat_time]

Местами этот код будет удобнее, чем императивный.
Но есть две проблемы.

Первая — у нас появилось несколько переменных, которые изначально были не нужны.

Второе — то, что, если часть этих операций становится не нужна, например отпадает необходимость в сортировке, нам недостаточно убрать одну строку с сортировкой. Нужно будет вычистить все, что касается промежуточной переменной.

Предлагаю отвлечься от Python и посмотреть, как это реализовано в других языках. Там иногда встречается такая форма композиции, как конвейер. Это тоже применение функции к результатам выполнения других функций, но в более линейном варианте, когда каждая последующая функция принимает аргументы слева и передает результат направо. Все это соединяется через специальный символ — точку, пайп и тому подобное.

any(map(filter(...,),) # Композиция
.filter(..).map(..).any(..) # Конвейер

В других языках, пусть даже не чисто функциональных, а, например, в C#, наша задача решалась бы следующим образом. Здесь есть библиотека LINQ, которая позволяет делать конвейеры:

file_rows.Select(PlanePos.FromRow)
    .Where(pos => pos.InPolygon(restrictedArea))
    .OrderBy(pos => pos.Lat)
    .ThenBy(pos => pos.Time)
    .Select(pos => $"{pos.Time}: {pos.Lat}-{pos.Lon}")

Функция Select — это некоторый аналог map, которая применит парсер к каждому элементу последовательности. После нее используем фильтр Where. Далее присутствует сортировка и еще один Map для форматирования итоговых строк.

Теперь код выглядит неплохо: читается хорошо, операции идут по порядку. Для реализации такого подхода в Python есть библиотеки, например pylinq или pipe. Возьмем для примера pipe — она позволяет сделать конвейер через вертикальную черту (пайп):

from pipe import select, where, sort

violations = (file_rows |
    select(PlanePos.from_row) |
    where(lambda pos: pos.in_polygon(restrictedArea)) |
    sort(key=lambda pos: (pos.lat, pos.time)) |
    select(lambda pos: f"{pos.time}: {pos.lat}-{pos.lon}"))

Здесь так же будут использованы функции Select, Where, Resort и они будут применены в той последовательности, в которой мы их напишем. Когда используется много обработок последовательностей, с pipe может получиться более читаемый код.

Если сравнить с императивным подходом, то получается немного интереснее:

Было:

parsed_filtered = []

for row in file_rows:
    pos = PlanePos.from_row(row)
    if pos.in_polygon(restrictedArea):
        parsed_filtered.append(pos)

sorted_data = sorted(parsed_filtered, key=lambda pos: (pos.lat, pos.time))

violations = []
for pos in sorted_data:
    violations.append(f"{pos.time}: {pos.lat}-{pos.lon}")

Стало:

violations = (file_rows |
    select(PlanePos.from_row) |
    where(lambda pos: pos.in_polygon(restrictedArea)) |
    sort(key=lambda pos: (pos.lat, pos.time)) |
    select(lambda pos: f"{pos.time}: {pos.lat}-{pos.lon}"))

Не обязательно использовать все эти многоэтажные выражения, если все помещается в одну строку. Допустим, мы хотим просто взять вылеты самолета ниже определенной широты и вывести их время в одну строку. Тоже получается интересно и лаконично:

from pipe import select, take_while
result = flight | take_while(lambda p: p.lat < 80) | select(lambda p: p.time)

Лямбды

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

Начнем с того, какие минусы видят в лямбдах на Python:

• Нельзя писать многострочные лямбды — мы не можем начать ее на одной строке и закончить на другой.
• Многословный синтаксис, даже в С++ получилось короче:

auto fun = []{}; // валидная c++ лямбда
fun();

Для примера рассмотрим такой код — я его увидел в какой-то библиотеке по статистике.

{
    "min": lambda x: x.min(),
    "25%": lambda x: x.quantile(0.25),
    "50%": lambda x: x.quantile(0.50),
    "median": lambda x: x.median,
    "#>0": lambda x: (x > 0)
}

Автору пришлось на каждой строке писать по слову lambda, хотя задача — всего-то вызвать какой-то метод, взять атрибут и выполнить сравнение. Чтобы сократить код, лямбды можно генерировать — написать методы call и get, которые будут выдавать лямбды:

def call(name, *args, **kwargs):
    return lambda obj: getattr(obj, name)(*args, **kwargs)

def get(item):
    return lambda obj: getattr(obj, item)

def gt(number):
return lambda x: x > number

{
    "min": call("min"),
    "25%": call("quantile", 0.25),
    "50%": call("quantile", 0.50),
    "median": get("median"),
    "# > 0": gt(0)
}

Как можно догадаться, все это уже давно существует. Есть модуль operator, который содержит в себе методы call и itemgetter, с помощью которых можно избавиться от лямбд.

from functools import partial
from operator import methodcaller as call, itemgetter as get, lt
{
    "min": call("min"),
    "25%": call("quantile", 0.25),
    "50%": call("quantile", 0.50),
    "median": get("median"),
    "# > 0": partial(lt, 0),
}

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

Тут стоит упомянуть о библиотеке whatever, реализующей магический объект «нижнее подчеркивание», которое позволяет вместо лямбды написать такое странное выражение:

from pipe import select, take_while
result = flight | take_while(lambda p: p.lat < 80) | select(lambda p: p.time)

from whatever import _
from pipe import select, take_while
result = flight | take_while(_.lat < 80) | select(_.time)

Здесь на место нижнего подчеркивания придет элемент последовательности — атрибуты lat и time будут взяты у него.

Для примера я взял кусок кода из библиотеки youtube-dl, где много лямбд. В ней нашелся ассоциативный массив с большим количеством атрибутов видео:

return {
    "title": try_get(video, lambda x: x['title']),
    "category": try_get(video, lambda x: x['category']),
    "age_limit": try_get(video, lambda x: x['is_adult']),
    "view_count": try_get(video, lambda x: x['hits']),
    "uploader": try_get(video, lambda x: x['author']),
    "duration": try_get(video, lambda x: x['duration']),
}

Если вместо лямбд использовать whatever, код получается немного короче.

return {
    "title": try_get(video, _['title']),
    "category": try_get(video, _['category']),
    "age_limit": try_get(video, _['is_adult']),
    "view_count": try_get(video, _['hits']),
    "uploader": try_get(video, _['author']),
    "duration": try_get(video, _['duration']),
}

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

«AAAABBBCCXYZDDDDEEEFFFAAAAAABB»

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

«4A3B2C1X1Y1Z4D3E3F6A2B»

Классическая реализация (через вложенный цикл) может выглядеть так:

def rle_encode(data: str) -> str: 
    encoded = [] 
    i = 0 
    while i <= len(data)-1: 
        count = 1 
        ch = data[i] 
        j = i 
        while j < len(data) - 1: 
            if data[j] == data[j+1]: 
                count = count+1 
                j = j + 1
            else:
                break
        encoded.append(str(count) + ch)
        i = j + 1
    return "".join(encoded)

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

from itertools import groupby

def rle_encode(data: str) -> str:
    return "".join(str(len(list(group_items))) + key
        for key, group_items in groupby(data))

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

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

Думаем на уровне типов

Типизация в функциональных языках — это отдельный большой и важный раздел. Часто говорят: «Если на каком-нибудь Haskell программа компилируется, значит, она, скорее всего, работает». Это, конечно, шутка, но зерно истины тут есть — строгая система типов позволяет совершать меньше ошибок.

Из интересного в функциональных языках есть фантомные и рекурсивные типы. Чтобы взглянуть на них подробнее, придумаем себе проблему.

Вспомним задачу с координатами самолета. Обычно их представляют как-то так:

@dataclass
class PlanePos:
    lat: float
    lon: float

Сами широта и долгота — просто числа, вещественный тип, выглядят они следующим образом.

lat = 50
lon = 120

Но так мы открываем себе простор для ошибок.

Во-первых, их можно поменять местами — случайно перепутать.

p = PlanePos(lon, lat)

С точки зрения тайпчекера, и то, и то float, то есть ошибки не будет.

Во-вторых, мы можем выполнять с ними бессмысленные операции, например:

p3.lat = p1.lat + p2.lon

Решают проблему фантомные типы. Так называют типы, которые не используют как минимум один из своих параметров.
Пояснить проще на примере. Предположим, есть класс MyType шаблонный аргумент T.

T = TypeVar('T')
class MyType(Generic[T]):
    value: int

Если он не используется, это и будет фантомный тип.

С помощью фантомных типов можно переписать наш пример:

class Latitude: pass
class Longitude: pass

T = TypeVar('T')
class Coordinate(Generic[T]): 
    def __init__(self, value: float): ... 
    def __add__(self, second: Coordinate[T]): ... 

@dataclass 
class PlanePos: 
    lat: Coordinate[Latitude] 
    lon: Coordinate[Longitude]

Простора для ошибок стало меньше. Теперь широта и долгота уже не float, а некий класс координат. В целом он одинаковый как для широты, так и для долготы, но мы специализируем эти классы дополнительными Latitude и Longitude. Таким образом классы становятся несовместимыми, и мы не можем их сложить — Mypy или любой другой тайпчекер не позволит.

latitude = Coordinate[Latitude](10)
longitude = Coordinate[Longitude](50)

PlanePos(longitude,latitude) #Mypy error
nonsense = latitude+longitude #Mypy error

В данном случае мы оставляем себе простор для других ошибок. Помимо возможности перепутать их местами, мы можем задать некорректные значения. Широта должна попадать в промежуток от –90 до 90 градусов, то есть –100 градусов будет некорректно:

latitude = Coordinate[Latitude](-100)
longitude = Coordinate[Longitude](190)

Прошлый пример никакой валидации не дает, хотя она тоже по сути часть типа. Здесь мы можем использовать библиотеку phantom-types (https://github.com/antonagestam/phantom-types), которая позволит сильно упростить пример, добавив дополнительные свойства типа.

Библиотека позволяет задать предикат для классов, который будет проверять, что переданное значение ему удовлетворяет (в нашем случае попадает в промежуток). Мы будем использовать стандартный предикат inclusive, но здесь можно было бы написать любой другой предикат, который проверит, что, например, строка удовлетворяет регулярному выражению или дерево является сбалансированным.
Вот что получится:

from phantom import Phantom
from phantom.predicates.interval import inclusive
class Latitude(float, Phantom, predicate=inclusive(-90, 90)):
...
class Longitude(float, Phantom, predicate=inclusive(-180, 180)):
...
@dataclass
class PlanePos:
    lat: Latitude
    lon: Longitude
PlanePos(500, -500) # mypy error
PlanePos(Latitude(500), Longitude(-500)) # runtime error
PlanePos(Longitude(120), Latitude(80)) # mypy error
PlanePos(Latitude(80), Longitude(120)) # ok

Теперь мы не можем создать широту и долготу с неправильными значениями. И по-прежнему не можем перепутать их местами, потому что это разные типы.
Единственное, что можем сделать, это передать в конструктор широту и долготу на нужных местах с правильными значениями. Это и есть принцип Parse, don't validate в действии.

Вторая интересная концепция — рекурсивные типы. Это типы, которые в определении используют сами себя. Отличный пример — дерево. У него есть правое и левое поддерево, чей тип — то же самое дерево.

from __future__ import annotations

class Tree:
    def __init__(self, left: Tree, right: Tree):
        self.left = left
        self.right = right

Самый простой пример, зачем они могут быть нужны, — это JSON. Здесь я привел определение JSON — это объединение нескольких простых типов, по сути тот же JSON (потому что JSON может содержаться сам в себе):

from __future__ import annotations
from typing import Union, Dict, List

JSON = Union[None, bool, str, float, int, List[JSON], Dict[str, JSON]]

good: JSON = {'a': ['b', {"c": [1, 2]}]}
bad: JSON = {'a': ['b', {3: [1, 2]}]}
ugly: JSON = {'a': ['b', {"c": [set(), 2]}]}

Я привел три JSON-а:

• хороший — правильный с точки зрения созданного типа,
• плохой — содержит int в качестве ключа,
• ужасный — содержит в качестве элементов list set (этого я в типе также не допустил).

Согласно нашим типам ошибка содержится уже во вложенном JSON. Вот что скажет тайпчекер (например, Mypy):

from __future__ import annotations
from typing import Union, Dict, List

JSON = Union[None, bool, str, float, int, List[JSON], Dict[str, JSON]]

good: JSON = {'a': ['b', {"c": [1, 2]}]} # ok
bad: JSON = {'a': ['b', {3: [1, 2]}]} # Dict entry 0 has incompatible type "int"
ugly: JSON = {'a': ['b', {"c": [set(), 2]}]} # List item 0 has incompatible type "Set[<nothing>]"

Видно, что ошибки детектятся. То есть Mypy сходил вглубь в рекурсию и нашел их. Это может пригодиться, главное помнить, что нужно передавать фича-флаг.

--enable-recursive-aliases

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

Иммутабельность

Иммутабельность я выделил отдельно. Вернемся к классу в нашем примере. Предположим, мы хотим хранить последовательность координат (широты и долготы) в List.

@dataclass
class PlanePos:
    lat: Latitude
    lon: Longitude
flight = [
    PlanePos(Latitude(10), Longitude(20)),
    PlanePos(Latitude(40), Longitude(40))
...
]

Идея неплохая, но где-то в другой функции значения можно попортить — задать другое значение для широты или удалить элемент списка:

flight[0].lat = Latitude(0)
del flight[1]

Чтобы от этого защититься, можно использовать Tuple вместо List, сделав последовательность иммутабельной. Вместо обычного DataClass можно использовать NamedTuple:

from typing import NamedTuple

class PlanePos(NamedTuple):
    lat: Latitude
    lon: Longitude
flight = (
    PlanePos(Latitude(10), Longitude(20)),
    PlanePos(Latitude(40), Longitude(40))
...
)

В этом случае удалить уже ничего не удастся.

flight[0].lat = Latitude(0) # AttributeError: can't set attribute
del flight[1] # TypeError: 'tuple' object doesn't support item deletion

Надо помнить, что NamedTuple — не единственный способ добиться такого эффекта. Можно использовать дата-класс Frozen.

@dataclass(frozen=True)
class PlanePos:
    lat: Latitude
    lon: Longitude
flight = (
    PlanePos(Latitude(10), Longitude(20)),
    PlanePos(Latitude(40), Longitude(40))
    ...
)

В этом случае при попытке что-то испортить тоже будет ошибка:

flight[0].lat = Latitude(0) # dataclasses.FrozenInstanceError: cannot assign to field 'lat'
del flight[1] # TypeError: 'tuple' object doesn't support item deletion

Теперь предположим, что мы хотим хранить элементы в словаре. В Python нет иммутабельной версии для dict.



Однако есть очень много библиотек, который реализуют самые разные иммутабельные структуры данных. Для dict есть несколько близких аналогов.



Рассмотрим для примера PMap — самый простой — и применим его к нашему примеру вместо Tuple:

from pyrsistent import pmap

@dataclass(frozen=True)
class PlanePos:
    lat: Latitude
    lon: Longitude

flight = pmap({0: PlanePos(Latitude(10), Longitude(20), []),
    1: PlanePos(Latitude(30), Longitude(40), []),
    ...
})

В результате мы больше не можем испортить значения:

del flight[0] # TypeError: 'PMap' object doesn't support item deletion

Фактически это еще один кирпичик в безопасности программы.

Но не стоит тешить себя иллюзиями. Python — очень динамический язык. И даже у Frozen дата-класса при желании можно изменить любое значение:

@dataclass(frozen=True)
class PlanePos:
    lat: Latitude
    lon: Longitude
    some_list: list

flight = pmap({0: PlanePos(Latitude(10), Longitude(20), []),
    1: PlanePos(Latitude(30), Longitude(40), []),
    ...
})

Все отлично сработает, и никто нас по пальцам не ударит:

object.__setattr__(flight[0], 'lat', Latitude(0)) # Ok
flight[1].some_list.append(1) # Ok too

Естественно, если в дата-классе будет какой-нибудь мутабельный объект, например тот же List, он будет изменяемым (тот факт, что дата-класс Frozen, не означает, что List рекурсивно станет неизменяемым). Об этом лучше помнить.

Существует проект Pyrthon (https://github.com/tobgu/pyrthon/), который позволяет запустить программу так, чтобы все стандартные структуры данных — в том числе List и прочие — были заменены на иммутабельные аналоги из pyrsistent. Его вполне можно посмотреть, если вы пишете очень критический код и хотите использовать именно Python. Но лично я его не пробовал.

Напоследок надо сказать, что тайп-хинты тоже позволяют указать, что атрибут или переменная больше неизменяемые и Mypy будет бить по рукам при попытке их изменить.

from typing import Final

class PlanePos:
    lat: Final[Latitude]
    lon: Final[Longitude]

    def __init__(self, lat: Latitude, lon: Longitude):
        self.lat = lat
        self.lon = lon

pos = PlanePos(Latitude(0), Longitude(0))
pos.lat = Latitude(10) # Mypy error: Cannot assign to final attribute "lat"

Выводы

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

Если вам интересно заглядывать в подобные детали Python, приходите к нам в «Лабораторию Касперского». Как раз сейчас мы ищем Python-разработчика в команду Kaspersky Anti Targeted Attack, которая работает над инструментом с жесткими требованиями по производительности. Давайте вместе выжмем из языка максимум :)

Материалы

Книги:

• Functional Python Programming — Steven F. Lott (https://www.amazon.com/Functional-Python-Programming-functional-expressive/dp/1803232579)
• Functional Programming in Python — David Mertz (https://www.oreilly.com/library/view/functional-programming-in/9781492048633/)

Статьи:

• The fate of reduce() in Python 3000 — by Guido van van Rossum (https://www.artima.com/weblogs/viewpost.jsp?thread=98196)
• Itertools: Itertools tricks (https://github.com/joelgrus/stupid-itertools-tricks-pydata), More itertools (https://www.gidware.com/real-world-more-itertools/), Tour of Python Itertools (https://martinheinz.dev/blog/16)
• Functional Programming HOWTO (https://docs.python.org/3/howto/functional.html)
• Some History of Functional Programming Languages. D. A. Turner (https://www.cs.kent.ac.uk/people/staff/dat/tfp12/tfp12.pdf)
• Python Type Hints are Turing Complete (https://arxiv.org/pdf/2208.14755.pdf) — Github (https://github.com/OriRoth/python-typing-machines/tree/release1.0)
• Рекурсия Tail Recursion Elimination (http://neopythonic.blogspot.com/2009/04/tail-recursion-elimination.html), Final Words on Tail Calls (http://neopythonic.blogspot.com/2009/04/final-words-on-tail-calls.html), Реализация TCO (https://medium.com/@tianlihua1024/python-tco-tail-call-optimization-12dce31d7dca)

Доклады:

• Berlin 2016 — Daniel Kirsch — Functional Programming in Python (https://www.youtube.com/watch?v=r2eZ7lhqzNE)
• Joel Grus: Learning Data Science Using Functional Python (https://www.youtube.com/watch?v=ThS4juptJjQ)