Skip to content

Utilities

Value codecs and exceptions used by the SQLite backend. Every stored field has exactly one column role, and encode_value/decode_value are the only places that convert between Python values and SQLite bindings — no reliance on SQLite affinity or pandas inference. See architecture.

Exceptions

research_tracker.utils.SyncConflictError

Bases: RuntimeError

Source code in src/research_tracker/utils.py 
11
12
class SyncConflictError(RuntimeError):
    pass

research_tracker.utils.SchemaMismatchError

Bases: RuntimeError

Source code in src/research_tracker/utils.py 
15
16
class SchemaMismatchError(RuntimeError):
    pass

Encoding and decoding

research_tracker.utils.encode_value(role, value)

Convert a Python value into the exact object bound to a SQLite column.

Source code in src/research_tracker/utils.py 
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
def encode_value(role: str, value: Any) -> Any:
    """
    Convert a Python value into the exact object bound to a SQLite column.
    """
    if is_missing(value):
        return None

    if role == "json":
        if hasattr(value, "item"):
            # numpy scalars (np.int64, np.bool_, ...) -> their Python scalar
            value = value.item()
        try:
            return json.dumps(value)
        except TypeError as error:
            raise TypeError(
                f"Cannot encode {type(value).__name__} as JSON: {value!r}"
            ) from error

    if role == "timestamp":
        timestamp = pd.Timestamp(value)
        if timestamp.tzinfo is None:
            timestamp = timestamp.tz_localize(UTC)
        else:
            timestamp = timestamp.tz_convert(UTC)
        return timestamp.isoformat()

    if role == "bool":
        return int(bool(value))

    if role == "int":
        return int(value)

    if role == "float":
        return float(value)

    return str(value)

research_tracker.utils.decode_value(role, value)

Source code in src/research_tracker/utils.py 
85
86
87
88
89
90
91
92
93
94
95
def decode_value(role: str, value: Any) -> Any:
    if is_missing(value):
        return None

    if role == "json" and isinstance(value, str):
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            return value

    return value

research_tracker.utils.is_missing(value)

Source code in src/research_tracker/utils.py 
37
38
39
40
41
42
43
44
def is_missing(value: Any) -> bool:
    if value is None or value is pd.NaT:
        return True

    try:
        return bool(pd.isna(value))
    except (TypeError, ValueError):
        return False

research_tracker.utils.store_to_sqlite_type(role)

Source code in src/research_tracker/utils.py 
33
34
def store_to_sqlite_type(role: str) -> str:
    return _SQLITE_TYPES[role]

Reading tables

research_tracker.utils.read_sqlite_frame(connection, table, columns)

SELECT the canonical columns ORDER BY rowid and build the frame with EXPLICIT dtypes, never through pandas type inference.

Source code in src/research_tracker/utils.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def read_sqlite_frame(
    connection: sqlite3.Connection,
    table: str,
    columns: tuple[tuple[str, str], ...],
) -> pd.DataFrame:
    """
    SELECT the canonical columns ORDER BY rowid and build the frame with EXPLICIT
    dtypes, never through pandas type inference.
    """
    names = [name for name, _ in columns]
    cursor = connection.execute(
        f"SELECT {', '.join(names)} FROM {table} ORDER BY rowid"
    )
    rows = cursor.fetchall()

    # `sqlite3` hands back Python ints for INTEGER columns, so a 64-bit seed
    # survives untouched as long as it is never inferred as a float column.
    data = {
        name: _series_for_role(role, [row[index] for row in rows])
        for index, (name, role) in enumerate(columns)
    }
    return pd.DataFrame(data, columns=names)