Stores

stores.base

class hamilton.caching.stores.base.MetadataStore

abstractmethod delete(cache_key: str) → None

Delete data_version keyed by cache_key.

abstractmethod delete_all() → None

Delete all stored metadata.

abstractmethod exists(cache_key: str) → bool

boolean check if a data_version is found for cache_key If True, .get() should successfully retrieve the data_version.

abstractmethod get(cache_key: str, **kwargs) → str | None

Try to retrieve data_version keyed by cache_key. If retrieval misses return None.

get_last_run() → Any

Return the metadata from the last started run.

abstractmethod get_run(run_id: str) → Sequence[dict]

Return a list of node metadata associated with a run.

For each node, the metadata should include cache_key (created or used) and data_version. These values allow to manually query the MetadataStore or ResultStore.

Decoding the cache_key gives the node_name, code_version, and dependencies_data_versions. Individual implementations may add more information or decode the cache_key before returning metadata.

abstractmethod get_run_ids() → Sequence[str]

Return a list of run ids, sorted from oldest to newest start time. A run_id is registered when the metadata_store .initialize() is called.

abstractmethod initialize(run_id: str) → None

Setup the metadata store and log the start of the run

property last_run_id: str

Return

abstractmethod set(cache_key: str, data_version: str, **kwargs) → Any | None

Store the mapping cache_key -> data_version. Can include other metadata (e.g., node name, run id, code version) depending on the implementation.

property size: int

Number of unique entries (i.e., cache_keys) in the metadata_store

exception hamilton.caching.stores.base.ResultRetrievalError

Raised by the SmartCacheAdapter when ResultStore.get() fails.

class hamilton.caching.stores.base.ResultStore

abstractmethod delete(data_version: str) → None

Delete result keyed by data_version.

abstractmethod delete_all() → None

Delete all stored results.

abstractmethod exists(data_version: str) → bool

boolean check if a result is found for data_version If True, .get() should successfully retrieve the result.

abstractmethod get(data_version: str, **kwargs) → Any | None

Try to retrieve result keyed by data_version. If retrieval misses, return None.

abstractmethod set(data_version: str, result: Any, **kwargs) → None

Store result keyed by data_version.

hamilton.caching.stores.base.search_data_adapter_registry(name: str, type_: type) → Tuple[Type[DataSaver], Type[DataLoader]]

Find pair of DataSaver and DataLoader registered with name and supporting type_

stores.file

class hamilton.caching.stores.file.FileResultStore(path: str, create_dir: bool = True)

delete(data_version: str) → None

Delete result keyed by data_version.

delete_all() → None

Delete all stored results.

exists(data_version: str) → bool

boolean check if a result is found for data_version If True, .get() should successfully retrieve the result.

get(data_version: str) → Any | None

Try to retrieve result keyed by data_version. If retrieval misses, return None.

set(data_version: str, result: Any, saver_cls: DataSaver | None = None, loader_cls: DataLoader | None = None) → None

Store result keyed by data_version.

stores.sqlite

class hamilton.caching.stores.sqlite.SQLiteMetadataStore(path: str, connection_kwargs: dict | None = None)

property connection: Connection

Connection to the SQLite database.

delete(cache_key: str) → None

Delete metadata associated with cache_key.

delete_all() → None

Delete all existing tables from the database

exists(cache_key: str) → bool

boolean check if a data_version is found for cache_key If True, .get() should successfully retrieve the data_version.

get(cache_key: str) → str | None

Try to retrieve data_version keyed by cache_key. If retrieval misses return None.

get_run(run_id: str) → List[dict]

Return a list of node metadata associated with a run.

Parameters:

run_id – ID of the run to retrieve

Returns:

List of node metadata which includes cache_key, data_version, node_name, and code_version. The list can be empty if a run was initialized but no nodes were executed.

Raises:

IndexError – if the run_id is not found in metadata store.

get_run_ids() → List[str]

Return a list of run ids, sorted from oldest to newest start time.

initialize(run_id) → None

Call initialize when starting a run. This will create database tables if necessary.

set(*, cache_key: str, data_version: str, run_id: str, node_name: str = None, code_version: str = None, **kwargs) → None

Store the mapping cache_key -> data_version. Can include other metadata (e.g., node name, run id, code version) depending on the implementation.

