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