Skip to content

API Reference

Decorator

stet._decorator.once(func=None, *, store='_stet_store.csv', key=None)

Decorator that skips already-completed experiment runs.

Tracks which parameter combinations have been executed and skips them on subsequent calls. The function's return value is not saved; only whether the combination has been run.

Can be used in three styles:

Style 1 - zero config::

@once
def run_experiment(alpha, beta):
    ...

Style 2 - specify store::

@once(store='_stet_store.csv')
def run_experiment(alpha, beta):
    ...

Style 3 - full control::

@once(store='_stet_store.csv', key=['alpha', 'beta'])
def run_experiment(alpha, beta, n_steps):
    ...

Parameters:

Name Type Description Default
func Callable[..., Any] | None

The function to decorate (when used without parentheses).

 None
store str | Path

Path to the store file. Backend is inferred from extension: .csv, .parquet, .sqlite, .json. Defaults to _stet_store.csv in the current directory.

 '_stet_store.csv'
key list[str] | None

Parameter names that uniquely identify a run. If omitted, all parameters are used.

 None

Returns:

Type Description
Any

The decorated function (or a decorator if called with arguments).
Example 
@once(store='_stet_store.csv', key=['alpha', 'seed'])
def run_experiment(alpha, seed, n_iter=1000):
    pass

run_experiment(alpha=0.1, seed=42)
run_experiment(alpha=0.1, seed=42)
Source code in src/stet/_decorator.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def once(
    func: Callable[..., Any] | None = None,
    *,
    store: str | Path = "_stet_store.csv",
    key: list[str] | None = None,
) -> Any:
    """Decorator that skips already-completed experiment runs.

    Tracks which parameter combinations have been executed and skips them
    on subsequent calls. The function's return value is not saved; only
    whether the combination has been run.

    Can be used in three styles:

    Style 1 - zero config::

        @once
        def run_experiment(alpha, beta):
            ...

    Style 2 - specify store::

        @once(store='_stet_store.csv')
        def run_experiment(alpha, beta):
            ...

    Style 3 - full control::

        @once(store='_stet_store.csv', key=['alpha', 'beta'])
        def run_experiment(alpha, beta, n_steps):
            ...

    Args:
        func: The function to decorate (when used without parentheses).
        store: Path to the store file. Backend is inferred from extension:
            ``.csv``, ``.parquet``, ``.sqlite``, ``.json``.
            Defaults to ``_stet_store.csv`` in the current directory.
        key: Parameter names that uniquely identify a run. If omitted,
            all parameters are used.

    Returns:
        The decorated function (or a decorator if called with arguments).

    Example:
        ```python
        @once(store='_stet_store.csv', key=['alpha', 'seed'])
        def run_experiment(alpha, seed, n_iter=1000):
            pass

        run_experiment(alpha=0.1, seed=42)
        run_experiment(alpha=0.1, seed=42)
        ```
    """
    decorator = _OnceDecorator(store=store, key=key)

    if func is not None:
        return decorator(func)

    return decorator

Utility functions

stet._utils.status(store_path=_DEFAULT_STORE)

Print a summary of completed experiments in a store.

Parameters:

Name Type Description Default
store_path str | Path

Path to the store file (any supported extension). Defaults to _stet_store.csv in the current directory.

 _DEFAULT_STORE

Raises:

Type Description
FileNotFoundError

If the store file does not exist.
Example 
stet.status()
# [stet] Store: _stet_store.csv
# [stet] 42 completed experiments recorded
# [stet] Last run: 2024-11-03T14:22:01
# [stet] Key columns: alpha, beta
Source code in src/stet/_utils.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def status(store_path: str | Path = _DEFAULT_STORE) -> None:
    """Print a summary of completed experiments in a store.

    Args:
        store_path: Path to the store file (any supported extension).
            Defaults to ``_stet_store.csv`` in the current directory.

    Raises:
        FileNotFoundError: If the store file does not exist.

    Example:
        ```python
        stet.status()
        # [stet] Store: _stet_store.csv
        # [stet] 42 completed experiments recorded
        # [stet] Last run: 2024-11-03T14:22:01
        # [stet] Key columns: alpha, beta
        ```
    """
    path = Path(store_path)
    if not path.exists():
        raise FileNotFoundError(f"Store not found: {path}")

    backend = get_backend(path)
    records = backend.load()

    print(f"[stet] Store: {path}")
    print(f"[stet] {len(records)} completed experiments recorded")

    if records:
        timestamps = [
            r["_stet_timestamp"]
            for r in records
            if "_stet_timestamp" in r and r["_stet_timestamp"]
        ]
        if timestamps:
            print(f"[stet] Last run: {max(timestamps)}")

        key_cols = [k for k in records[0] if k != "_stet_timestamp"]
        if key_cols:
            print(f"[stet] Key columns: {', '.join(key_cols)}")

