Datalevin Python Bindings

Python bindings for Datalevin over the JVM interop bridge.

Visit datalevin.org for Datalevin documentation.

Install

python -m pip install datalevin

Requirements:

• Python 3.10+
• Java 21+

Published wheels bundle the shared Datalevin runtime jar, so normal usage does not require building Datalevin from source.

Quick Start

from datalevin import connect, q, tx

entity = q.var("entity")
name = q.var("name")
all_names = q.query(
    find=q.collection(name),
    where=[q.datom(entity, "name", name)],
)

with connect(
    "/tmp/dtlv-py",
    schema={
        ":name": {
            ":db/valueType": ":db.type/string",
            ":db/unique": ":db.unique/identity",
        }
    },
) as conn:
    conn.transact(
        tx.data(
            tx.entity(-1, {"name": "Ada"}),
            tx.entity(-2, {"name": "Bob"}),
        )
    )

    names = conn.query(all_names)
    ada = conn.pull([":name"], 1)

    print(names)
    print(ada)

The q and tx builders are pure Python values: composing them does not start the JVM. The connection lowers them to the active backend only when query() or transact() is called.

Their contract is an immutable snapshot, not a mutable builder. Construction recursively detaches Python dictionaries, lists, sets, bytearray, and memoryview values; mappings become read-only mappings, sequences become tuples, and sets become frozenset values. Mutating an input container later therefore cannot change an existing form, and to_form() does not expose a mutable structural container. q.kw() and q.sym() tokens use Python value equality and hashing. Non-structural objects such as backend handles are treated as atomic values.

Prepared reads and acknowledgement-only writes

Connection and Database expose prepare_pull(selector, opts=None) and prepare_query(query). KV and KVTransaction expose prepare_get_value(dbi_name, k_type=None, v_type=None, ignore_key=True). The same names are available as top-level helpers, with the connection, database, or KV handle as their first argument.

read_person = conn.prepare_pull([":name", ":age"])
person = read_person.execute([":name", "Ada"])
lookup = conn.prepare_query(
    "[:find ?age . :in $ ?name :where [?e :name ?name] [?e :age ?age]]"
)
age = lookup(["Ada"])

Prepared queries accept EDN or typed q forms. Execute with a list of the remaining :in values, excluding the database, or [] when there are none. Results and input conversion match ordinary queries, including dynamic :limit and :offset bindings. Prepared objects can also be called directly or passed to execute_prepared(prepared, input).

A local prepared pull or KV read accepts execute(view, input), also exposed as execute_prepared(prepared, view, input), to read an explicit transaction view. Prepared queries and remote preparations do not accept explicit views; prepare them from the transaction handle instead. Keep the owning store open. Prepared reads require no separate close, and transaction-bound reads retain the transaction's lifetime and thread restrictions.

kv.update(dbi_name, key, fn, k_type=":data", v_type=":data", *args) and update_kv(kv, ...) atomically replace one value and return ":transacted". Embedded callbacks receive the old value (None for a missing key), followed by extra arguments. For remote updates, supply restricted Clojure function source:

kv.update("counts", "visits", lambda old: (old or 0) + 1)
remote_kv.update("counts", "visits",
                 "(fn [old amount] (+ (long (or old 0)) (long amount)))",
                 ":string", ":long", 1)

Functions must be free of side effects because updates may retry. Invalid values and callback exceptions abort the update; explicit transactions control commit.

conn.listen_db(callback, key=None) and listen_db(conn, ...) subscribe to committed changes to a remote database, including writes from other clients. Callbacks run on a dedicated thread with {":type": ":db-changed", ":db-name": "app"} and may query the connection to refresh data. Several commits may coalesce into one event. Terminal failures deliver a ":subscription-error" event with ":error". Registering the same key replaces its subscription. conn.unlisten_db(key) or unlisten_db(conn, key) stops a subscription; closing the connection stops all its subscriptions. conn.listen continues to deliver transaction reports.

conn.transact_ack(tx_data, tx_meta=None) and transact_ack(conn, ...) return ":transacted" on success. Validation and durability match transact, and listeners still receive full reports. Inside an explicit transaction, success means staged; the surrounding transaction controls commit or abort.

Use schema_attr(no_index=True) for :db/noindex. Reads by entity and pull remain available; query conditions on that attribute require conn.index_attr(attr) (or index_attr(conn, attr)) to atomically backfill AVE.

