DiskStore¶

SQLite-backed MutableMapping / Mapping storage

Fast disk storage built on top of the Python standard-library sqlite3 module. Keys and values are serialised via a pluggable configuration system supporting plain blobs, JSON, NamedTuples, dataclasses and Pydantic models.

Features¶

Pure-Python on top of the standard-library sqlite3 module, with an optional APSW accelerator
100 % test coverage
Thread-safe and process-safe (fork‑safe)
Developed on Python 3.14, tested on CPython 3.10–3.14
Tested using GitHub Actions on Linux and macOS

Quickstart¶

from diskstore import DiskStore

ds = DiskStore("/tmp/diskstore_quickstart.db")
ds["key"] = "my value"
print(ds["key"])
assert len(ds) == 1
assert "key" in ds
assert list(ds) == ["key"]
del ds["key"]

Basic operations¶

DiskStore implements the full MutableMapping interface:

from diskstore import DiskStore

ds = DiskStore("/tmp/diskstore_basic.db")

# set
ds["one"] = 1
ds["two"] = 2

# get
assert ds["one"] == 1

# contains
assert "two" in ds

# iteration
assert set(ds) == {"one", "two"}

# length
assert len(ds) == 2

# keys / values / items
assert sorted(ds.keys()) == ["one", "two"]
assert sorted(ds.values()) == [1, 2]

# delete
del ds["one"]
assert len(ds) == 1

Bulk writes¶

Each individual write (__setitem__, __delitem__, …) creates an implicit SQLite transaction. Wrapping many writes in transact() eliminates that overhead:

from diskstore import DiskStore

ds = DiskStore("/tmp/diskstore_bulk.db")

# slow: one implicit transaction per write
for i in range(100):
    ds[i] = "value"

# fast: single explicit transaction
with ds.transact():
    for i in range(100):
        ds[i] = "value"

# update() is transactional by default
ds.update({i: "value" for i in range(100)})

transact() yields the active backend's cursor (sqlite3.Cursor, or apsw.Cursor when APSW is installed) for callers that need direct SQL. Nested calls are idempotent (the outer transaction is reused).

Auto-increment keys¶

When key_type=int is used, passing None as the key triggers SQLite auto-increment via INTEGER PRIMARY KEY NULL → rowid:

from diskstore import DiskStore
from diskstore.config import BaseConfig

ds = DiskStore("/tmp/diskstore_autoinc.db", config=BaseConfig(key_type=int))

key = ds.add(None, "auto-generated")
assert key is not None
assert isinstance(key, int)
assert ds[key] == "auto-generated"

Other `DiskStore` methods¶

from diskstore import DiskStore

ds = DiskStore("/tmp/diskstore_misc.db")
ds["one"] = 1
ds["two"] = 2

# pop
assert ds.pop("one") == 1
assert ds.pop("missing", "default") == "default"

# popitem (last inserted item)
_key, _value = ds.popitem()

# setdefault
ds.setdefault("three", 3)
assert ds["three"] == 3

# clear
ds.clear()
assert len(ds) == 0

# check integrity (optional VACUUM)
ds["three"] = 3
warnings = ds.check()
assert warnings == []
ds.check(vacuum=True)

# get read-only instance (shares the same DB file)
ro = ds.get_readonly_instance()
assert "three" in ro

`DiskRead` - read-only access¶

DiskRead is a lightweight Mapping implementation that opens the database read‑only:

from diskstore import DiskRead, DiskStore
from diskstore.config import BaseConfig

with DiskStore("/tmp/diskstore_autoinc.db", config=BaseConfig(key_type=int)) as wds:
    wds.add(None, "auto-generated")

# context-manager support (opens and closes the connection)
with DiskRead("/tmp/diskstore_autoinc.db") as ds:
    assert len(ds) > 0
    for key, value in ds.items():
        assert isinstance(key, int)

Query¶