stet._utils.reset(store_path=_DEFAULT_STORE, key_dict=None)

Remove entries from a store.

Parameters:

Name Type Description Default
store_path str | Path

Path to the store file. Defaults to _stet_store.csv in the current directory.

 _DEFAULT_STORE
key_dict dict[str, Any] | None

If provided, remove only this specific key combination. If None, clear the entire store (prompts for confirmation in interactive mode).

 None
Example 
stet.reset(key_dict={'alpha': '0.1', 'beta': '2'})
stet.reset()
stet.reset('markov_runs.csv')
Source code in src/stet/_utils.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def reset(
    store_path: str | Path = _DEFAULT_STORE,
    key_dict: dict[str, Any] | None = None,
) -> None:
    """Remove entries from a store.

    Args:
        store_path: Path to the store file. Defaults to ``_stet_store.csv``
            in the current directory.
        key_dict: If provided, remove only this specific key combination.
            If None, clear the entire store (prompts for confirmation
            in interactive mode).

    Example:
        ```python
        stet.reset(key_dict={'alpha': '0.1', 'beta': '2'})
        stet.reset()
        stet.reset('markov_runs.csv')
        ```
    """
    path = Path(store_path)
    backend = get_backend(path)

    if key_dict is None:
        if sys.stdin.isatty():
            answer = (
                input(f"[stet] Clear all records in {path}? [y/N] ").strip().lower()
            )
            if answer != "y":
                print("[stet] Aborted.")
                return
        backend.clear()
        print(f"[stet] Store cleared: {path}")
    else:
        backend.remove(key_dict)
        print(f"[stet] Removed entry: {key_dict}")

Backends

stet.backends._base.BaseBackend

Bases: ABC

Abstract base class defining the backend interface.

All backends must implement :meth:has, :meth:record, :meth:load, :meth:remove, and :meth:clear.

Parameters:

Name Type Description Default
path Path

Path to the store file.

 required
Source code in src/stet/backends/_base.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class BaseBackend(ABC):
    """Abstract base class defining the backend interface.

    All backends must implement :meth:`has`, :meth:`record`,
    :meth:`load`, :meth:`remove`, and :meth:`clear`.

    Args:
        path: Path to the store file.
    """

    def __init__(self, path: Path) -> None:
        self.path = path

    @abstractmethod
    def has(self, key_dict: dict[str, Any]) -> bool:
        """Check if a key combination exists in the store.

        Args:
            key_dict: Parameter names and values to look up.

        Returns:
            True if this combination has been recorded, False otherwise.
        """

    @abstractmethod
    def record(self, key_dict: dict[str, Any]) -> None:
        """Write a key combination to the store.

        Adds ``_stet_timestamp`` with the current UTC time.

        Args:
            key_dict: Parameter names and values to record.
        """

    @abstractmethod
    def load(self) -> list[dict[str, Any]]:
        """Return all stored records.

        Returns:
            List of dicts, each representing one recorded experiment run.
        """

    @abstractmethod
    def remove(self, key_dict: dict[str, Any]) -> None:
        """Remove a specific key combination from the store.

        Args:
            key_dict: The exact key combination to remove.
        """

    @abstractmethod
    def clear(self) -> None:
        """Remove all records from the store."""

clear() abstractmethod

Remove all records from the store.

Source code in src/stet/backends/_base.py 
60
61
62
@abstractmethod
def clear(self) -> None:
    """Remove all records from the store."""

has(key_dict) abstractmethod

Check if a key combination exists in the store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to look up.

 required

Returns:

Type Description
bool

True if this combination has been recorded, False otherwise.
Source code in src/stet/backends/_base.py 
23
24
25
26
27
28
29
30
31
32
@abstractmethod
def has(self, key_dict: dict[str, Any]) -> bool:
    """Check if a key combination exists in the store.

    Args:
        key_dict: Parameter names and values to look up.

    Returns:
        True if this combination has been recorded, False otherwise.
    """

load() abstractmethod

Return all stored records.

Returns:

Type Description
list[dict[str, Any]]

List of dicts, each representing one recorded experiment run.
Source code in src/stet/backends/_base.py 
44
45
46
47
48
49
50
@abstractmethod
def load(self) -> list[dict[str, Any]]:
    """Return all stored records.

    Returns:
        List of dicts, each representing one recorded experiment run.
    """

record(key_dict) abstractmethod

Write a key combination to the store.

Adds _stet_timestamp with the current UTC time.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to record.

 required
Source code in src/stet/backends/_base.py 
34
35
36
37
38
39
40
41
42
@abstractmethod
def record(self, key_dict: dict[str, Any]) -> None:
    """Write a key combination to the store.

    Adds ``_stet_timestamp`` with the current UTC time.

    Args:
        key_dict: Parameter names and values to record.
    """

remove(key_dict) abstractmethod

Remove a specific key combination from the store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

The exact key combination to remove.

 required
Source code in src/stet/backends/_base.py 
52
53
54
55
56
57
58
@abstractmethod
def remove(self, key_dict: dict[str, Any]) -> None:
    """Remove a specific key combination from the store.

    Args:
        key_dict: The exact key combination to remove.
    """

stet.backends._csv.CsvBackend

Bases: BaseBackend

Store experiment records in a CSV file using pandas.

Parameters:

Name Type Description Default
path Path

Path to the .csv file. Created on first write.

 required
Example 
from stet.backends import CsvBackend
from pathlib import Path

backend = CsvBackend(Path('_stet_store.csv'))
backend.record({'alpha': 0.1, 'beta': 2})
backend.has({'alpha': 0.1, 'beta': 2})
Source code in src/stet/backends/_csv.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class CsvBackend(BaseBackend):
    """Store experiment records in a CSV file using pandas.

    Args:
        path: Path to the ``.csv`` file. Created on first write.

    Example:
        ```python
        from stet.backends import CsvBackend
        from pathlib import Path

        backend = CsvBackend(Path('_stet_store.csv'))
        backend.record({'alpha': 0.1, 'beta': 2})
        backend.has({'alpha': 0.1, 'beta': 2})
        ```
    """

    def __init__(self, path: Path) -> None:
        super().__init__(path)
        self._lock = filelock.FileLock(str(path) + ".lock")

    def _read(self) -> pd.DataFrame:
        if not self.path.exists():
            return pd.DataFrame()
        return pd.read_csv(self.path, dtype=str)

    def has(self, key_dict: dict[str, Any]) -> bool:
        """Check if key combination exists in the CSV store.

        Args:
            key_dict: Parameter names and values to look up.

        Returns:
            True if this combination has been recorded, False otherwise.
        """
        with self._lock:
            df = self._read()
        if df.empty:
            return False
        cols = list(key_dict.keys())
        missing = [c for c in cols if c not in df.columns]
        if missing:
            return False
        mask = pd.Series([True] * len(df), dtype=bool)
        for col, val in key_dict.items():
            mask = mask & (df[col] == str(val))
        return bool(mask.any())

    def record(self, key_dict: dict[str, Any]) -> None:
        """Append a key combination to the CSV store.

        Args:
            key_dict: Parameter names and values to record.
        """
        row = {k: str(v) for k, v in key_dict.items()}
        row["_stet_timestamp"] = datetime.datetime.now(datetime.UTC).isoformat()
        with self._lock:
            df = self._read()
            new_row = pd.DataFrame([row])
            df = pd.concat([df, new_row], ignore_index=True)
            df.to_csv(self.path, index=False)

    def load(self) -> list[dict[str, Any]]:
        """Return all records from the CSV store.

        Returns:
            List of dicts, each representing one recorded experiment run.
        """
        with self._lock:
            df = self._read()
        if df.empty:
            return []
        return df.to_dict(orient="records")

    def remove(self, key_dict: dict[str, Any]) -> None:
        """Remove a specific key combination from the CSV store.

        Args:
            key_dict: The exact key combination to remove.
        """
        with self._lock:
            df = self._read()
            if df.empty:
                return
            mask = pd.Series([True] * len(df), dtype=bool)
            for col, val in key_dict.items():
                if col in df.columns:
                    mask = mask & (df[col] == str(val))
            df = df[~mask]
            df.to_csv(self.path, index=False)

    def clear(self) -> None:
        """Remove all records from the CSV store."""
        with self._lock:
            if self.path.exists():
                self.path.unlink()

clear()

Remove all records from the CSV store.

Source code in src/stet/backends/_csv.py 
106
107
108
109
110
def clear(self) -> None:
    """Remove all records from the CSV store."""
    with self._lock:
        if self.path.exists():
            self.path.unlink()

has(key_dict)

