SavePearlHarbor

Функциональное программирование в Python: ежедневные рецепты

—

от автора

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

В своей команде — команде разработки инструментов для разработчиков под 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, которая работает над инструментом с жесткими требованиями по производительности. Давайте вместе выжмем из языка максимум 🙂

Материалы

Книги:

Статьи:

Доклады:


ссылка на оригинал статьи https://habr.com/ru/articles/762788/

Комментарии

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *