Skip to content

Storage

JSONValue module-attribute 

JSONValue: TypeAlias = (
    "str | int | float | bool | dict[str, JSONValue] | list[JSONValue] | None"
)

Any value that round-trips through json.dumps/json.loads, used for the values a Storage backend reads and writes.

SQLiteStorage 

SQLiteStorage(
    database: str | Path = ":memory:", **kwargs: Any
)

Bases: Storage

SQLite-backed storage.

Parameters:

Name Type Description Default
database str | Path

Path to the sqlite database file or :memory:.

 ':memory:'
**kwargs Any

Additional keyword arguments passed to sqlite3.connect.

 {}
Source code in src/signalbot/storage.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def __init__(
    self,
    database: str | Path = ":memory:",
    **kwargs: Any,  # noqa: ANN401 -- Using object, causes in type error in the sqlite3.connect signature
) -> None:
    """Initialize SQLite-backed storage.

    Args:
        database: Path to the sqlite database file or `:memory:`.
        **kwargs: Additional keyword arguments passed to `sqlite3.connect`.
    """
    self._sqlite = sqlite3.connect(database, **kwargs)
    self._sqlite.execute(
        "CREATE TABLE IF NOT EXISTS signalbot (key text unique, value text)",
    )

delete 

delete(key: str) -> None

Delete a key from SQLite storage.

Parameters:

Name Type Description Default
key str

Storage key to delete.

 required

Raises:

Type Description
StorageError

If the delete operation fails.
Source code in src/signalbot/storage.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def delete(self, key: str) -> None:
    """Delete a key from SQLite storage.

    Args:
        key: Storage key to delete.

    Raises:
        StorageError: If the delete operation fails.
    """
    try:
        self._sqlite.execute("DELETE FROM signalbot WHERE key = ?", [key])
        self._sqlite.commit()
    except sqlite3.Error as e:
        raise StorageError(StorageBackend.SQLITE, StorageOperation.DELETE, e) from e

exists 

exists(key: str) -> bool

Check whether a key exists in SQLite storage.

Parameters:

Name Type Description Default
key str

Storage key to check.

 required

Returns:

Type Description
bool

True if the key exists, otherwise False.
Source code in src/signalbot/storage.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
def exists(self, key: str) -> bool:
    """Check whether a key exists in SQLite storage.

    Args:
        key: Storage key to check.

    Returns:
        True if the key exists, otherwise False.
    """
    return self._sqlite.execute(
        "SELECT EXISTS(SELECT 1 FROM signalbot WHERE key = ?)",
        [key],
    ).fetchone()[0]

read 

read(key: str) -> JSONValue

Read and deserialize a value from SQLite storage.

Parameters:

Name Type Description Default
key str

Storage key to read.

 required

Returns:

Type Description
JSONValue

The deserialized value stored under the key.

Raises:

Type Description
StorageError

If the key does not exist or deserialization fails.
Source code in src/signalbot/storage.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def read(self, key: str) -> JSONValue:
    """Read and deserialize a value from SQLite storage.

    Args:
        key: Storage key to read.

    Returns:
        The deserialized value stored under the key.

    Raises:
        StorageError: If the key does not exist or deserialization fails.
    """
    try:
        result = self._sqlite.execute(
            "SELECT value FROM signalbot WHERE key = ?",
            [key],
        ).fetchone()[0]
        return json.loads(result)
    except (sqlite3.Error, TypeError, json.JSONDecodeError) as e:
        raise StorageError(StorageBackend.SQLITE, StorageOperation.LOAD, e) from e

save 

save(key: str, value: JSONValue) -> None

Serialize and save a value to SQLite storage.

Parameters:

Name Type Description Default
key str

Storage key to write.

 required
value JSONValue

JSON-serializable value to store.

 required

Raises:

Type Description
StorageError

If serialization or database write fails.
Source code in src/signalbot/storage.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def save(self, key: str, value: JSONValue) -> None:
    """Serialize and save a value to SQLite storage.

    Args:
        key: Storage key to write.
        value: JSON-serializable value to store.

    Raises:
        StorageError: If serialization or database write fails.
    """
    try:
        serialized = json.dumps(value)
        self._sqlite.execute(
            "INSERT INTO signalbot VALUES (?, ?) "
            "ON CONFLICT(key) DO UPDATE SET value=?",
            [key, serialized, serialized],
        )
        self._sqlite.commit()
    except (sqlite3.Error, TypeError) as e:
        raise StorageError(StorageBackend.SQLITE, StorageOperation.SAVE, e) from e

RedisStorage 