Check if key combination exists in the CSV store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to look up.

 required

Returns:

Type Description
bool

True if this combination has been recorded, False otherwise.
Source code in src/stet/backends/_csv.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def has(self, key_dict: dict[str, Any]) -> bool:
    """Check if key combination exists in the CSV store.

    Args:
        key_dict: Parameter names and values to look up.

    Returns:
        True if this combination has been recorded, False otherwise.
    """
    with self._lock:
        df = self._read()
    if df.empty:
        return False
    cols = list(key_dict.keys())
    missing = [c for c in cols if c not in df.columns]
    if missing:
        return False
    mask = pd.Series([True] * len(df), dtype=bool)
    for col, val in key_dict.items():
        mask = mask & (df[col] == str(val))
    return bool(mask.any())

load()

Return all records from the CSV store.

Returns:

Type Description
list[dict[str, Any]]

List of dicts, each representing one recorded experiment run.
Source code in src/stet/backends/_csv.py 
77
78
79
80
81
82
83
84
85
86
87
def load(self) -> list[dict[str, Any]]:
    """Return all records from the CSV store.

    Returns:
        List of dicts, each representing one recorded experiment run.
    """
    with self._lock:
        df = self._read()
    if df.empty:
        return []
    return df.to_dict(orient="records")

record(key_dict)

Append a key combination to the CSV store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to record.

 required
Source code in src/stet/backends/_csv.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def record(self, key_dict: dict[str, Any]) -> None:
    """Append a key combination to the CSV store.

    Args:
        key_dict: Parameter names and values to record.
    """
    row = {k: str(v) for k, v in key_dict.items()}
    row["_stet_timestamp"] = datetime.datetime.now(datetime.UTC).isoformat()
    with self._lock:
        df = self._read()
        new_row = pd.DataFrame([row])
        df = pd.concat([df, new_row], ignore_index=True)
        df.to_csv(self.path, index=False)

remove(key_dict)

Remove a specific key combination from the CSV store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

The exact key combination to remove.

 required
Source code in src/stet/backends/_csv.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def remove(self, key_dict: dict[str, Any]) -> None:
    """Remove a specific key combination from the CSV store.

    Args:
        key_dict: The exact key combination to remove.
    """
    with self._lock:
        df = self._read()
        if df.empty:
            return
        mask = pd.Series([True] * len(df), dtype=bool)
        for col, val in key_dict.items():
            if col in df.columns:
                mask = mask & (df[col] == str(val))
        df = df[~mask]
        df.to_csv(self.path, index=False)

stet.backends._json.JsonBackend

Bases: BaseBackend

Store experiment records in a JSON file.

Uses Python's built-in :mod:json. No extra dependencies required.

Parameters:

Name Type Description Default
path Path

Path to the .json file. Created on first write.

 required
Example 
from stet.backends._json import JsonBackend
from pathlib import Path

backend = JsonBackend(Path('_stet_store.json'))
backend.record({'alpha': 0.1, 'beta': 2})
backend.has({'alpha': 0.1, 'beta': 2})
Source code in src/stet/backends/_json.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class JsonBackend(BaseBackend):
    """Store experiment records in a JSON file.

    Uses Python's built-in :mod:`json`. No extra dependencies required.

    Args:
        path: Path to the ``.json`` file. Created on first write.

    Example:
        ```python
        from stet.backends._json import JsonBackend
        from pathlib import Path

        backend = JsonBackend(Path('_stet_store.json'))
        backend.record({'alpha': 0.1, 'beta': 2})
        backend.has({'alpha': 0.1, 'beta': 2})
        ```
    """

    def __init__(self, path: Path) -> None:
        super().__init__(path)
        self._lock = filelock.FileLock(str(path) + ".lock")

    def _read(self) -> list[dict[str, Any]]:
        if not self.path.exists():
            return []
        with open(self.path) as f:
            data = json.load(f)
        return data if isinstance(data, list) else []

    def _write(self, records: list[dict[str, Any]]) -> None:
        with open(self.path, "w") as f:
            json.dump(records, f, indent=2, default=str)

    def has(self, key_dict: dict[str, Any]) -> bool:
        """Check if key combination exists in the JSON store.

        Args:
            key_dict: Parameter names and values to look up.

        Returns:
            True if this combination has been recorded, False otherwise.
        """
        with self._lock:
            records = self._read()
        str_key = {k: str(v) for k, v in key_dict.items()}
        for rec in records:
            if all(str(rec.get(k)) == v for k, v in str_key.items()):
                return True
        return False

    def record(self, key_dict: dict[str, Any]) -> None:
        """Append a key combination to the JSON store.

        Args:
            key_dict: Parameter names and values to record.
        """
        row: dict[str, Any] = {k: str(v) for k, v in key_dict.items()}
        row["_stet_timestamp"] = datetime.datetime.now(datetime.UTC).isoformat()
        with self._lock:
            records = self._read()
            records.append(row)
            self._write(records)

    def load(self) -> list[dict[str, Any]]:
        """Return all records from the JSON store.

        Returns:
            List of dicts, each representing one recorded experiment run.
        """
        with self._lock:
            return self._read()

    def remove(self, key_dict: dict[str, Any]) -> None:
        """Remove a specific key combination from the JSON store.

        Args:
            key_dict: The exact key combination to remove.
        """
        str_key = {k: str(v) for k, v in key_dict.items()}
        with self._lock:
            records = self._read()
            records = [
                r
                for r in records
                if not all(str(r.get(k)) == v for k, v in str_key.items())
            ]
            self._write(records)

    def clear(self) -> None:
        """Remove all records from the JSON store."""
        with self._lock:
            if self.path.exists():
                self.path.unlink()

