Python functools cached property

Декоратор cached_property

Допустим у нас есть класс с property , которое вычислять довольно долго, но мы им пользуемся часто и не хочется вычислять его несколько раз.

import dataclasses import hashlib @dataclasses.dataclass class User: first_name: str last_name: str @property def signature(self) -> bytes: return hashlib.sha512((self.first_name + self.last_name).encode()).digest()

Наивная реализация

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

import dataclasses from typing import Optional import hashlib @dataclasses.dataclass class User: first_name: str last_name: str _signature: Optional[bytes] = dataclasses.field(init=False, repr=False, compare=False, hash=False, default=None) @property def signature(self) -> bytes: if self._signature is None: self._signature = hashlib.sha512((self.first_name + self.last_name).encode()).digest() return self._signature

И получится довольно хорошее решение. В нём есть один недостаток — нам приходится добавлять приватный метод, если у нас таких property много у класса, то у нас будет очень много атрибутов, что не очень хорошо.

Решение из модуля functools

Тогда стоит обратить внимание на cached_property в модуле functools .

Ниже представлен пример с использованием functools.cached_property

import dataclasses import functools from typing import Optional import hashlib @dataclasses.dataclass class User: first_name: str last_name: str @functools.cached_property def signature(self) -> bytes: return hashlib.sha512((self.first_name + self.last_name).encode()).digest() 

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

То есть код ниже вызовет функцию hashlib.sha512 только один раз

user = User(first_name='Andrei', last_name='Berenda') tasks = [ threading.Thread(target=lambda: user.signature) for i in range(10) ] for task in tasks: task.start() for task in tasks: task.join()

Но нужно разобраться каким образом это сделано.
Если посмотреть на реализацию, то можем увидеть, что cached_property использует локи и лок берется на весь класс, а не на объект класса. То есть мы не сможем начать выполнять параллельно несколько сигнатур для разных объектов класса.

Проблемы с functools.cached_property

Если мы в метод signature поместим запрос в базу или поход по http (то есть любую операцию, которая не блокирует GIL), мы всё равно будем ждать завершения метода, перед тем, как начать выполнять эту же функцию на другом объекте

import dataclasses import datetime import functools import time from typing import Optional import hashlib import threading @dataclasses.dataclass class User: first_name: str last_name: str _signature: Optional[bytes] = dataclasses.field(init=False, repr=False, compare=False, hash=False) @functools.cached_property def signature(self) -> bytes: time.sleep(1) return b'signed' tasks = [ threading.Thread(target=lambda: User(first_name='Andrei', last_name='Berenda').signature) for i in range(10) ] now = datetime.datetime.now() for task in tasks: task.start() for task in tasks: task.join() print('finished', datetime.datetime.now() - now)

Код выше будет выполняться больше 10 секунд (для упрощения я использовал time.sleep(1) , но можно было использовать поход в базу).

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

import dataclasses import datetime import time from typing import Optional import threading @dataclasses.dataclass class User: first_name: str last_name: str _signature: Optional[bytes] = dataclasses.field(init=False, repr=False, compare=False, hash=False, default=None) @property def signature(self) -> bytes: if self._signature is None: time.sleep(1) self._signature = b'signed' return self._signature tasks = [ threading.Thread(target=lambda: User(first_name='Andrei', last_name='Berenda').signature) for i in range(10) ] now = datetime.datetime.now() for task in tasks: task.start() for task in tasks: task.join() print('finished', datetime.datetime.now() - now)

Решение от Django

Эту особенность заметили в Django и написали свой декоратор cached_property , который не гарантирует что метод будет вызван только один раз, но работает намного быстрее в многопоточном приложении (каким и является приложение с использованием Django).

import dataclasses import datetime import threading import time from django.utils.functional import cached_property @dataclasses.dataclass class User: first_name: str last_name: str @cached_property def signature(self) -> bytes: time.sleep(1) return b'signed' tasks = [ threading.Thread(target=lambda: User(first_name='Andrei', last_name='Berenda').signature) for i in range(10) ] now = datetime.datetime.now() for task in tasks: task.start() for task in tasks: task.join() print('finished', datetime.datetime.now() - now)

Код выше будет работать примерно так же как и наше решение(будет отрабатывать за 1 секунду).

Подведение итогов

Если у нас есть функция, которую вы хотите кешировать и её вызывать несколько раз для одного и того же объекта крайне нежелательно, то в таком случае можно использовать functools.cached_property (или можно попробовать написать свой декоратор, который будет брать локи на уровне объекта, а не на уровне класса), а во всех остальных случаях я бы использовать cached_property из Django (если вы не используете django, то можно просто скопировать код, там не очень много кода).

Источник

6 Python декораторов, которые значительно упростят ваш код

Лучшая функция Python, которая применяет эту философию из «дзен Python», — это декоратор.

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

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

Болтать не буду. Давайте посмотрим на отобранные мной 6 декораторов, которые покажут вам, насколько элегантен Python.