stores.memory

class hamilton.caching.stores.memory.InMemoryMetadataStore

delete(cache_key: str) → None

Delete the data_version for cache_key.

delete_all() → None

Delete all stored metadata.

exists(cache_key: str) → bool

Indicate if cache_key exists and it can retrieve a data_version.

get(cache_key: str) → str | None

Retrieve the data_version for cache_key.

get_run(run_id: str) → List[Dict[str, str]]

Return a list of node metadata associated with a run.

get_run_ids() → List[str]

Return a list of all run_id values stored.

initialize(run_id: str) → None

Set up and log the beginning of the run.

classmethod load_from(metadata_store: MetadataStore) → InMemoryMetadataStore

Load in-memory metadata from another MetadataStore instance.

Parameters:

metadata_store – MetadataStore instance to load from.

Returns:

InMemoryMetadataStore copy of the metadata_store.

from hamilton import driver
from hamilton.caching.stores.sqlite import SQLiteMetadataStore
from hamilton.caching.stores.memory import InMemoryMetadataStore
import my_dataflow

sqlite_metadata_store = SQLiteMetadataStore(path="./.hamilton_cache")
in_memory_metadata_store = InMemoryMetadataStore.load_from(sqlite_metadata_store)

# create the Driver with the in-memory metadata store
dr = (
  driver.Builder()
  .with_modules(my_dataflow)
  .with_cache(metadata_store=in_memory_metadata_store)
  .build()
)

persist_to(metadata_store: MetadataStore | None = None) → None

Persist in-memory metadata using another MetadataStore implementation.

Parameters:

metadata_store – MetadataStore implementation to use for persistence. If None, a SQLiteMetadataStore is created with the default path “./.hamilton_cache”.

from hamilton import driver
from hamilton.caching.stores.sqlite import SQLiteMetadataStore
from hamilton.caching.stores.memory import InMemoryMetadataStore
import my_dataflow

dr = (
  driver.Builder()
  .with_modules(my_dataflow)
  .with_cache(metadata_store=InMemoryMetadataStore())
  .build()
)

# execute the Driver several time. This will populate the in-memory metadata store
dr.execute(...)

# persist to disk in-memory metadata
dr.cache.metadata_store.persist_to(SQLiteMetadataStore(path="./.hamilton_cache"))

set(cache_key: str, data_version: str, run_id: str, **kwargs) → Any | None

Set the data_version for cache_key and associate it with the run_id.

class hamilton.caching.stores.memory.InMemoryResultStore(persist_on_exit: bool = False)

delete(data_version: str) → None

Delete result keyed by data_version.

delete_all() → None

Delete all stored results.

exists(data_version: str) → bool

boolean check if a result is found for data_version If True, .get() should successfully retrieve the result.

get(data_version: str) → Any | None

Try to retrieve result keyed by data_version. If retrieval misses, return None.

classmethod load_from(result_store: ResultStore, metadata_store: MetadataStore | None = None, data_versions: Sequence[str] | None = None) → InMemoryResultStore

Load in-memory results from another ResultStore instance.

Since result stores do not store an index of their keys, you must provide a MetadataStore instance or a list of data_version for which results should be loaded in memory.

Parameters:

• result_store – ResultStore instance to load results from.

• metadata_store – MetadataStore instance from which all data_version are retrieved.

Returns:

InMemoryResultStore copy of the result_store.

from hamilton import driver
from hamilton.caching.stores.sqlite import SQLiteMetadataStore
from hamilton.caching.stores.memory import InMemoryMetadataStore
import my_dataflow

sqlite_metadata_store = SQLiteMetadataStore(path="./.hamilton_cache")
in_memory_metadata_store = InMemoryMetadataStore.load_from(sqlite_metadata_store)

# create the Driver with the in-memory metadata store
dr = (
  driver.Builder()
  .with_modules(my_dataflow)
  .with_cache(metadata_store=in_memory_metadata_store)
  .build()
)

persist_to(result_store: ResultStore | None = None) → None

Persist in-memory results using another ResultStore implementation.

Parameters:

result_store – ResultStore implementation to use for persistence. If None, a FileResultStore is created with the default path “./.hamilton_cache”.

set(data_version: str, result: Any, **kwargs) → None

Store result keyed by data_version.