Session Store API Reference

class BaseSessionStore:
    """
    Base class for session stores that persist session data.
    Session stores save information about Kedro runs for tracking and reproducibility.
    """

    def __init__(self, path: str, session_id: str):
        """
        Initialize session store.

        Parameters:
        - path: Path to store location
        - session_id: Unique identifier for the session
        """

    @property
    def session_id(self) -> str:
        """
        Get the session identifier.

        Returns:
        Unique session ID string
        """

    def save(self, data: dict[str, Any]) -> None:
        """
        Save session data.

        Parameters:
        - data: Dictionary of session data to persist

        Note:
        Implementation should handle serialization and storage location
        """

    def read(self) -> dict[str, Any]:
        """
        Read session data.

        Returns:
        Dictionary of stored session data

        Raises:
        FileNotFoundError: If session data doesn't exist
        """

from kedro.framework.session.store import BaseSessionStore
from pathlib import Path
import json

class JSONSessionStore(BaseSessionStore):
    """Session store that saves data to JSON files."""

    def __init__(self, path: str, session_id: str):
        super().__init__(path, session_id)
        self._store_path = Path(path) / ".kedro" / "sessions" / f"{session_id}.json"
        self._store_path.parent.mkdir(parents=True, exist_ok=True)

    def save(self, data: dict[str, Any]) -> None:
        """Save session data to JSON file."""
        with open(self._store_path, 'w') as f:
            json.dump(data, f, indent=2, default=str)

    def read(self) -> dict[str, Any]:
        """Read session data from JSON file."""
        if not self._store_path.exists():
            raise FileNotFoundError(f"Session data not found: {self.session_id}")

        with open(self._store_path) as f:
            return json.load(f)

# src/my_project/settings.py
from my_project.stores import JSONSessionStore

SESSION_STORE_CLASS = JSONSessionStore
SESSION_STORE_ARGS = {}  # Additional constructor arguments

session_data = {
    "session_id": "2024-01-15T10.30.45.123Z",
    "project_path": "/path/to/project",
    "env": "local",
    "extra_params": {"key": "value"},
    "username": "user@example.com",
    "git_sha": "abc123def456",
    "package_name": "my_project"
}

from kedro.framework.session import KedroSession

with KedroSession.create(save_on_close=True) as session:
    # Session data is automatically saved on close
    session.run()
# Session store persists run information

from pathlib import Path
from my_project.stores import JSONSessionStore

# Load past session data
session_id = "2024-01-15T10.30.45.123Z"
store = JSONSessionStore(session_id, Path.cwd())

try:
    data = store.read()
    print(f"Session {session_id} ran in {data['env']} environment")
    print(f"Git commit: {data['git_sha']}")
except FileNotFoundError:
    print("Session not found")