Декоратор cached_property

В этой статье хочется рассмотреть декоратор cached_property. Почему он есть и в стандартной библиотеке и в Django. Чем они отличаются и когда какой лучше использовать

Проблема

Допустим у нас есть класс с 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, то можно просто скопировать код, там не очень много кода).