Skip to content

Getting started

kv0 is a library: there is no server to run. You open a database by path or URI and use it from your process; several processes, on one machine or many, can open the same database.

Install

pip install kv0        # Python 3.11 or newer

Until the first release, build from a checkout with maturin: maturin develop --release.

cargo add kv0

Building needs protoc with the well-known .proto includes, which Lance compiles against: apt install protobuf-compiler libprotobuf-dev or brew install protobuf.

A key-value database

"""Key-value quickstart: the ordered key-value API on a local directory.

The same code works on object storage: pass a URI such as ``s3://bucket/prefix/db`` and the
credentials or endpoint as ``storage_options``.
"""

import tempfile

import kv0

path = tempfile.mkdtemp() + "/example_db"
db = kv0.DB(path, kv0.Options(create_if_missing=True), read_only=False)

batch = kv0.WriteBatch()
batch.put(b"key_1", b"value_1")
batch.put(b"key_2", b"value_2")
db.write(batch)

print(db.get(b"key_1"))  # b'value_1'

it = db.iteritems()
it.seek(b"key_")
while (item := it.next()) is not None:
    print(item)

db.flush()  # commit buffered writes as one Lance version
print("version", db.version)
db.close()

//! Key-value and tracking quickstart: `cargo run -p kv0 --example quickstart`.

use kv0::kv::{Db, Options, WriteBatch};
use kv0::tracking::{NewRun, Repo, RepoOptions, StepRange, Track};
use serde_json::json;

fn main() -> kv0::Result<()> {
    let dir = std::env::temp_dir().join(format!("kv0-quickstart-{}", std::process::id()));

    let db = Db::open(dir.join("kv").to_str().unwrap(), Options::default())?;
    let mut batch = WriteBatch::new();
    batch.put(b"key_1", b"value_1").put(b"key_2", b"value_2");
    db.write(batch)?;
    let mut it = db.iter();
    it.seek(b"key_");
    for (k, v) in it {
        println!(
            "{} = {}",
            String::from_utf8_lossy(&k),
            String::from_utf8_lossy(&v)
        );
    }
    db.close()?;

    let uri = dir.join("tracking").to_string_lossy().into_owned();
    kv0::rt::block_on(async move {
        let repo = Repo::open(&uri, RepoOptions::default()).await?;
        let run = repo.create_run(NewRun {
            name: Some("baseline".into()),
            ..NewRun::default()
        })?;
        run.set_params(&json!({"lr": 3e-4}))?;
        for step in 0..100 {
            run.track("loss", 1.0 / (step as f64 + 1.0), Track::default())?;
        }
        run.finish(kv0::tracking::RunStatus::Finished).await?;
        let tail = repo
            .read_metric(
                &run.id(),
                "loss",
                &json!({}),
                StepRange {
                    start: Some(95),
                    stop: None,
                },
            )
            .await?;
        println!("steps {:?}", tail.steps);
        repo.close().await
    })
}

What happened:

  1. DB(path) created a Lance dataset at path (it would be s3://bucket/prefix on S3).
  2. The batch was logged to a local write-ahead log, applied to the in-memory view (so get sees it at once), and buffered.
  3. flush() committed the buffer as one Lance version; close() would have done the same.

Open the same database elsewhere

reader = kv0.DB(path, read_only=True)       # another process, or another machine on S3
reader.get(b'key_1')                        # sees every committed write; refreshes every second
old = kv0.DB(path, version=1)               # a read-only view of an earlier version

A tracking repository

"""Tracking quickstart: a run with parameters and two metric series."""

import math
import tempfile

from kv0.tracking import Repo

repo = Repo(tempfile.mkdtemp() + "/tracking")

with repo.create_run(name="baseline", experiment="demo", tags=["synthetic"]) as run:
    run["hparams"] = {"lr": 3e-4, "optimizer": {"name": "adamw", "weight_decay": 0.01}}
    for step in range(200):
        run.track(math.exp(-step / 50), name="loss", context={"subset": "train"})
        if step % 20 == 0:
            run.track(1 - math.exp(-step / 80), name="accuracy", step=step, context={"subset": "val"})

print(repo.runs()[0]["status"])  # finished
print(repo.params(run.id)["hparams"]["optimizer"])
loss = repo.metric(run.id, "loss", context={"subset": "train"}, start=190)
print(list(zip(loss["steps"], (round(v, 4) for v in loss["values"]))))
repo.optimize()  # compact files and index metric series

Next