Composing Queries

Use normal Python control flow to assemble clauses. Variables, attributes, symbols, and keyword values are explicit, so strings beginning with ? or : remain literal strings.

The same rule applies recursively to runtime inputs passed to a typed query. Use q.kw("active") when an input is a keyword. EDN-string and native-list queries keep the legacy ":active" keyword shorthand for compatibility.

entity = q.var("entity")
name = q.var("name")
age = q.var("age")
minimum = q.var("minimum")

where = [
    q.datom(entity, "person/name", name),
    q.datom(entity, "person/age", age),
]
if adults_only:
    where.append(q.predicate(">=", age, minimum))

people = q.query(
    find=q.collection(name),
    inputs=[q.DB, minimum] if adults_only else [],
    where=where,
)

names = conn.query(people, 18) if adults_only else conn.query(people)

Available helpers cover relation, collection, tuple, and scalar finds; aggregates and pull expressions; database patterns, predicate, function-binding, rule, and logical clauses; input bindings, joins, rules, ordering, limits, offsets, and timeouts. q.datom(e, a, v) is the common three-term form; use variable-arity q.pattern(e, a) for a presence pattern or other supported database-pattern arity. q.raw() remains the structured escape hatch for new syntax; its tokens must use q.kw() and q.sym() explicitly. Relation queries return a list of row lists. Without order_by, row order is unspecified.

Typed forms validate structural grammar when they are composed: keyed result names must match a relation/tuple find, join variables must be non-empty and distinct, ordering must reference distinct projected variables or valid column indexes, and branches of one rule name must have matching required/free arity. q.and_() is a branch group for q.or_() or q.or_join() and is rejected as a top-level :where, not, not-join, or rule-body clause. q.raw() deliberately bypasses these typed checks.

Composing search clauses

Full-text, vector, and indexed-document searches have dedicated clauses and position-aware option builders:

entity = q.var("entity")
attribute = q.var("attribute")
value = q.var("value")
score = q.var("score")
distance = q.var("distance")

text_clause = q.fulltext(
    q.var("term"),
    [entity, attribute, value, score],
    attribute="document/text",
    options=q.fulltext_options(top=5, display="refs+scores"),
)

vector_clause = q.vec_neighbors(
    q.var("embedding"),
    [entity, attribute, value, distance],
    options=q.vector_search_options(
        domains=["documents"], top=10, display="refs+dists"
    ),
)

idoc_clause = q.idoc_match(
    q.var("predicate"),
    [entity, attribute, value],
    options=q.idoc_match_options(domains=["profiles"]),
)

Pass either attribute= or a domain list, as supported by the corresponding index. Result sequences are lowered to relation bindings; an explicit q.relation_binding(...) is also accepted. Static display options validate the result width while composing the query: full-text supports refs, refs+scores, texts, offsets, and texts+offsets; vector search supports refs and refs+dists. Use a q.var() as options= when the option map is a runtime query input. from_source= selects a non-default source.

Composing pull selectors

Pull selectors use position-aware forms so option values remain ordinary Python data:

person = q.selector(
    "person/name",
    q.pull_attr("person/nickname", default="none", as_="display"),
    q.pull_attr("person/age", xform="str"),
    q.pull_nested("person/friend", q.selector("person/name")),
    q.pull_recursive("person/manager", depth=2),
)

result = conn.pull(person, entity_id)

default and as_ accept arbitrary values and preserve strings recursively. A string xform is a symbol name; an explicit q.sym() is also accepted. Omit depth from pull_recursive for unbounded ... recursion. Existing native-list selectors remain supported and are normalized by pull grammar position rather than by their string contents.

Composing Transactions

Transaction items compose the same way:

items = [tx.entity(-1, {"person/name": "Ada"})]
if nickname is not None:
    items.append(tx.add(-1, "person/nickname", nickname))

report = conn.transact(tx.data(items))

In addition to entity maps, tx provides add, retract, retract_attribute, retract_entity, compare_and_swap/cas, call, ensure, and patch_idoc operations. Context-sensitive transaction values also have explicit builders:

eve = tx.lookup_ref("user/handle", "eve")

