Async in-memory cache using twisted

deferredcache.py

from typing import TypeVar, Generic, Callable

import attr
from twisted.internet import defer


@attr.s(slots=True)
class _Result:
    pending = attr.ib()
    result = attr.ib()
    waiters = attr.ib()


K = TypeVar('K')

class DeferredCache(Generic[K]):

    def __init__(self, loader: Callable[[K], defer.Deferred]):
        self.__cache = {}
        self.loader = loader

    def get(self, key: K) -> defer.Deferred:
        cached = self.__cache.get(key)
        if cached is None:
            cached = _Result(defer.maybeDeferred(self.loader, key), None, [])
            self.__cache[key] = cached

            def onload(result):
                cached.pending = None
                cached.result = result
                dfrs, cached.waiters = cached.waiters, []
                for dfr in dfrs:
                    dfr.callback(result)

            def onfail(failure):
                for dfr in cached.waiters:
                    dfr.errback(failure)
                self.__cache.pop(key, None)

            cached.pending.addCallbacks(onload, onfail)

        if cached.pending is not None:
            d = defer.Deferred()
            cached.waiters.append(d)
            return d
        else:
            return defer.succeed(cached.result)

    def clear(self, key: K):
        self.__cache.pop(key, None)