RedisStorage(
    host: str, port: int, password: str | None = None
)

Bases: Storage

Redis-backed storage.

Parameters:

Name Type Description Default
host str

Redis host address.

 required
port int

Redis server port.

 required
password str | None

Optional Redis password.

 None
Source code in src/signalbot/storage.py 
193
194
195
196
197
198
199
200
201
def __init__(self, host: str, port: int, password: str | None = None) -> None:
    """Initialize Redis-backed storage.

    Args:
        host: Redis host address.
        port: Redis server port.
        password: Optional Redis password.
    """
    self._redis = redis.Redis(host=host, port=port, db=0, password=password)

delete 

delete(key: str) -> None

Delete a key from Redis storage.

Parameters:

Name Type Description Default
key str

Storage key to delete.

 required

Raises:

Type Description
StorageError

If the delete operation fails.
Source code in src/signalbot/storage.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
def delete(self, key: str) -> None:
    """Delete a key from Redis storage.

    Args:
        key: Storage key to delete.

    Raises:
        StorageError: If the delete operation fails.
    """
    try:
        self._redis.delete(key)
    except redis.RedisError as e:
        raise StorageError(StorageBackend.REDIS, StorageOperation.DELETE, e) from e

exists 

exists(key: str) -> bool

Check whether a key exists in Redis storage.

Parameters:

Name Type Description Default
key str

Storage key to check.

 required

Returns:

Type Description
bool

True if the key exists, otherwise False.
Source code in src/signalbot/storage.py 
203
204
205
206
207
208
209
210
211
212
def exists(self, key: str) -> bool:
    """Check whether a key exists in Redis storage.

    Args:
        key: Storage key to check.

    Returns:
        True if the key exists, otherwise False.
    """
    return bool(self._redis.exists(key))

read 

read(key: str) -> JSONValue

Read and deserialize a value from Redis storage.

Parameters:

Name Type Description Default
key str

Storage key to read.

 required

Returns:

Type Description
JSONValue

The deserialized value stored under the key.

Raises:

Type Description
StorageError

If the key does not exist or deserialization fails.
Source code in src/signalbot/storage.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def read(self, key: str) -> JSONValue:
    """Read and deserialize a value from Redis storage.

    Args:
        key: Storage key to read.

    Returns:
        The deserialized value stored under the key.

    Raises:
        StorageError: If the key does not exist or deserialization fails.
    """
    try:
        # This client is never configured with decode_responses=True, so
        # get() only ever returns bytes or None.
        result_bytes = cast("bytes | None", self._redis.get(key))
    except redis.RedisError as e:
        raise StorageError(StorageBackend.REDIS, StorageOperation.LOAD, e) from e

    if result_bytes is None:
        error = KeyError(f"Key '{key}' does not exist")
        raise StorageError(
            StorageBackend.REDIS, StorageOperation.LOAD, error
        ) from error

    try:
        return json.loads(result_bytes.decode("utf-8"))
    except json.JSONDecodeError as e:
        raise StorageError(StorageBackend.REDIS, StorageOperation.LOAD, e) from e

save 

save(key: str, value: JSONValue) -> None

Serialize and save a value to Redis storage.

Parameters:

Name Type Description Default
key str

Storage key to write.

 required
value JSONValue

JSON-serializable value to store.

 required

Raises:

Type Description
StorageError

If serialization or Redis write fails.
Source code in src/signalbot/storage.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def save(self, key: str, value: JSONValue) -> None:
    """Serialize and save a value to Redis storage.

    Args:
        key: Storage key to write.
        value: JSON-serializable value to store.

    Raises:
        StorageError: If serialization or Redis write fails.
    """
    try:
        serialized = json.dumps(value)
        self._redis.set(key, serialized)
    except (redis.RedisError, TypeError) as e:
        raise StorageError(StorageBackend.REDIS, StorageOperation.SAVE, e) from e

StorageError 

StorageError(
    backend: StorageBackend,
    operation: StorageOperation,
    cause: Exception,
)

Bases: SignalBotError

Raised when a storage backend operation fails.

Source code in src/signalbot/storage.py 
87
88
89
90
91
92
93
94
95
96
def __init__(
    self,
    backend: StorageBackend,
    operation: StorageOperation,
    cause: Exception,
) -> None:
    super().__init__(f"{backend.value} {operation.value} failed: {cause}")
    self.backend = backend
    self.operation = operation
    self.cause = cause

StorageBackend

Bases: Enum

Storage backend a StorageError was raised from.

StorageOperation

Bases: Enum

Storage operation a StorageError was raised from.