report = conn.transact(tx.data(
    tx.entity(-1, {
        "user/handle": "alice",
        "user/friend": eve,
        "user/child": tx.entity(-2, {"user/handle": "child"}),
    }),
    tx.patch_idoc(eve, "user/profile", [
        tx.patch_set(["status"], "active"),
        tx.patch_update(["visits"], "inc"),
        tx.patch_unset(["obsolete"]),
    ]),
    tx.invoke("people/audit", eve),
))

lookup_ref turns only its attribute into a keyword, so its lookup value stays literal. The patch builders similarly type only set, unset, update, and the update operation; paths, assigned values, and update arguments stay ordinary Python values. invoke emits the direct form for a transaction function installed under a database ident, while call emits :db.fn/call.

Wrap a nested entity map in tx.entity(...), as above. A plain nested dictionary with ordinary string keys remains data; the builder does not recursively convert it into an entity. More generally, values remain ordinary Python data; q.kw() and q.sym() add explicit Datalevin tokens where needed.

EDN lists and quoting

Python lists and tuples represent EDN vectors. Use q.edn_list() when EDN syntax requires a list, such as an IDoc predicate, and q.quote() for the common (quote value) form used by nested queries:

age_filter = {
    "profile": {"age": q.edn_list(q.sym(">="), 30)},
}

inner = q.query(
    find=q.relation(q.aggregate("min", q.var("age"))),
    where=[q.datom(q.IGNORE, "person/age", q.var("age"))],
)
quoted_inner = q.quote(inner)

Both forms are immutable, backend-neutral values and do not start the JVM. Their top-level aliases are edn_list() and quote(). Ordinary strings inside an EDN list remain strings; use q.sym() for an operator or other symbol. This keeps these cases composable without read_edn().

Data Style

The existing EDN and native-form APIs remain supported alongside the builders:

• schemas are dictionaries keyed by colon-prefixed attribute strings
• transaction data may use tx, dictionaries/lists, or existing tx_* helpers
• query and pull forms may be builder values, EDN strings, or Python lists
• colon-prefixed strings are converted to keywords in schema/query/form positions for the legacy native-form API
• use pure q.kw() in builder values, or keyword() with the legacy API, when the stored value itself is a keyword

from datalevin import keyword, read_edn, schema_attr, tx_add, tx_entity, write_edn

schema = {
    ":name": schema_attr(value_type=":db.type/string", unique=":db.unique/identity"),
    ":status": schema_attr(value_type=":db.type/keyword"),
}

tx = [
    tx_entity(-1, {":name": "Ada", ":status": keyword(":active")}),
    tx_add(-1, ":nickname", "A"),
]

form = read_edn("[:find ?e :where [?e :name _]]")
text = write_edn([":find", "?e", ":where", ["?e", ":name", "_"]])

Lazy Entity Example

conn.entity() returns a lazy entity wrapper. Use get() or index syntax for individual attributes, and call touch() only when you want a fully materialized dictionary. entity_map() is available for the old eager touched-map shape.

entity = conn.entity([":name", "Ada"])

print(entity.id)
print(entity.get(":name"))
print(entity[":db/id"])

touched = entity.touch()
eager = conn.entity_map(1)

Search, Vector, and Idoc Builders

Use helper builders for search/vector/idoc schema and option maps instead of hand-writing every namespaced key:

from datalevin import (
    embedding_attr,
    embedding_options,
    fulltext_attr,
    idoc_attr,
    search_domain,
    search_options,
    vector_attr,
    vector_options,
)

schema = {
    ":doc/text": fulltext_attr(domains=["docs"], auto_domain=True),
    ":doc/body": embedding_attr(domains=["docs"], auto_domain=True),
    ":doc/vec": vector_attr(domains=["docs"]),
    ":doc/json": idoc_attr(format="json", domain="profiles"),
}

opts = {
    ":search-domains": {"docs": search_domain(index_position=True)},
    ":search-opts": search_options(top=5, display="refs+scores"),
    ":vector-opts": vector_options(dimensions=384, metric_type="cosine"),
    ":embedding-opts": embedding_options(
        provider="openai-compatible",
        model="text-embedding-3-small",
        base_url="https://api.openai.com/v1",
        api_key_env="OPENAI_API_KEY",
        request_dimensions=1536,
        metric_type="cosine",
    ),
}

Standalone Vector Index Example