DiskRead.query() supports WHERE, ORDER BY, LIMIT and OFFSET:

from diskstore import DiskRead, DiskStore

with DiskStore("/tmp/diskstore_bulk.db") as wds:
    wds.update({i: str(i) for i in range(10)})

with DiskRead("/tmp/diskstore_bulk.db") as ds:
    results = list(ds.query(where="_key > ?", parameters=(5,), limit=3))
    assert len(results) == 3
    assert results[0][0] == 6

Reversed iteration:

from diskstore import DiskRead, DiskStore

with DiskStore("/tmp/diskstore_bulk.db") as wds:
    wds.update({i: str(i) for i in range(3)})

with DiskRead("/tmp/diskstore_bulk.db") as ds:
    keys = list(reversed(ds))
    assert keys == sorted(keys, reverse=True)

Configuration¶

The config system controls how keys and values are stored in SQLite columns.

`BaseConfig` - single BLOB column¶

The default - stores values as raw SQLite BLOBs:

from diskstore import DiskStore
from diskstore.config import BaseConfig

config = BaseConfig(key_type=str)
ds = DiskStore("/tmp/diskstore_config_base.db", config=config)
ds["msg"] = "hello"
assert ds["msg"] == "hello"

`JsonConfig` - JSON-serialised TEXT column¶

from diskstore import DiskStore
from diskstore.config import JsonConfig

config = JsonConfig(key_type=str)
ds = DiskStore("/tmp/diskstore_config_json.db", config=config)
ds["nested"] = {"a": [1, 2, 3]}
assert ds["nested"] == {"a": [1, 2, 3]}

`NamedTupleConfig` - one column per field¶

from typing import NamedTuple
from diskstore import DiskStore
from diskstore.config import NamedTupleConfig

class Point(NamedTuple):
    x: float
    y: float

config = NamedTupleConfig(Point, key_type=str)
ds = DiskStore("/tmp/diskstore_config_nt.db", config=config)
ds["origin"] = Point(0.0, 0.0)
pt = ds["origin"]
assert pt.x == 0.0 and pt.y == 0.0

`DataclassConfig` - one column per field¶

from dataclasses import dataclass
from diskstore import DiskStore
from diskstore.config import DataclassConfig

@dataclass
class Item:
    name: str
    price: float

config = DataclassConfig(Item, key_type=str)
ds = DiskStore("/tmp/diskstore_config_dc.db", config=config)
ds["widget"] = Item("Widget", 9.99)
item = ds["widget"]
assert item.name == "Widget" and item.price == 9.99

`PydanticConfig` - JSON-serialised model column¶

from pydantic import BaseModel
from diskstore import DiskStore
from diskstore.config import PydanticConfig

class Task(BaseModel):
    title: str
    done: bool = False

config = PydanticConfig(Task, key_type=str)
ds = DiskStore("/tmp/diskstore_config_pd.db", config=config)
ds["task1"] = Task(title="Write docs")
task = ds["task1"]
assert task.title == "Write docs"
assert task.done is False

`StructtypeConfig` - JSON-serialised BLOB column¶

from structtype import Struct
from diskstore import DiskStore
from diskstore.config import StructtypeConfig

class Task(Struct):
    title: str
    done: bool = False

config = StructtypeConfig(Task, key_type=str)
ds = DiskStore("/tmp/diskstore_config_st.db", config=config)
ds["task1"] = Task(title="Write docs")
task = ds["task1"]
assert task.title == "Write docs"
assert task.done is False

Configuration options¶

Every config class accepts:

Option	Default	Description
`tablename`	`"DiskStore"`	SQLite table name
`key_type`	`bytes` (BLOB)	`int`, `str`, `float`, `bytes` or a SQLite type string
`timeout`	`10.0`	Busy timeout in seconds
`pragmas`	`{}`	Extra PRAGMAs merged with built-in defaults
`auto_migrate`	`True`	Auto-add missing columns at connection start