clear()

Remove all records from the JSON store.

Source code in src/stet/backends/_json.py 
104
105
106
107
108
def clear(self) -> None:
    """Remove all records from the JSON store."""
    with self._lock:
        if self.path.exists():
            self.path.unlink()

has(key_dict)

Check if key combination exists in the JSON store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to look up.

 required

Returns:

Type Description
bool

True if this combination has been recorded, False otherwise.
Source code in src/stet/backends/_json.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def has(self, key_dict: dict[str, Any]) -> bool:
    """Check if key combination exists in the JSON store.

    Args:
        key_dict: Parameter names and values to look up.

    Returns:
        True if this combination has been recorded, False otherwise.
    """
    with self._lock:
        records = self._read()
    str_key = {k: str(v) for k, v in key_dict.items()}
    for rec in records:
        if all(str(rec.get(k)) == v for k, v in str_key.items()):
            return True
    return False

load()

Return all records from the JSON store.

Returns:

Type Description
list[dict[str, Any]]

List of dicts, each representing one recorded experiment run.
Source code in src/stet/backends/_json.py 
79
80
81
82
83
84
85
86
def load(self) -> list[dict[str, Any]]:
    """Return all records from the JSON store.

    Returns:
        List of dicts, each representing one recorded experiment run.
    """
    with self._lock:
        return self._read()

record(key_dict)

Append a key combination to the JSON store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to record.

 required
Source code in src/stet/backends/_json.py 
66
67
68
69
70
71
72
73
74
75
76
77
def record(self, key_dict: dict[str, Any]) -> None:
    """Append a key combination to the JSON store.

    Args:
        key_dict: Parameter names and values to record.
    """
    row: dict[str, Any] = {k: str(v) for k, v in key_dict.items()}
    row["_stet_timestamp"] = datetime.datetime.now(datetime.UTC).isoformat()
    with self._lock:
        records = self._read()
        records.append(row)
        self._write(records)

remove(key_dict)

Remove a specific key combination from the JSON store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

The exact key combination to remove.

 required
Source code in src/stet/backends/_json.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def remove(self, key_dict: dict[str, Any]) -> None:
    """Remove a specific key combination from the JSON store.

    Args:
        key_dict: The exact key combination to remove.
    """
    str_key = {k: str(v) for k, v in key_dict.items()}
    with self._lock:
        records = self._read()
        records = [
            r
            for r in records
            if not all(str(r.get(k)) == v for k, v in str_key.items())
        ]
        self._write(records)

stet.backends._sqlite.SqliteBackend

Bases: BaseBackend

Store experiment records in a SQLite database.

Uses Python's built-in :mod:sqlite3. No extra dependencies required.

Parameters:

Name Type Description Default
path Path

Path to the .sqlite or .db file. Created on first write.

 required
Example 
from stet.backends._sqlite import SqliteBackend
from pathlib import Path