1. @lru_cache: Ускоряем программы кэшированием

Самый простой способ ускорить работу функций Python с помощью трюков кэширования — использовать декоратор @lru_cache.

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

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

Рассмотрим интуитивно понятный пример:

import time def fibonacci(n): if n < 2: return n return fibonacci(n - 1) + fibonacci(n - 2) start_time = time.perf_counter() print(fibonacci(30)) end_time = time.perf_counter() print(f"The execution time: seconds") # The execution time: 0.18129450 seconds

Приведенная выше программа вычисляет N-ое число Фибоначчи с помощью функции Python. Это занимает много времени, поскольку при вычислении fibonacci(30) многие предыдущие числа Фибоначчи будут вычисляться много раз в процессе рекурсии.

Теперь давайте ускорим этот процесс с помощью декоратора @lru_cache:

from functools import lru_cache import time @lru_cache(maxsize=None) def fibonacci(n): if n < 2: return n return fibonacci(n - 1) + fibonacci(n - 2) start_time = time.perf_counter() print(fibonacci(30)) end_time = time.perf_counter() print(f"The execution time: seconds") # The execution time: 0.00002990 seconds

Как видно из приведенного выше кода, после использования декоратора @lru_cache мы можем получить тот же результат за 0,00002990 секунды, что намного быстрее, чем предыдущие 0,18129450 секунды.

Декоратор @lru_cache имеет параметр maxsize, который определяет максимальное количество результатов для хранения в кэше. Когда кэш заполнен и необходимо сохранить новый результат, наименее использованный результат вытесняется из кэша, чтобы освободить место для нового. Это называется стратегией наименее использованного результата (LRU).

По умолчанию maxsize установлен на 128. Если оно установлено в None, как в нашем примере, функции LRU отключены, и кэш может расти без ограничений.

2. @total_ordering: Добавляем недостающие методы сравнения

Декоратор @total_ordering из модуля functools используется для генерации недостающих методов сравнения для класса Python на основе тех, которые определены.

from functools import total_ordering @total_ordering class Student: def __init__(self, name, grade): self.name = name self.grade = grade def __eq__(self, other): return self.grade == other.grade def __lt__(self, other): return self.grade < other.grade student1 = Student("Alice", 85) student2 = Student("Bob", 75) student3 = Student("Charlie", 85) print(student1 < student2) # False print(student1 >student2) # True print(student1 == student3) # True print(student1 = student2) # True

Как видно из приведенного выше кода, в классе Student нет определений для методов ge, gt и le. Однако благодаря декоратору @total_ordering результаты наших сравнений между различными экземплярами будут правильными.

Преимущества этого декоратора очевидны:

• Он может сделать ваш код чище и сэкономить ваше время. Поскольку вам не нужно писать все методы сравнения.
• Некоторые старые классы могут не определять достаточно методов сравнения. Безопаснее добавить к нему декоратор @total_ordering для дальнейшего использования.

3. @contextmanager: Кастомный менеджер контекстов

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

В основном нам нужно просто использовать операторы with:

with open("test.txt",'w') as f: f.write("Yang is writing!")

Как показано в приведенном выше коде, мы можем открыть файл с помощью оператора with, чтобы он был закрыт автоматически после записи. Нам не нужно явно вызывать функцию f.close(), чтобы закрыть файл.

Иногда нам нужно определить индивидуальный менеджер контекста для каких-то особых требований. В этом случае декоратор @contextmanager - наш друг.

Например, следующий код реализует простой настраиваемый контекстный менеджер, который может выводить соответствующую информацию при открытии или закрытии файла.

from contextlib import contextmanager @contextmanager def file_manager(filename, mode): print("The file is opening. ") file = open(filename,mode) yield file print("The file is closing. ") file.close() with file_manager('test.txt', 'w') as f: f.write('Yang is writing!') # The file is opening. # The file is closing. 

4. @property: Настраиваем геттеры и сеттеры для классов

Геттеры и сеттеры - важные понятия в объектно-ориентированном программировании (ООП).

Для каждой переменной экземпляра класса метод getter возвращает ее значение, а метод setter устанавливает или обновляет ее значение. Учитывая это, геттеры и сеттеры также известны как аксессоры и мутаторы, соответственно.

Они используются для защиты данных от прямого и неожиданного доступа или изменения.

Различные языки ООП имеют разные механизмы для определения геттеров и сеттеров. В Python мы можем просто использовать декоратор @property.

class Student: def __init__(self): self._score = 0 @property def score(self): return self._score @score.setter def score(self, s): if 0 

from functools import cached_property class Circle: def __init__(self, radius): self.radius = radius @cached_property def area(self): return 3.14 * self.radius ** 2 circle = Circle(10) print(circle.area) # prints 314.0 print(circle.area) # returns the cached result (314.0) directly

import atexit @atexit.register def goodbye(): print("Bye bye!") print("Hello Yang!")