Use new_vector_index() when you want a KV-backed vector index without a Datalog schema attribute.

from datalevin import new_vector_index, open_kv, vector_options

with open_kv("/tmp/dtlv-py-vec") as kv:
    index = new_vector_index(kv, vector_options(dimensions=2))
    index.add_vec("doc-1", [1.0, 0.0])
    index.add_vec("doc-2", [0.0, 1.0])

    print(index.search_vec([1.0, 0.0], {":top": 1}))
    index.force_checkpoint()
    print(index.info())
    index.close()

Local Llama Example

Use new_llama_embedder() and new_llama_generator() with local GGUF models when you want direct llama.cpp handles outside Datalog embedding/search setup.

from datalevin import new_llama_embedder, new_llama_generator

with new_llama_embedder("/models/embed.gguf") as embedder:
    vector = embedder.embed("Datalevin stores facts.")
    print(embedder.dimensions(), len(vector))
    print(embedder.token_count("Datalevin stores facts."))

with new_llama_generator("/models/generate.gguf") as generator:
    print(generator.generate("Write one database tagline:", 32))

Async Transaction Example

transact() uses Datalevin's async transaction batching path and waits until the transaction commits before returning the report.

Use transact_async() for ingestion and application-server workloads that benefit from Datalevin's async transaction batching. It returns a standard concurrent.futures.Future.

future = conn.transact_async([{":db/id": -1, ":name": "Cara"}])
report = future.result(timeout=10)

Composing UDFs

Use the immutable UdfDescriptor value with the typed API. It preserves keyword types when nested in query inputs, transactions, schema attributes, or search-domain options. q.bind_udf() binds a query-function result and q.udf_predicate() creates a predicate clause. Transaction UDFs have explicit tx.call_udf(), tx.install_udf(), and tx.uninstall_udf() helpers; an installed descriptor can subsequently be called with tx.invoke(). Both transaction UDFs and predicate UDFs used by tx.ensure() receive a Database as their first callback argument.

from datalevin import UdfDescriptor, connect, create_udf_registry, q

registry = create_udf_registry()
descriptor = UdfDescriptor.query_fn("math/inc")

@registry.register(descriptor)
def inc(value):
    return value + 1

descriptor_input = q.var("descriptor")
number = q.var("number")
value = q.var("value")
increment = q.query(
    find=q.scalar(value),
    inputs=[q.DB, descriptor_input, number],
    where=[q.bind_udf(descriptor_input, value, number)],
)

with connect(
    None,
    opts={
        ":kv-opts": {":inmemory?": True},
        ":runtime-opts": {":udf-registry": registry},
    },
) as conn:
    assert conn.query(increment, descriptor, 41) == 42

udf_descriptor() remains available and returns the original mutable colon-string dictionary for EDN-string and native-list compatibility code. UdfDescriptor factories, bare IDs passed to a registry, and the query_udf()/predicate_udf()/tx_udf()/analyzer registration conveniences default to :python. The legacy helper retains its :java default. Both forms are accepted by a registry, explicit lang="java" remains available, and the typed UDF helpers normalize legacy descriptors when they are used explicitly.

Ordered Custom Types

Register a database-wide type through KV.register_type() or Connection.register_type(). Supply an order function returning an existing backing type, plus optional payload serde functions. Nippy is the default payload codec; a custom serializer returns bytes and its deserializer receives bytes.

import json
from datalevin import UdfDescriptor, create_udf_registry, open_kv

registry = create_udf_registry()

@registry.order_udf("task/order")
def task_order(task):
    return task["rank"]

@registry.serializer_udf("task/encode")
def task_encode(task):
    return json.dumps(task).encode("utf-8")

@registry.deserializer_udf("task/decode")
def task_decode(payload):
    return json.loads(payload.decode("utf-8"))

definition = {
    "index": {"type": ":long", "order-fn": UdfDescriptor.order_fn("task/order")},
    "payload": {
        "serialize": UdfDescriptor.serializer("task/encode"),
        "deserialize": UdfDescriptor.deserializer("task/decode"),
    },
}
with open_kv("/tmp/datalevin-custom-example",
             opts={":runtime-opts": {":udf-registry": registry}}) as kv:
    kv.register_type("app/task", definition)
    kv.open_dbi("tasks", {":key-type": ":app/task"})
    task = {"rank": 2, "name": "Compile"}
    kv.transact([(":put", task, "queued")], dbi_name="tasks")
    assert kv.get_value("tasks", task) == "queued"