backend = SqliteBackend(Path('_stet_store.sqlite'))
backend.record({'alpha': 0.1, 'beta': 2})
backend.has({'alpha': 0.1, 'beta': 2})
Source code in src/stet/backends/_sqlite.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class SqliteBackend(BaseBackend):
    """Store experiment records in a SQLite database.

    Uses Python's built-in :mod:`sqlite3`. No extra dependencies required.

    Args:
        path: Path to the ``.sqlite`` or ``.db`` file. Created on first write.

    Example:
        ```python
        from stet.backends._sqlite import SqliteBackend
        from pathlib import Path

        backend = SqliteBackend(Path('_stet_store.sqlite'))
        backend.record({'alpha': 0.1, 'beta': 2})
        backend.has({'alpha': 0.1, 'beta': 2})
        ```
    """

    def __init__(self, path: Path) -> None:
        super().__init__(path)
        self._lock = filelock.FileLock(str(path) + ".lock")

    def _connect(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.path)
        conn.row_factory = sqlite3.Row
        return conn

    def _ensure_table(self, conn: sqlite3.Connection, columns: list[str]) -> None:
        cols_sql = ", ".join(f'"{c}" TEXT' for c in [*columns, "_stet_timestamp"])
        conn.execute(f"CREATE TABLE IF NOT EXISTS {_TABLE} ({cols_sql})")
        existing = {row[1] for row in conn.execute(f"PRAGMA table_info({_TABLE})")}
        for col in [*columns, "_stet_timestamp"]:
            if col not in existing:
                conn.execute(f'ALTER TABLE {_TABLE} ADD COLUMN "{col}" TEXT')

    def has(self, key_dict: dict[str, Any]) -> bool:
        """Check if key combination exists in the SQLite store.

        Args:
            key_dict: Parameter names and values to look up.

        Returns:
            True if this combination has been recorded, False otherwise.
        """
        with self._lock:
            if not self.path.exists():
                return False
            with self._connect() as conn:
                try:
                    where = " AND ".join(f'"{k}" = ?' for k in key_dict)
                    vals = [str(v) for v in key_dict.values()]
                    row = conn.execute(
                        f"SELECT 1 FROM {_TABLE} WHERE {where} LIMIT 1", vals
                    ).fetchone()
                    return row is not None
                except sqlite3.OperationalError:
                    return False

    def record(self, key_dict: dict[str, Any]) -> None:
        """Insert a key combination into the SQLite store.

        Args:
            key_dict: Parameter names and values to record.
        """
        row = {k: str(v) for k, v in key_dict.items()}
        row["_stet_timestamp"] = datetime.datetime.now(datetime.UTC).isoformat()
        with self._lock:
            with self._connect() as conn:
                self._ensure_table(conn, list(key_dict.keys()))
                cols = ", ".join(f'"{c}"' for c in row)
                placeholders = ", ".join("?" for _ in row)
                conn.execute(
                    f"INSERT INTO {_TABLE} ({cols}) VALUES ({placeholders})",
                    list(row.values()),
                )

    def load(self) -> list[dict[str, Any]]:
        """Return all records from the SQLite store.

        Returns:
            List of dicts, each representing one recorded experiment run.
        """
        with self._lock:
            if not self.path.exists():
                return []
            with self._connect() as conn:
                try:
                    rows = conn.execute(f"SELECT * FROM {_TABLE}").fetchall()
                    return [dict(r) for r in rows]
                except sqlite3.OperationalError:
                    return []

    def remove(self, key_dict: dict[str, Any]) -> None:
        """Remove a specific key combination from the SQLite store.

        Args:
            key_dict: The exact key combination to remove.
        """
        with self._lock:
            if not self.path.exists():
                return
            with self._connect() as conn:
                try:
                    where = " AND ".join(f'"{k}" = ?' for k in key_dict)
                    vals = [str(v) for v in key_dict.values()]
                    conn.execute(f"DELETE FROM {_TABLE} WHERE {where}", vals)
                except sqlite3.OperationalError:
                    pass

    def clear(self) -> None:
        """Remove all records from the SQLite store."""
        with self._lock:
            if not self.path.exists():
                return
            with self._connect() as conn:
                conn.execute(f"DROP TABLE IF EXISTS {_TABLE}")

clear()

Remove all records from the SQLite store.

Source code in src/stet/backends/_sqlite.py 
127
128
129
130
131
132
133
def clear(self) -> None:
    """Remove all records from the SQLite store."""
    with self._lock:
        if not self.path.exists():
            return
        with self._connect() as conn:
            conn.execute(f"DROP TABLE IF EXISTS {_TABLE}")

has(key_dict)

Check if key combination exists in the SQLite store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to look up.

 required

Returns:

Type Description
bool

