An in-memory `CacheAccess` implementation#

A CacheAccess that accumulates translations in process memory, keyed by ID, with a per-ID TTL. Click here to download the full script.

This is the no-pickle, stays-online alternative to Translator.load_persistent_instance(). Compared to that method, it:

has no pickle dependency – load_persistent_instance serializes the whole Translator to disk using pickle, and
keeps the Translator online, fetching unseen IDs on demand.

Unlike the on-disk example, which caches whole fetch_all tables, this caches the hot subset of IDs as they are translated – the access pattern of a typical online service. The trade-off is that the cache lives only as long as the process.

Partial hits#

CacheAccess.load() may return a PartialCacheHit – the rows it holds, plus (optionally) the IDs it vouches for. The AbstractFetcher then fetches only the uncovered IDs and merges them with the cached rows, so an accumulating by-ID cache no longer forces a full re-fetch when a single ID is missing. This cache returns None only when it holds nothing for the source (or lacks a requested placeholder).

Note

This example assumes a source is always fetched with the same placeholders. It stores one row per ID for the most recent placeholder layout and resets a source’s cache if those placeholders change. A cache that mixes placeholder sets per source would instead key rows by layout and verify coverage in load().

Design goals#

Cache individual IDs as they are translated (no fetch_all required).
Hold data in process memory.
Expire entries after a per-ID TTL, and cap the number of cached IDs per source.

Implementation#

State is a per-source record of the placeholder layout plus an id -> (timestamp, row) mapping.

The __init__ method.#

def __init__(self, ttl: float, max_ids: int = 100_000) -> None:
    super().__init__()
    self._ttl = ttl  # In seconds.
    self._max_ids = max_ids
    self._cache: dict[SourceType, _SourceCache] = {}

store() indexes each returned row by its ID. If the source’s placeholder layout changed, the cache is reset first (see the note above); oldest entries are dropped once the per-source cap is exceeded.

The InMemoryCacheAccess.store() method.#

def store(
    self,
    instr: FetchInstruction[SourceType, IdType],
    translations: PlaceholderTranslations[SourceType],
) -> None:
    source = translations.source
    sc = self._cache.get(source)
    if sc is None or sc.placeholders != translations.placeholders:
        # New source, or the placeholder layout changed (see
        # module docstring): start fresh.
        sc = _SourceCache(
            translations.placeholders,
            translations.id_pos,
            aliases=dict(translations.placeholder_aliases),
            rows={},
        )
        self._cache[source] = sc

    now = time.monotonic()
    id_pos = translations.id_pos
    for row in translations.records:
        sc.rows[row[id_pos]] = (now, tuple(row))

    self._evict(sc)

load() returns a PartialCacheHit with whatever hot rows it holds; covered defaults to those rows, so missing or expired IDs are re-fetched (and re-cached via store()). It returns None when nothing is cached for the source or a requested placeholder is missing. fetch_all requests always miss, since an accumulated cache cannot prove that it holds every ID.

The InMemoryCacheAccess.load() method.#

def load(
    self,
    instr: FetchInstruction[SourceType, IdType],
) -> PartialCacheHit[SourceType, IdType] | None:
    if instr.ids is None:
        return None  # An accumulated cache cannot prove it holds *all* IDs (fetch_all).

    sc = self._cache.get(instr.source)
    if sc is None or not set(instr.placeholders).issubset(sc.placeholders):
        # Nothing cached for this source, or we lack a requested placeholder. Let the fetcher fetch everything;
        # store() will (re)set the layout.
        return None

    deadline = time.monotonic() - self._ttl
    records = [sc.rows[id_][1] for id_ in instr.ids if id_ in sc.rows and sc.rows[id_][0] >= deadline]

    # Return whatever subset is hot; the fetcher fetches the rest at our layout and merges. `covered` is left to
    # default to the IDs in these rows, so any missing/expired IDs are re-fetched (and re-cached via store()).
    return PartialCacheHit(
        PlaceholderTranslations(
            source=instr.source,
            placeholders=sc.placeholders,
            records=records,
            id_pos=sc.id_pos,
            placeholder_aliases=dict(sc.aliases),
        )
    )

Creating a cached fetcher#

All AbstractFetcher implementations accept an optional cache_access keyword argument.

Creating a Translator with a cached fetcher.#

def create(ttl: float = 3600) -> Translator[str, str, int]:
    cache_access = InMemoryCacheAccess(ttl=ttl)
    fetcher = MemoryFetcher(
        data={"people": {1904: "Fred", 1999: "Sofia"}},
        cache_access=cache_access,
    )
    return Translator(fetcher)

Hint

To configure caching using TOML, add a [fetching.cache]-section. The type key is required; other keys are forwarded to the implementation.

Equivalent caching section of a TOML fetcher config.#

[fetching.cache]
type = "__main__.InMemoryCacheAccess"
ttl = 3600

See the Configuration page for more information.

An in-memory CacheAccess implementation#

Partial hits#

Design goals#

Implementation#

Creating a cached fetcher#

An in-memory `CacheAccess` implementation#