Boltons retry decorator and cached property helpers under pytest: working versions (tested 2026-10-16)

boltons==26.2.0
pytest==9.1.1

import time

from boltons.cacheutils import cachedmethod, cachedproperty
from boltons.funcutils import wraps


class RetryError(Exception):
    """Raised when a retried callable exhausts every attempt."""


def retry(attempts=3, retry_on=Exception, delay=0.0, sleep=time.sleep):
    """Signature-preserving retry decorator built on boltons.funcutils.wraps.

    Calls the wrapped function up to *attempts* times, retrying only when an
    exception matching *retry_on* is raised. The final failure is re-raised as
    a RetryError whose __cause__ is the last original exception.
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last = None
            for attempt in range(attempts):
                try:
                    return func(*args, **kwargs)
                except retry_on as exc:
                    last = exc
                    if attempt < attempts - 1:
                        sleep(delay)
            raise RetryError(f"{func.__name__} failed after {attempts} attempts") from last
        return wrapper
    return decorator


class Rollup:
    """In-memory aggregator showing boltons cachedproperty and cachedmethod."""

    def __init__(self, values):
        self.values = list(values)
        self.calls = {"total": 0, "bucket": 0}
        self._buckets = {}

    @cachedproperty
    def total(self):
        self.calls["total"] += 1
        return sum(self.values)

    @cachedmethod("_buckets")
    def bucket(self, width):
        self.calls["bucket"] += 1
        return self.total // width


class Slotted:
    __slots__ = ("x",)

    def __init__(self, x):
        self.x = x

    @cachedproperty
    def doubled(self):
        return self.x * 2

import inspect

import pytest

from helpers import RetryError, Rollup, Slotted, retry


def test_retry_retries_until_success():
    calls = []

    @retry(attempts=3)
    def flaky():
        calls.append(1)
        if len(calls) < 3:
            raise OSError("transient")
        return "ok"

    assert flaky() == "ok"
    assert len(calls) == 3


def test_retry_exhausts_and_chains_cause():
    @retry(attempts=2)
    def always_fails():
        raise OSError("down")

    with pytest.raises(RetryError) as excinfo:
        always_fails()
    assert isinstance(excinfo.value.__cause__, OSError)


def test_retry_preserves_signature_via_wraps():
    @retry(attempts=1)
    def add(x, y=0):
        return x + y

    assert str(inspect.signature(add)) == "(x, y=0)"


def test_cachedproperty_caches_and_delattr_invalidates():
    r = Rollup([1, 2, 3])

    assert r.total == 6
    assert r.total == 6
    assert r.calls["total"] == 1

    del r.total
    assert r.total == 6
    assert r.calls["total"] == 2


def test_cachedproperty_needs_instance_dict():
    s = Slotted(4)
    with pytest.raises(AttributeError):
        _ = s.doubled


def test_cachedmethod_is_per_instance_and_needs_hashable_args():
    a, b = Rollup([10, 10]), Rollup([10, 10])

    assert a.bucket(4) == 5
    assert a.bucket(4) == 5
    assert a.calls["bucket"] == 1
    assert b.calls["bucket"] == 0

    b.bucket(4)
    assert b.calls["bucket"] == 1
    assert len(a._buckets) == 1 and len(b._buckets) == 1

    with pytest.raises(TypeError):
        a.bucket([4])

[pytest]
pythonpath = .