Модуль functools предлагает множество полезных функций более высокого порядка, которые взаимодействуют с другими функциями или возвращают их. Эти функции можно использовать для реализации кэширования функций/методов, перегрузки, создания декораторов и в целом для того, чтобы сделать код немного более чистым.
functools.reduce();Кэширование результатов представлено тремя функциями (можно использовать в качестве декоратора)- functools.lru_cache(),- functools.cache(),- functools.cached_property().
Первая из них - @functools.lru_cache() запоминает возвращаемый результат в соответствии с переданными аргументами. Такое поведение может сэкономить время и ресурсы:
from functools import lru_cache import requests @lru_cache(maxsize=32) def get_with_cache(url): try: r = requests.get(url) return r.text except: return "Not Found" for url in ["https://google.com/", "https://yandex.ru/", "https://mail.ru/", "https://google.com/", "https://yandex.ru/", "https://google.com/"]: get_with_cache(url) print(get_with_cache.cache_info()) # CacheInfo(hits=3, misses=3, maxsize=32, currsize=3) print(get_with_cache.cache_parameters()) # {'maxsize': 32, 'typed': False}
В этом примере, с помощью декоратора @lru_cache выполняются GET-запросы и кэшируются их результаты (до 32 кэшированных результатов). Чтобы убедиться, что кэширование действительно работает, можно проверить информацию о кэше, используя метод .cache_info(), который показывает количество попаданий и промахов в кэш. Декоратор также предоставляет методы .clear_cache() и .cache_parameters() для аннулирования кэшированных результатов и проверки параметров соответственно.
Если необходимо иметь немного более продвинутое кэширование, то можно включить необязательный аргумент typed=True, который заставляет аргументы разных типов кэшироваться отдельно.
Еще одним декоратором кэширования в functools является функция, называемая просто functools.cache(). Это простая обертка поверх functools.lru_cache(), в которой отсутствует аргумент max_size, т.е. она не ограничивает количество кэшированных значений.
Декоратор @functools.cached_property() используется для кэширования результатов атрибутов класса. Это очень полезно, если есть свойство, которое дорого вычислять, но при этом является неизменяемым. Работает аналогично встроенной функции property() с добавлением кэширования.
from functools import cached_property class Page: @cached_property def render(self, value): # Что-то делается с аргументом `value`, что приводит к дорогому # вычислению, и как следствие отображение HTML-страницы. # ... return html
Этот простой пример показывает, как можно использовать functools.cached_property(), например, для кэширования отрендеренной HTML-страницы, которая будет возвращаться пользователю снова и снова. То же самое можно сделать для определенных запросов к базе данных или длительных математических вычислений.
Предупреждение для всех кэширующих декораторов: не используйте кэширующие декораторы, если функция имеет какие-либо побочные эффекты или если она создает изменяемые объекты при каждом вызове.
Примечание. Побочный эффект функции - возможность в процессе выполнения своих вычислений: читать и модифицировать значения глобальных переменных, осуществлять операции ввода-вывода, реагировать на исключительные ситуации, вызывать их обработчики.
В Python можно реализовать операторы сравнения, такие как <, >= или ==, используя __lt__, __gt__ или __eq__. Однако реализация каждого из __eq__, __lt__, __le__, __gt__ или __ge__ может быть довольно раздражающей. Модуль functools включает в себя декоратор @functools.total_ordering(), который может помочь в этом - все, что нужно сделать, это реализовать __eq__ и один из оставшихся методов, а остальные будут автоматически созданы декоратором:
from functools import total_ordering @total_ordering class Number: def __init__(self, value): self.value = value def __lt__(self, other): return self.value < other.value def __eq__(self, other): return self.value == other.value print(Number(20) > Number(3)) # True print(Number(1) < Number(5)) # True print(Number(15) >= Number(15)) # True print(Number(10) <= Number(2)) # False
Код показывает, что, несмотря на то, что реализованы только __eq__ и __lt__, можно использовать все расширенные операции сравнения. Наиболее очевидным и важным здесь является сокращение кода и его читабельность.
Всех учили, что перегрузка функций невозможна в Python, но на самом деле есть простой способ реализовать ее, используя две функции в модуле functools - это functools.singledispatch() и/или functools.singledispatchmethod(). Эти функции помогают реализовать то, что называется алгоритмом множественной отправки, который позволяет языкам программирования с динамической типизацией, таким как Python, различать типы во время выполнения.
Перегрузка функций является большой темой, по этому смотрите отдельные материалы:
Часто приходится работать с различными внешними библиотеками или фреймворками, многие из которых предоставляют функции и интерфейсы, требующие от нас передачи функций обратного вызова, например, для асинхронных операций или для прослушивателей событий. В этом нет ничего нового, но что, если встает необходимость передать вместе с функцией обратного вызова некоторые аргументы? Вот где пригодится functools.partial(), ее можно использовать для замораживания некоторых (или всех) аргументов функции, создавая новый объект с упрощенной сигнатурой функции.
Рассмотрим несколько практических примеров:
def output_result(result, log=None): if log is not None: log.debug(f"Result is: {result}") def concat(a, b): return a + b import logging from multiprocessing import Pool from functools import partial logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger("default") p = Pool() p.apply_async(concat, ("Hello ", "World"), callback=partial(output_result, log=logger)) p.close() p.join()
Приведенный выше фрагмент демонстрирует, как можно использовать функцию partial() для передачи функции output_result() ее аргумента log=logger в качестве функции обратного вызова callback. Здесь используется multiprocessing.apply_async(), которая асинхронно вычисляет результат предоставленной функции concat() и возвращает ее результат функции обратного вызова. Но, метод .apply_async() всегда будет передавать результат функции concat() в качестве первого аргумента, а если нужно включить какие-либо дополнительные аргументы, например, как в данном случае log=logger, то необходимо использовать functools.partial().
Это был довольно сложный вариант использования. Более простым примером может быть простое создание функции, которая вместо stdout печатает в stderr:
import sys from functools import partial print_stderr = partial(print, file=sys.stderr) print_stderr("This goes to standard error output")
С помощью этого простого трюка создали новую вызываемую функцию, которая всегда будет передавать аргумент с ключевым словом file=sys.stderr для печати, что позволяет упростить код за счет отсутствия необходимости каждый раз указывать аргумент с ключевым словом.
Функция functools.partial() можно также применить к малоизвестной особенности функции iter(), например можно создать итератор, передав в iter() вызываемую функцию и дозорное значение, что может быть полезно в следующем приложении:
from functools import partial RECORD_SIZE = 64 # Чтение двоичного файла... with open("file.data", "rb") as file: records = iter(partial(file.read, RECORD_SIZE), b'') for r in records: # Сделаем что-нибудь с записью...
Обычно, при чтении файла нужно перебирать строки, но в случае двоичных данных вместо строк приходится перебирать записи фиксированного размера. Это можно сделать, создав вызываемый объект с использованием функции partial(), которая считывает указанный фрагмент данных и передает его итератору, который потом создает из него итератор. Затем этот итератор вызывает функцию чтения до тех пор, пока не будет достигнут конец файла, всегда принимая только указанный фрагмент данных RECORD_SIZE. Наконец, когда достигнут конец файла, возвращается сигнальное значение b'', и итерация останавливается.
При инспектировании кода, когда проверяется имя и строка документации декорированной функции, часто обнаруживается, что они были заменены значениями функции-декоратора. Такое поведение нежелательно, так как невозможно каждый раз переписывать все имена функций и строки документации, когда используется какой-либо декоратор. Это проблема решается при помощи декоратора functools.wraps():
from functools import wraps def decorator(func): @wraps(func) def actual_func(*args, **kwargs): """ Внутренняя функция внутри декоратора, которая выполняет фактическую работу """ print(f"Перед вызовом {func.__name__}") func(*args, **kwargs) print(f"После вызова {func.__name__}") return actual_func @decorator def greet(name): """Здоровается с кем-нибудь""" print(f"Hello, {name}!") print(greet.__name__) # greet print(greet.__doc__) # Здоровается с кем-нибудь
Единственная задача функции functools.wraps() - скопировать имя, строку документа, список аргументов и т.д., для предотвращения их перезаписи. И учитывая, что functools.wraps() также является декоратором, можно просто поместить его перед actual_func, и проблема решена!
functools.reduce().Функция functools.reduce() берет итерируемый объект и уменьшает (или складывает) все его значения в одно. У этого есть много разных применений, вот некоторые из них:
from functools import reduce import operator def product(iterable): return reduce(operator.mul, iterable, 1) def factorial(n): return reduce(operator.mul, range(1, n)) def sum(numbers): """Внимание! Переопределяется встроенная функция `sum()`""" return reduce(operator.add, numbers, 1) def reverse(iterable): return reduce(lambda x, y: y+x, iterable) print(product([1, 2, 3])) # 6 print(factorial(5)) # 24 print(sum([2, 6, 8, 3])) # 20 print(reverse("hello")) # olleh
Функция functools.reduce может упростить и уменьшить код в одну строку, который в противном случае был бы намного длиннее. С учетом сказанного, чрезмерное использование этой функции только ради сокращения кода обычно является плохой идеей.
Кроме того, учитывая, что использование сокращения обычно приводит к однострочникам, то это идеальный кандидат для functools.partial():
from functools import reduce, partial import operator product = partial(reduce, operator.mul) print(product([1, 2, 3])) # 6
И, наконец, если нужен не только конечный сокращенный результат, но и промежуточный, то можно использовать itertools.accumulate() - функцию из другого встроенного модуля itertools. Вот как можно его использовать для вычисления максимального рабочего времени:
from itertools import accumulate data = [3, 4, 1, 3, 5, 6, 9, 0, 1] print(list(accumulate(data, max))) # [3, 4, 4, 4, 5, 6, 9, 9, 9]