For Datalog, register through the connection and set an attribute's :db/valueType to :app/task. Ordered list items use the DBI option {":value-type": ":app/task"}. Registrations persist and are shared by both APIs. Supply runtime bindings again when reopening. Remote stores execute functions using bindings installed on the server.

For native Python classes, bind the class to its payload descriptors before opening the local environment. Ordinary operations then accept and return instances automatically:

from dataclasses import dataclass
import json
from datalevin import UdfDescriptor, create_udf_registry, open_kv

@dataclass
class Task:
    rank: int
    label: str

registry = create_udf_registry()
registry.order_udf("native/order")(lambda task: task.rank)
registry.serializer_udf("native/encode")(
    lambda task: json.dumps([task.rank, task.label]).encode("utf-8"))
registry.deserializer_udf("native/decode")(
    lambda payload: Task(*json.loads(payload)))

definition = {
    "index": {"type": ":long", "order-fn": UdfDescriptor.order_fn("native/order")},
    "payload": {"serialize": UdfDescriptor.serializer("native/encode"),
                "deserialize": UdfDescriptor.deserializer("native/decode")},
}
registry.bind_native_type("app/task", Task, definition)
with open_kv("/tmp/dtlv-python-native", opts={":runtime-opts": {":udf-registry": registry}}) as kv:
    kv.register_type("app/task", definition)
    kv.open_dbi("tasks", {":key-type": ":app/task"})
    kv.transact([(":put", Task(1, "Compile"), "queued")], dbi_name="tasks")
    assert kv.get_value("tasks", Task(1, "Compile")) == "queued"
    assert kv.get_range("tasks", [":all"]) == [[Task(1, "Compile"), "queued"]]

The same binding supports custom Datalog attributes, query inputs, pull/entity results, and typed transaction builders. It is scoped to the supplied UDF registry and must be recreated on reopen. Exact matching uses Python == on complete values; equal objects do not need identical serialized bytes or a Python hash function.