Table migration¶

When auto_migrate=True (the default), _migrate_table() is called on every first connection per-process. It creates the table if absent, then adds any columns that exist in the config but are missing from the existing table.

This makes schema evolution seamless - define a new version of your dataclass with additional fields and old data is preserved with defaults:

from dataclasses import dataclass
from diskstore import DiskStore
from diskstore.config import DataclassConfig

# V1 schema: name + price
@dataclass
class ProductV1:
    name: str
    price: float

config = DataclassConfig(ProductV1, tablename="products")
ds = DiskStore("/tmp/diskstore_migrate_dc.db", config=config)
ds[1] = ProductV1("Widget", 9.99)
assert ds[1] == ProductV1("Widget", 9.99)
ds.close()

# V2 schema: adds in_stock with a default
# auto_migrate=True adds the column automatically
@dataclass
class ProductV2:
    name: str
    price: float
    in_stock: bool = True

config_v2 = DataclassConfig(ProductV2, tablename="products")
ds2 = DiskStore("/tmp/diskstore_migrate_dc.db", config=config_v2)

# Old row gets the default for the new column
assert ds2[1] == ProductV2("Widget", 9.99, True)

# New rows use both old and new fields
ds2[2] = ProductV2("Gadget", 24.99, False)
assert ds2[2] == ProductV2("Gadget", 24.99, False)

Performance notes¶

DiskStore uses the Python standard-library sqlite3 module by default, so it has no required third-party dependency. If APSW is installed (pip install diskstore[apsw]) it is used automatically; force a backend with the DISKSTORE_BACKEND environment variable (apsw or sqlite3). The stdlib path requires SQLite 3.35 or newer (for INSERT ... RETURNING); the version bundled with the running CPython release is used. APSW bundles its own recent SQLite.

Default pragmas (src/diskstore/const.py):

Pragma	Value	Effect
`journal_mode`	`WAL`	Concurrent reads during writes
`page_size`	16 KB	Fewer B-tree pages and a smaller database (new databases only)
`cache_size`	32 MB (`-32768`, KiB)	Page cache per connection
`mmap_size`	256 MB	Memory-mapped I/O
`synchronous`	`NORMAL`	Balance speed / durability
`temp_store`	`MEMORY`	Temp tables and indexes kept in memory
`auto_vacuum`	`NONE`	No automatic vacuuming
`wal_autocheckpoint`	5000 pages (~82 MB at 16 KB)	WAL checkpoint threshold

cache_size and wal_autocheckpoint are counted in pages, so they scale with page_size. A negative cache_size is interpreted by SQLite as KiB and is therefore independent of the page size.

APSW vs stdlib `sqlite3`¶

scripts/bench_ab.py compares the two backends of the current checkout by running the same workloads in two subprocesses, one with DISKSTORE_BACKEND=apsw and one with DISKSTORE_BACKEND=sqlite3. The ratio is sqlite3 / apsw, so values above 1.0 mean APSW is faster. Sample results (Python 3.14, SQLite 3.53.1, 20 000 ops, 10 000 keys, 1 KB values, 4 processes, median of 3 rounds):

Workload	`sqlite3 / apsw`
`set` (single key)	1.00 – 1.08×
`get` (single key)	1.24 – 1.29×
`delete` (single key)	0.96 – 1.11×
`update` (bulk upsert)	1.05 – 1.20×
`set` inside `transact()`	1.19 – 1.38×
concurrent `get` (4 procs)	1.16 – 1.21×

The standard-library driver is near parity for writes and up to ~40 % slower for reads and transaction-batched writes; installing the optional apsw extra recovers that. Run make bench-ab to reproduce on your machine.

Benchmark scripts are available at scripts/benchmark.py, scripts/benchmark_core.py and scripts/bench_ab.py (apsw vs stdlib sqlite3 A/B comparison).

License¶

DiskStore is distributed under the terms of the BSD-3-Clause license.