True if this combination has been recorded, False otherwise.
Source code in src/stet/backends/_sqlite.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def has(self, key_dict: dict[str, Any]) -> bool:
    """Check if key combination exists in the SQLite store.

    Args:
        key_dict: Parameter names and values to look up.

    Returns:
        True if this combination has been recorded, False otherwise.
    """
    with self._lock:
        if not self.path.exists():
            return False
        with self._connect() as conn:
            try:
                where = " AND ".join(f'"{k}" = ?' for k in key_dict)
                vals = [str(v) for v in key_dict.values()]
                row = conn.execute(
                    f"SELECT 1 FROM {_TABLE} WHERE {where} LIMIT 1", vals
                ).fetchone()
                return row is not None
            except sqlite3.OperationalError:
                return False

load()

Return all records from the SQLite store.

Returns:

Type Description
list[dict[str, Any]]

List of dicts, each representing one recorded experiment run.
Source code in src/stet/backends/_sqlite.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def load(self) -> list[dict[str, Any]]:
    """Return all records from the SQLite store.

    Returns:
        List of dicts, each representing one recorded experiment run.
    """
    with self._lock:
        if not self.path.exists():
            return []
        with self._connect() as conn:
            try:
                rows = conn.execute(f"SELECT * FROM {_TABLE}").fetchall()
                return [dict(r) for r in rows]
            except sqlite3.OperationalError:
                return []

record(key_dict)

Insert a key combination into the SQLite store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to record.

 required
Source code in src/stet/backends/_sqlite.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def record(self, key_dict: dict[str, Any]) -> None:
    """Insert a key combination into the SQLite store.

    Args:
        key_dict: Parameter names and values to record.
    """
    row = {k: str(v) for k, v in key_dict.items()}
    row["_stet_timestamp"] = datetime.datetime.now(datetime.UTC).isoformat()
    with self._lock:
        with self._connect() as conn:
            self._ensure_table(conn, list(key_dict.keys()))
            cols = ", ".join(f'"{c}"' for c in row)
            placeholders = ", ".join("?" for _ in row)
            conn.execute(
                f"INSERT INTO {_TABLE} ({cols}) VALUES ({placeholders})",
                list(row.values()),
            )

remove(key_dict)

Remove a specific key combination from the SQLite store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

The exact key combination to remove.

 required
Source code in src/stet/backends/_sqlite.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def remove(self, key_dict: dict[str, Any]) -> None:
    """Remove a specific key combination from the SQLite store.

    Args:
        key_dict: The exact key combination to remove.
    """
    with self._lock:
        if not self.path.exists():
            return
        with self._connect() as conn:
            try:
                where = " AND ".join(f'"{k}" = ?' for k in key_dict)
                vals = [str(v) for v in key_dict.values()]
                conn.execute(f"DELETE FROM {_TABLE} WHERE {where}", vals)
            except sqlite3.OperationalError:
                pass

stet.backends._parquet.ParquetBackend

Bases: BaseBackend

Store experiment records in a Parquet file using pandas and pyarrow.

Requires the parquet extra: uv add stet[parquet].

Parameters:

Name Type Description Default
path Path

Path to the .parquet file. Created on first write.

 required
Example 
from stet.backends._parquet import ParquetBackend
from pathlib import Path