Native class support covers local and remote operations, including query spilling to disk. For remote handles, use the same caller registry in :runtime-opts and install matching order/serde UDFs on the server through datalevin.server/*server-runtime-opts-fn*. Only type names and payload bytes cross the network; results are reconstructed by the caller's deserializer. Native values in untyped :data and helpers that convert values before a database handle is available remain unsupported. The current constant JVM hash can make large hash joins/deduplication expensive. See the custom data guide.

Fulltext Analyzer UDF Example

Use analyzer UDFs when a Datalog fulltext domain needs host-language tokenizing. The document analyzer runs while transactions and re-indexing update the fulltext index; the query analyzer runs during fulltext query evaluation.

import re

from datalevin import (
    connect,
    create_udf_registry,
    schema_attr,
    search_domain,
    UdfDescriptor,
)

registry = create_udf_registry()
analyzer = UdfDescriptor.analyzer("text/hashtags")
query_analyzer = UdfDescriptor.query_analyzer("text/plain-query")

@registry.register(analyzer)
def hashtags(text):
    return [
        [match.group(0)[1:], pos, match.start()]
        for pos, match in enumerate(re.finditer(r"#\w+", text))
    ]

@registry.register(query_analyzer)
def plain_query(text):
    return [[token, pos, pos] for pos, token in enumerate(text.split())]

with connect(
    None,
    schema={
        ":text": schema_attr(
            value_type=":db.type/string",
            fulltext=True,
            fulltext_auto_domain=True,
        )
    },
    opts={
        ":kv-opts": {":inmemory?": True},
        ":runtime-opts": {":udf-registry": registry},
        ":search-domains": {
            "text": search_domain(
                index_position=True,
                analyzer=analyzer,
                query_analyzer=query_analyzer,
            )
        },
    },
) as conn:
    conn.transact([
        {":db/id": 1, ":text": "alpha #needle"},
        {":db/id": 2, ":text": "needle without hash"},
    ])

    assert conn.query(
        "[:find [?e ...] :in $ ?q :where [(fulltext $ :text ?q) [[?e ?a ?v]]]]",
        "needle",
    ) == [1]

Datalog-Backed KV Example

Use datalog_kv() when you need ordinary KV tables in the same store as a Datalog connection. The returned KV handle is borrowed from the connection; do not close it separately.

from datalevin import datalog_kv

kv = datalog_kv(conn)
kv.open_dbi("app-state")
kv.transact([(":put", "k", "v")], "app-state", ":string", ":string")

Datom Inspection Example

Connection objects expose index-level reads for debugging, teaching, and migration tooling. Datom reads return dictionaries with :e, :a, :v, :tx, and :added keys; fulltext_datoms() returns [e, attr, value] triples.

print(conn.datoms(":eav", 1, ":name", limit=10))
print(conn.seek_datoms(":ave", ":name", "Ada", limit=5))
print(conn.rseek_datoms(":ave", ":name", "Bob", limit=5))
print(conn.index_range(":name", "A", "C"))
print(conn.count_datoms(None, ":name", "Ada"))
print(conn.fulltext_datoms("database", opts=search_options(limit=5, offset=10)))
print(conn.datalog_index_cache_limit())
conn.datalog_index_cache_limit(1024)
print(conn.tx_data_to_simulated_report([{":db/id": -1, ":name": "Dry Run"}]))

Bulk Load Example

Use init_db() and fill_db() when you already have Datom-shaped data and want the fast bulk-load path. Datoms can be compact tuples in (entity_id, attr, value) shape, and datom() creates the same shape.

For a server database, pass its authenticated dtlv:// URI as dir instead of a local path. Both fill_db(conn, datoms) and conn.fill_db(datoms) support remote connections. Remote loads preserve server WAL settings and commit in batches; outside an explicit transaction, earlier batches remain committed if a later batch fails. Use assigned entity IDs and trusted data matching the schema.

from datalevin import datom, fill_db, init_db

schema = {":name": {":db/valueType": ":db.type/string"}}

with init_db([(1, ":name", "Ada")], dir="/tmp/dtlv-py-bulk", schema=schema) as conn:
    fill_db(conn, [(2, ":name", "Bob")])
    conn.fill_db([datom(3, ":name", "Cara")])

KV Example

from datalevin import open_kv

with open_kv("/tmp/dtlv-py-kv") as kv:
    kv.open_dbi("items")
    kv.transact(
        [
            (":put", 1, "alpha"),
            (":put", 2, "beta"),
        ],
        dbi_name="items",
        k_type=":long",
        v_type=":string",
    )

    print(
        kv.get_value(
            "items",
            2,
            k_type=":long",
            v_type=":string",
            ignore_key=True,
        )
    )
    print(kv.get_range("items", [":all"], k_type=":long", v_type=":string"))
    print(kv.get_rank("items", 2, k_type=":long"))
    print(kv.get_entry_by_rank("items", 1, k_type=":long", v_type=":string"))
    print(kv.get_first_n("items", 2, [":all"], k_type=":long", v_type=":string"))

    kv.open_list_dbi("tags")
    kv.put_list_items("tags", "doc-1", ["clj", "db"], ":string", ":string")
    print(kv.get_list("tags", "doc-1", ":string", ":string"))
    print(kv.list_range("tags", [":all"], ":string", [":all"], ":string"))
    print(kv.list_range_first("tags", [":all"], ":string", [":all"], ":string"))
    print(kv.list_range_first_n("tags", 2, [":all"], ":string", [":all"], ":string"))
    print(kv.list_range_count("tags", [":all"], ":string"))
    print(kv.key_range_list_count("tags", [":all"], ":string"))
    print(
        kv.list_range_filter(
            "tags",
            lambda key, value: key == "doc-1" and value.startswith("c"),
            [":all"],
            ":string",
            [":all"],
            ":string",
        )
    )
    print(
        kv.list_range_keep(
            "tags",
            lambda key, value: f"{key}:{value}" if value == "db" else None,
            [":all"],
            ":string",
            [":all"],
            ":string",
        )
    )

Operational Example

KV stores expose backup, durability, snapshot, and WAL inspection helpers without raw JSON calls.

from datalevin import open_kv

with open_kv("/tmp/dtlv-py-ops", opts={":wal?": True}) as kv:
    kv.open_dbi("items")
    kv.transact([(":put", "a", "alpha")], "items", ":string", ":string")

    kv.sync()
    kv.copy("/tmp/dtlv-py-ops-copy")

    print(kv.tx_log_watermarks())
    print(kv.open_tx_log(1, limit=10))
    print(kv.create_snapshot())
    print(kv.list_snapshots())
    print(kv.gc_tx_log_segments())

Remote Client Example

Use new_client() for server administration against a running Datalevin server:

from datalevin import new_client

CLIENT_OPTS = {
    ":pool-size": 1,
    ":time-out": 5000,
    ":ha-write-retry-timeout-ms": 5000,
    ":ha-write-retry-delay-ms": 100,
}

client = new_client("dtlv://datalevin:datalevin@localhost", opts=CLIENT_OPTS)
created = False
opened = False

try:
    client.create_database("demo", "datalog")
    created = True
    info = client.open_database(
        "demo",
        "datalog",
        schema={
            ":name": {
                ":db/valueType": ":db.type/string",
                ":db/unique": ":db.unique/identity",
            }
        },
        info=True,
    )
    opened = True

    print(info)
    print(client.list_databases())
    print(client.replica_status("demo"))

    # For consensus HA databases, operator membership changes are available as:
    # client.ha_update_membership("demo", {":ha-members": [...], ...})
finally:
    if opened:
        client.close_database("demo")
    if created:
        client.drop_database("demo")
    client.disconnect()

Embedding Search Options

Python bindings include helper builders for newer store features such as :embedding-opts, :embedding-domains, and remote :openai-compatible embedding providers. Raw Datalevin option maps are still passed through when needed:

from datalevin import connect, embedding_options

with connect(
    "/tmp/dtlv-py-embed",
    schema={
        ":doc/id": {
            ":db/valueType": ":db.type/string",
            ":db/unique": ":db.unique/identity",
        },
        ":doc/text": {
            ":db/valueType": ":db.type/string",
            ":db/embedding": True,
            ":db.embedding/domains": ["docs"],
            ":db.embedding/autoDomain": True,
        },
    },
    opts={
        ":embedding-opts": embedding_options(
            provider="openai-compatible",
            model="text-embedding-3-small",
            base_url="https://api.openai.com/v1",
            api_key_env="OPENAI_API_KEY",
            request_dimensions=1536,
            metric_type="cosine",
        )
    },
) as conn:
    pass

Notes

• Datalevin values come back as ordinary Python values where possible.
• Remote client options such as :ha-write-retry-timeout-ms and :ha-write-retry-delay-ms can be passed to new_client().
• interop() is available for advanced raw-handle access when you need it.

Development

From this repo, the wrapper can run against:

1. DATALEVIN_JAR=/path/to/datalevin-runtime-<version>.jar
2. a vendored jar under src/datalevin/jars/
3. a repo-local build in target/

Typical local flow:

clojure -T:build vendor-jar
cd bindings/python
python -m venv .venv
. .venv/bin/activate
pip install -e '.[dev]'
pytest

The test suite also executes every case selected by the active release in the sibling dtlvtest golden spec. It discovers ../dtlvtest automatically; set DTLVTEST_ROOT=/path/to/dtlvtest for a checkout elsewhere. The conformance adapter reads spec/manifest.edn, preserves EDN keyword and symbol types, and lowers each dataset, transaction, and query through the typed Python builders. The test is skipped only when no dtlvtest checkout is available.

vendor-jar builds a platform-specific runtime jar for the current build host by default. To keep the cross-platform native payloads, pass clojure -T:build vendor-jar :native-platform all.

Wheel builds do this automatically with the all-platform runtime jar and produce a universal py3-none-any wheel. The supported release path is wheel-only:

python -m pip wheel --no-build-isolation bindings/python -w dist/

FreeBSD users should use the platform's own package instead of the PyPI wheel.

.github/workflows/release.python.yml builds the universal wheel on demand, smoke-tests it on Linux amd64, Linux arm64, macOS arm64, and Windows amd64, then uploads the wheel as an artifact. It does not publish to PyPI or TestPyPI.

For a local manual packaging helper, see script/deploy-python.md.

The hosted package workflow currently smoke-tests Linux amd64, Linux arm64, macOS arm64, and Windows amd64.

For ad hoc development against a different build, set DATALEVIN_JAR to point at another embeddable Datalevin runtime jar, preferably target/datalevin-runtime-<version>.jar.