dataclass with cached field

dataclass-with-cached-field.py

import time
from dataclasses import dataclass, field
from typing import Generic, TypeVar
from collections.abc import AsyncIterable


T = TypeVar('T')


async def async_list(stream: AsyncIterable[T]) -> list[T]:
    return [item async for item in stream]


@dataclass
class Cached(Generic[T]):
    _data: T
    duration: int
    last_updated: int = None
    stream: bool = False

    def should_refresh(self) -> bool:
        if past := self.last_updated:
            return int(time.time()) - past > self.duration
        return True

    async def read(self, fn, *args, **kwargs) -> T:
        if self.should_refresh():
            res = fn(*args, **kwargs)
            if self.stream:
                self._data = await async_list(res)
            else:
                self._data = await res
            self.last_updated = int(time.time())
        return self._data

    @staticmethod
    def field(duration: int, default: T = None, stream: bool = False, **kwargs):
        return field(init=False, default_factory=lambda: Cached(_data=default, duration=duration, stream=stream), **kwargs)