backend = ParquetBackend(Path('_stet_store.parquet'))
backend.record({'alpha': 0.1, 'beta': 2})
backend.has({'alpha': 0.1, 'beta': 2})
Source code in src/stet/backends/_parquet.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class ParquetBackend(BaseBackend):
    """Store experiment records in a Parquet file using pandas and pyarrow.

    Requires the ``parquet`` extra: ``uv add stet[parquet]``.

    Args:
        path: Path to the ``.parquet`` file. Created on first write.

    Example:
        ```python
        from stet.backends._parquet import ParquetBackend
        from pathlib import Path

        backend = ParquetBackend(Path('_stet_store.parquet'))
        backend.record({'alpha': 0.1, 'beta': 2})
        backend.has({'alpha': 0.1, 'beta': 2})
        ```
    """

    def __init__(self, path: Path) -> None:
        super().__init__(path)
        self._lock = filelock.FileLock(str(path) + ".lock")

    def _read(self) -> Any:
        import pandas as pd

        if not self.path.exists():
            return pd.DataFrame()
        return pd.read_parquet(self.path)

    def has(self, key_dict: dict[str, Any]) -> bool:
        """Check if key combination exists in the Parquet store.

        Args:
            key_dict: Parameter names and values to look up.

        Returns:
            True if this combination has been recorded, False otherwise.
        """
        import pandas as pd

        with self._lock:
            df = self._read()
        if df.empty:
            return False
        mask = pd.Series([True] * len(df), dtype=bool)
        for col, val in key_dict.items():
            if col not in df.columns:
                return False
            mask = mask & (df[col].astype(str) == str(val))
        return bool(mask.any())

    def record(self, key_dict: dict[str, Any]) -> None:
        """Append a key combination to the Parquet store.

        Args:
            key_dict: Parameter names and values to record.
        """
        import pandas as pd

        row = {k: str(v) for k, v in key_dict.items()}
        row["_stet_timestamp"] = datetime.datetime.now(datetime.UTC).isoformat()
        with self._lock:
            df = self._read()
            new_row = pd.DataFrame([row])
            df = pd.concat([df, new_row], ignore_index=True)
            df.to_parquet(self.path, index=False)

    def load(self) -> list[dict[str, Any]]:
        """Return all records from the Parquet store.

        Returns:
            List of dicts, each representing one recorded experiment run.
        """
        with self._lock:
            df = self._read()
        if df.empty:
            return []
        return df.to_dict(orient="records")

    def remove(self, key_dict: dict[str, Any]) -> None:
        """Remove a specific key combination from the Parquet store.

        Args:
            key_dict: The exact key combination to remove.
        """
        import pandas as pd

        with self._lock:
            df = self._read()
            if df.empty:
                return
            mask = pd.Series([True] * len(df), dtype=bool)
            for col, val in key_dict.items():
                if col in df.columns:
                    mask = mask & (df[col].astype(str) == str(val))
            df = df[~mask]
            df.to_parquet(self.path, index=False)

    def clear(self) -> None:
        """Remove all records from the Parquet store."""
        with self._lock:
            if self.path.exists():
                self.path.unlink()

clear()

Remove all records from the Parquet store.

Source code in src/stet/backends/_parquet.py 
113
114
115
116
117
def clear(self) -> None:
    """Remove all records from the Parquet store."""
    with self._lock:
        if self.path.exists():
            self.path.unlink()

has(key_dict)

Check if key combination exists in the Parquet store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to look up.

 required

Returns:

Type Description
bool

True if this combination has been recorded, False otherwise.
Source code in src/stet/backends/_parquet.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def has(self, key_dict: dict[str, Any]) -> bool:
    """Check if key combination exists in the Parquet store.

    Args:
        key_dict: Parameter names and values to look up.

    Returns:
        True if this combination has been recorded, False otherwise.
    """
    import pandas as pd

    with self._lock:
        df = self._read()
    if df.empty:
        return False
    mask = pd.Series([True] * len(df), dtype=bool)
    for col, val in key_dict.items():
        if col not in df.columns:
            return False
        mask = mask & (df[col].astype(str) == str(val))
    return bool(mask.any())

load()

Return all records from the Parquet store.

Returns:

Type Description
list[dict[str, Any]]

List of dicts, each representing one recorded experiment run.
Source code in src/stet/backends/_parquet.py 
82
83
84
85
86
87
88
89
90
91
92
def load(self) -> list[dict[str, Any]]:
    """Return all records from the Parquet store.

    Returns:
        List of dicts, each representing one recorded experiment run.
    """
    with self._lock:
        df = self._read()
    if df.empty:
        return []
    return df.to_dict(orient="records")

record(key_dict)

Append a key combination to the Parquet store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

Parameter names and values to record.

 required
Source code in src/stet/backends/_parquet.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def record(self, key_dict: dict[str, Any]) -> None:
    """Append a key combination to the Parquet store.

    Args:
        key_dict: Parameter names and values to record.
    """
    import pandas as pd

    row = {k: str(v) for k, v in key_dict.items()}
    row["_stet_timestamp"] = datetime.datetime.now(datetime.UTC).isoformat()
    with self._lock:
        df = self._read()
        new_row = pd.DataFrame([row])
        df = pd.concat([df, new_row], ignore_index=True)
        df.to_parquet(self.path, index=False)

remove(key_dict)

Remove a specific key combination from the Parquet store.

Parameters:

Name Type Description Default
key_dict dict[str, Any]

The exact key combination to remove.

 required
Source code in src/stet/backends/_parquet.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def remove(self, key_dict: dict[str, Any]) -> None:
    """Remove a specific key combination from the Parquet store.

    Args:
        key_dict: The exact key combination to remove.
    """
    import pandas as pd

    with self._lock:
        df = self._read()
        if df.empty:
            return
        mask = pd.Series([True] * len(df), dtype=bool)
        for col, val in key_dict.items():
            if col in df.columns:
                mask = mask & (df[col].astype(str) == str(val))
        df = df[~mask]
        df.to_parquet(self.path, index=False)