Source code for kedro.framework.session.shelvestore

"""This module implements a dict-like store object used to persist Kedro sessions.
This module is separated from to ensure it's only imported when exported explicitly.
from __future__ import annotations

import dbm
import shelve
from multiprocessing import Lock
from pathlib import Path
from typing import Any

from .store import BaseSessionStore

[docs] class ShelveStore(BaseSessionStore): """Stores the session data on disk using `shelve` package. This is an example of how to persist data on disk.""" _lock = Lock() @property def _location(self) -> Path: return Path(self._path).expanduser().resolve() / self._session_id / "store" def read(self) -> dict[str, Any]: """Read the data from disk using `shelve` package.""" data: dict[str, Any] = {} try: with, flag="r") as _sh: # noqa: S301 data = dict(_sh) except dbm.error: pass return data def save(self) -> None: """Save the data on disk using `shelve` package.""" location = self._location location.parent.mkdir(parents=True, exist_ok=True) with self._lock, as _sh: # noqa: S301 keys_to_del = _sh.keys() - for key in keys_to_del: del _sh[key] _sh.update(