Skip to content

API

DiskStore (MutableMapping) API.

Examples:

>>> from diskstore import DiskStore
>>> ds = DiskStore("/tmp/data.db")
>>> ds["one"] = 1
>>> ds["one"]
1

DiskStore

Bases: DiskRead, MutableMapping

filename property

DiskStore filename for DB.

tablename property

Tablename used to get data from.

timeout property

SQLite connection timeout value in seconds.

__contains__(key)

Check if key exists in the store.

__delitem__(key)

Delete key, raising KeyError if it is absent.

__getitem__(key)

Get value for key, raises KeyError if not found.

__init__(filename, config=None)

SQLite based MutableMapping disk storage.

Parameters:

Name Type Description Default
filename PathLike | str

DiskStore DB filename.

 required
config ConfigProtocol | None

Configuration as specified in ConfigProtocol

 None

__iter__()

Iterate over keys in insertion order.

__len__()

Return the number of items in the store.

COUNT(*) uses SQLite's specialised b-tree count instead of a row-by-row scan that evaluates the key column, which is substantially faster.

__reversed__()

Iterate over keys in reverse insertion order.

__setitem__(key, value)

Set key to value, replacing any existing entry.

add(key, value)

Insert a row and return its key.

With config=BaseConfig(key_type=int), passing None as key lets SQLite assign the next rowid. Returns None when the insert produced no row.

check(vacuum=False)

Run PRAGMA integrity_check and return a list of warnings.

The list is empty when the database is healthy. Pass vacuum=True to also reclaim free pages, which cannot run inside a transaction.

close()

Close the database connection if open.

get_readonly_instance()

Return a DiskRead over the same file.

The reader opens its own read-only connection, so it is not affected by this store's transaction state. The caller owns it and should close it.

items()

Return a set-like view of (key, value) pairs in the mapping.

keys()

Return a set-like view of keys in the mapping.

open()

Open (or re-open) the database connection and return self.

query(where=None, parameters=None, order=None, limit=None, offset=None)

Query rows with optional filtering, ordering, limit and offset.

Parameters:

Name Type Description Default
where Optional[str]

SQL WHERE clause (without the WHERE keyword).

 None
parameters Optional[Sequence | dict]

Parameters for the WHERE clause.

 None
order Optional[str]

ORDER BY clause (without the ORDER BY keyword).

 None
limit int | None

Maximum number of rows to return.

 None
offset int | None

Number of rows to skip.

 None

Yields:

Type Description
tuple

(key, value) tuples.

Examples:

>>> from diskstore import DiskRead
>>> ds = DiskRead("/tmp/data.db")
>>> list(ds.query(where="_key > ?", parameters=(1,), limit=5))
[(2, 'two'), (3, 'three')]

transact()

Wrapper for performance sensitive bulk writes.

Every write operation (__setitem__, __delitem__, add, etc.) runs in an implicit transaction when called outside of transact(). The implicit begin/commit overhead adds up when writing many items in a loop:

.. code:: python

# slow: one implicit transaction per write
for i in range(1000):
    store[i] = value

# fast: single explicit transaction
with store.transact():
    for i in range(1000):
        store[i] = value

Uses BEGIN IMMEDIATE to avoid deadlocks in concurrent workloads. Nested calls are idempotent (reuse the same transaction). Yields a sqlite3.Cursor for callers that need direct SQL execution.

update(other=(), /, **kwargs)

Bulk upsert from a mapping or iterable.

Wrapped in transact() so all upserts share a single transaction — significantly faster than setting keys individually in a loop.

values()

Return a set-like view of values in the mapping.

DiskRead (Mapping) API.

Examples:

>>> from diskstore import DiskRead
# DB must exist!
>>> ds = DiskRead("/tmp/data.db")
>>> ds["one"]
1

DiskRead

Bases: Mapping

filename property

DiskStore filename for DB.

tablename property

Tablename used to get data from.

timeout property

SQLite connection timeout value in seconds.

__contains__(key)

Check if key exists in the store.

__getitem__(key)

Get value for key, raises KeyError if not found.

__init__(filename, config=None)

SQLite read only disk storage.

Database is opened read only on demand.

Parameters:

Name Type Description Default
filename PathLike | str

filename for DB to use.

 required
config ConfigProtocol | None

Configuration

 None

__iter__()

Iterate over keys in insertion order.

__len__()

Return the number of items in the store.

COUNT(*) uses SQLite's specialised b-tree count instead of a row-by-row scan that evaluates the key column, which is substantially faster.

__reversed__()

Iterate over keys in reverse insertion order.

close()

Close the database connection if open.

items()

Return a set-like view of (key, value) pairs in the mapping.

keys()

Return a set-like view of keys in the mapping.

open()

Open (or re-open) the database connection and return self.

query(where=None, parameters=None, order=None, limit=None, offset=None)

Query rows with optional filtering, ordering, limit and offset.

Parameters:

Name Type Description Default
where Optional[str]

SQL WHERE clause (without the WHERE keyword).

 None
parameters Optional[Sequence | dict]

Parameters for the WHERE clause.

 None
order Optional[str]

ORDER BY clause (without the ORDER BY keyword).

 None
limit int | None

Maximum number of rows to return.

 None
offset int | None

Number of rows to skip.

 None

Yields:

Type Description
tuple

(key, value) tuples.

Examples:

>>> from diskstore import DiskRead
>>> ds = DiskRead("/tmp/data.db")
>>> list(ds.query(where="_key > ?", parameters=(1,), limit=5))
[(2, 'two'), (3, 'three')]

values()

