Функциональное программирование в Python: ежедневные рецепты
Как говорится, спроси пять программистов, что такое функциональное программирование, получишь шесть разных ответов. В целом это программирование через функции в их математическом понимании, то есть когда функция принимает что-то на вход и что-то возвращает на выходе, не меняя глобального состояния.
В своей команде — команде разработки инструментов для разработчиков под 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»
Мы должны выдать новую строку, в которой все повторения будут заменены на количество этих символов:
»4A3B2C1×1Y1Z4D3E3F6A2B»
Классическая реализация (через вложенный цикл) может выглядеть так:
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[]"
Видно, что ошибки детектятся. То есть 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, которая работает над инструментом с жесткими требованиями по производительности. Давайте вместе выжмем из языка максимум :)
Материалы
Книги:
Статьи:
Доклады:
