Cache stores

Pass store= to Pipeline to keep step results and BatchStep items in a CacheStore instead of files under cache_dir.

All stores import from adag.cache_store.

Store	Extra
FileSystemCacheStore (default)	none
MemoryCacheStore	none
SqliteCacheStore	none
SqlCacheStore	adag[sql]
RedisCacheStore	adag[redis]

"""One SQLite file holds every step result and every batch item."""

import tempfile
from dataclasses import dataclass
from pathlib import Path

from pydantic import BaseModel

from adag import BatchItem, BatchStep, InputParam, NullObserver, Pipeline
from adag.cache_store import SqliteCacheStore


class Greeting(BaseModel):
    text: str


class Greetings(BaseModel):
    items: list[Greeting]


@dataclass(frozen=True)
class Greet(BatchStep[Greetings, Greeting]):
    items_source_keys = ("names",)

    def items_data(self, names: list[str]) -> list[str]:
        return names

    def process_item(self, item: BatchItem, names: list[str]) -> Greeting:
        print("greeting", item.data)
        return Greeting(text=f"hello {item.data}")

    def assemble(self, results: list[Greeting], names: list[str]) -> Greetings:
        return Greetings(items=results)


store = SqliteCacheStore(Path(tempfile.mkdtemp()) / "cache.db")
pipeline = Pipeline(Greet()(names=InputParam[list[str]]("names")), store=store)

pipeline.execute(observer=NullObserver(), names=["ana", "bo"])
pipeline.execute(observer=NullObserver(), names=["ana", "bo", "cy"])
print(len(list(store.iter_keys())), "keys in", store.path.name)

greeting ana
greeting bo
greeting cy
9 keys in cache.db

The second run greets only cy: ana and bo are read from the store.

The 9 keys: 3 items (greet/<key>.json), and for each of the 2 runs a result <key>, its <key>.done marker, and its key record .keys/<key>.json (see Caching).

Open a store from a URL

store_from_url(url) opens a store from text, so a remote worker can open the same store as the main process (see Dispatch).

• sqlite:///relative.db or sqlite:////absolute.db: SqliteCacheStore.
• redis://host:6379/0: RedisCacheStore.
• postgresql://... or any other SQLAlchemy URL: SqlCacheStore.
• file:///path or a plain path: FileSystemCacheStore.

Use one ItemCache directly

ItemCache also takes store= when you use it outside a pipeline:

"""ItemCache on a different backend: SQLite instead of one file per item."""

import tempfile
from pathlib import Path

from pydantic import BaseModel

from adag import ItemCache
from adag.cache_store import SqliteCacheStore


class Answer(BaseModel):
    text: str


folder = Path(tempfile.mkdtemp())
store = SqliteCacheStore(folder / "items.db")
cache = ItemCache[Answer](folder, "answers", Answer, store=store)

keys = [cache.key(question) for question in ["q1", "q2", "q3"]]
cache.save(keys[0], Answer(text="first"))

cached, missing = cache.partition(keys)
print("cached:", len(cached), "missing:", len(missing))
print(cache.load(keys[0]))

cached: 1 missing: 2
text='first'