Return a set-like view of values in the mapping.

Default constants used in diskstore.

DEFAULT_PRAGMAS = {'auto_vacuum': 0, 'cache_size': -(32 * 1024), 'page_size': 4 * 4096, 'journal_mode': 'wal', 'mmap_size': 2 ** 28, 'synchronous': 1, 'temp_store': 2, 'wal_autocheckpoint': 5000} module-attribute

default pragma settings

DEFAULT_RO_PRAGMAS = {'cache_size': -(32 * 1024), 'mmap_size': 2 ** 28, 'temp_store': 2, 'synchronous': 1} module-attribute

default read only pragma settings

TIMEOUT = 10.0 module-attribute

default timout in seconds

NO_DEFAULT

Sentinel marker for fields without a default value.

DiskStore configuration classes and helpers for config.

BaseConfig

Bases: ConfigProtocol

Default configuration: a single BLOB column named value.

Values are stored as-is with no serialisation, so anything SQLite can bind round-trips unchanged. This is the config used when none is passed to DiskStore.

Parameters:

Name Type Description Default
tablename str | None

SQLite table name, default "DiskStore".

 None
key_type Any

int, str, float, bytes or a SQLite type name such as "TEXT"; see get_sqlite_type(). int enables auto-increment keys via DiskStore.add().

 None
timeout float | None

seconds to wait for a locked database, default TIMEOUT (10.0). A negative value is treated as "use the default".

 None
pragmas dict | None

extra PRAGMAs merged over DEFAULT_PRAGMAS.

 None
auto_migrate bool | None

create the table if missing and add missing columns at connection start, default True.

 None

Subclass it to change how values are serialised; see JsonConfig and DataclassConfig.

dump_value(key, value)

Return value as the parameter tuple for INSERT/UPDATE.

load_data(data)

Return the value from a row tuple (key at index 0).

ConfigProtocol

Bases: Protocol

Configuration Protocol

Attributes:

Name Type Description
tablename str

Table name as string
key_type str

key type as string or basic Python type like str, int, float
timeout float

Timeout used to wait if someone blocks connections or with writes
pragmas dict

Dictionary with PRAGMAs to set when connections is initialized
fields Iterable

Iterable used for select, update and create statement with field name, type, default. [("value", str, NO_DEFAULT), ("value2", int, 0)]

dump_value(key, value) abstractmethod

Called with the key and value, should return a Sequence with the key as first element, suitable as parameter tuple for the INSERT/SET statement.

load_data(data) abstractmethod

load(db_data): called with the tuple selected from DB, including the key at index 0. Should be converted to the type normally received as value.

DataclassConfig

Bases: BaseConfig

One SQLite column per dataclass field.

Annotations are optional and default to bytes (BLOB) when omitted. A field default becomes the column default only if it is a bindable SQL literal (None, str, bytes, int or float); anything else is stored as NO_DEFAULT.

The tablename defaults to the dataclass name. _key is reserved for the primary key and raises ValueError if used as a field name.

Because _migrate_table() adds missing columns, adding a field with a default to an existing dataclass migrates old rows without data loss.

dump_value(key, value)

Flatten the dataclass into the column parameter tuple.

get_fields(dataclass) staticmethod

Build the (name, sqlite_type, default) column tuples.

load_data(data)

Rebuild the dataclass instance from a row tuple.

JsonConfig

Bases: BaseConfig

Store values as JSON in a single TEXT column.

Values are serialised with json.dumps on write and parsed with json.loads on read, so any JSON-serialisable object round-trips. The tablename stays "DiskStore" — the class name is not used.

dump_value(key, value)

JSON-encode value for the single TEXT column.

load_data(data)

JSON-decode the value from a row tuple.

NamedTupleConfig

Bases: BaseConfig

One SQLite column per typing.NamedTuple field.

Each annotated field becomes a column; type annotations are optional and default to bytes (BLOB) when omitted. Defaults come from _field_defaults and are only used when they are bindable SQL literals.

The tablename defaults to the NamedTuple's class name.

_key is reserved for the primary key and raises ValueError if used as a field name.

dump_value(key, value)

Flatten the NamedTuple into the column parameter tuple.

get_fields(value_class) staticmethod

Build the (name, sqlite_type, default) column tuples.

load_data(data)

Rebuild the NamedTuple from a row tuple.

PydanticConfig

Bases: BaseConfig

Store values as JSON in a single TEXT column, via Pydantic.

Values are serialised with model_dump_json() and validated back with model_validate_json(), so nested models round-trip. Requires pydantic (an optional dependency).

The tablename defaults to the model's class name.

dump_value(key, value)

JSON-encode the model with model_dump_json.

load_data(data)

Validate the stored JSON back into the model.

StructtypeConfig

Bases: BaseConfig

Store values as JSON in a single BLOB column, via structtype.

Values are serialised with struct_dump_json() and validated back with struct_validate_json(). Requires structtype (an optional dependency); it is faster than Pydantic but validates less strictly.

The tablename defaults to the struct's class name.

dump_value(key, value)

JSON-encode the struct with struct_dump_json.

load_data(data)

Validate the stored JSON back into the struct.

escape_name(name)

Quote name as a SQL identifier, doubling embedded quotes.

get_sqlite_type(type_)

Map a Python type or SQLite type name to a SQLite column type.

str becomes TEXT, int and bool become INTEGER, float becomes REAL, and anything else becomes BLOB. The four names "BLOB", "TEXT", "INTEGER" and "REAL" are passed through unchanged.

is_bindable_default(value)

Whether value can be encoded as a SQL literal.