xyzdb 1.0.1

Official Python driver for xyzDB — a thin, zero-dependency TCP client.

xyzDB — Python driver

A thin, zero-dependency Python client for xyzDB. It speaks the binary wire protocol directly (pure standard library), with a fluent query API, parameter binding, and bearer-token auth.

• Requires: Python ≥ 3.9 and a running xyzdb-server (default localhost:2505).
• Dependencies: none at runtime.

The engine itself is developed at the official xyzDB repo: github.com/Tunolabs/xyzdb.

Source & repository. This driver's source ships inside the published package — the sdist and wheel contain it, so pip download xyzdb gives you the code. The clients monorepo is not public; the engine, the wire spec PROTOCOL.md, and the reference client live at github.com/Tunolabs/xyzdb.

Install

pip install xyzdb

Quickstart

import xyzdb

with xyzdb.connect("127.0.0.1", 2505) as db:
    # Schema (optional — lobes auto-create on first PUT)
    db.create_lobe("clients", hint="Customer records")
    db.create_anchor("rfc", unique_in="clients")

    # Write
    db.put("clients", {"rfc": "ACME-001", "name": "Acme Corp", "region": "US-West"})

    # Read one (anchor lookup, O(1))
    acme = db.find("clients", where={"rfc": "ACME-001"}).execute()
    print(acme.name)                      # -> "Acme Corp"

    # Read many (top-N)
    top = list(
        db.scan("clients", where={"region": "US-West"})
          .order_by("revenue", descending=True)
          .limit(10)
    )

connect() returns a Client. Use it as a context manager (auto-closes) or call db.close() yourself. A client owns one TCP connection and is not thread-safe — use one per thread.

Connecting & auth

db = xyzdb.connect(host="127.0.0.1", port=2505)

If the server runs with --auth-token, set XYZDB_TOKEN in the environment; the driver sends the bearer token automatically on connect:

export XYZDB_TOKEN="my-secret-token"

Prefer 127.0.0.1 over localhost under high concurrency (avoids the dual IPv4/IPv6 connection budget).

Writing

# Single record. Prefix a field with * to make it a gravity field
# (records sharing the value are co-located on disk).
db.put("creditos", {"*rfc": "ACME-001", "_type": "Credit", "monto": 50000})

# Linked write (relationship to a parent record).
db.put("creditos",
       {"*rfc": "ACME-001", "_type": "Payment", "amount": 1000},
       link_to={"lobe": "clients", "where": {"rfc": "ACME-001"}, "as": "owner"})

# Upsert on the anchor.
db.put("clients", {"rfc": "ACME-001", "name": "Acme Inc"}, on_conflict="update")

# Atomic batch (≤ 10,000 records per call).
db.put_batch("clients", [
    {"rfc": "A-1", "name": "One"},
    {"rfc": "A-2", "name": "Two"},
])

on_conflict only supports "update".

Reading

find() and scan() return a QueryBuilder. Chain trailing clauses and pipeline steps, then terminate with .execute(), iteration, or a terminal verb. Clauses are assembled in the order the engine requires (WHERE → ORDER BY → LIMIT → CURSOR, then | steps) regardless of call order.

# Graph fetch: a record plus its linked descendants.
tree = db.find("clients", where={"rfc": "ACME-001"}).pull(depth=2).execute()

# Update in place.
db.find("clients", where={"rfc": "ACME-001"}).set(status="inactive")

# Delete matching records.
db.scan("creditos", where={"status": "cancelled"}).delete()

# Grouped aggregate.
r = (db.scan("creditos", where={"_type": "Credit"})
       .group_by("status")
       .aggregate("count()", "sum(monto)"))
print(r.count, r.sum_monto)

# Semantic top-k. The engine never embeds — declare the vector field, store
# embeddings you produced, then NEAREST over them (pass a query vector from the
# same model, or a REF to a stored record's vector).
db.create_lobe("mem")
db.create_vector("emb", "mem")            # VECTOR emb IN "mem"
db.put("mem", {"id": "a", "conv": "c1", "emb": [0.1, 0.2, 0.3]})
hits = (db.scan("mem", where={"conv": "c1"})
          .limit(10000)
          .nearest("emb", query_vector, k=10, metric="cosine")
          .execute())

Metrics for nearest(): "cosine", "dot", "l2".

Filters (where=)

where= takes a dict (an AND conjunction) or a raw predicate string:

# Equality (AND-combined)
db.scan("fintech", where={"status": "overdue", "region": "US-West"})

# List → IN; operator dicts → comparisons / set / null tests
db.scan("fintech", where={"status": ["overdue", "defaulted"]})   # status IN [...]
db.scan("fintech", where={"amount": {">": 5000}})
db.scan("clients", where={"tags": {"contains": "tech"}})
db.scan("data", where={"score": {"is_not_null": True}})

# Raw string → the full OR / NOT / parenthesised tree (build from trusted text;
# for untrusted values use execute() with $params)
db.scan("fintech", where='(status = "overdue" OR status = "defaulted") AND amount > 5000')

FIND is AND-only (anchor/gravity fast path); use SCAN for OR/NOT.

More reads: FETCH, FOLLOW, SHAPE, TAKE

# FETCH — several co-located lobes in ONE call (shared key). Returns one
# envelope whose fields are a named section per lobe.
env = db.fetch(["clientes", "creditos", "operaciones"],
               where={"rfc": "ACME-001"},
               as_names=["cliente", "creditos", "operaciones"])

# FOLLOW — jump across lobes: each message's cited_doc resolved in "documents".
docs = (db.find("messages", where={"thread": "T-1"})
          .follow("cited_doc", to="documents", on="doc_id").execute())

# SHAPE — project each record down to the fields you need.
rows = (db.scan("clients", where={"tier": "gold"})
          .order_by("score", descending=True).limit(10)
          .shape("name", "score").execute())

# TAKE — top-N over an aggregate (O(N) when a metric-ordered ghost exists),
# or a plain stream truncation.
from xyzdb import metric
top = (db.scan("creditos", where={"_type": "Credit", "status": ["active", "overdue"]})
         .group_by("rfc")
         .agg(metric("sum(monto)", as_="exposicion"), "count()")
         .take(100, by="exposicion").execute())

top5 = db.scan("clients").limit(1000).take(5).execute()   # truncate the stream

Use the non-terminal .agg(...) step (not the .aggregate(...) terminal) when a TAKE follows the aggregate.

Conditional aggregates (per-metric filter)

Each metric may carry its own alias and WHERE, computing several conditional aggregates in one pass. Build them with metric():

from xyzdb import metric

r = (db.scan("credits").group_by("empresa_id").aggregate(
        metric("count()", as_="total"),
        metric("count()", as_="active", where={"status": "active"}),
        metric("sum(amount)", as_="overdue_amount", where={"status": "overdue"}),
     ))

Deleting & purging

DELETE requires a selection — a where= (or a pipeline that produced records). Calling .delete() on an unfiltered scan raises ValueError and points you to purge(), so you never empty a lobe by accident. There is no cascade.

db.scan("creditos", where={"status": "cancelled"}).delete()   # matched rows only
db.purge("scratch")                                           # empty the whole lobe (deliberate)

Pagination

SCAN without .limit() caps at 1,000 rows; the hard ceiling is 10,000. Page beyond it by round-tripping the opaque cursor token:

resp = db.execute('SCAN "creditos" WHERE rfc = $r LIMIT 1000', {"r": "ACME-001"})
while resp.get("has_more"):
    resp = (db.scan("creditos", where={"rfc": "ACME-001"})
              .limit(1000).cursor(resp["cursor"]).execute())

Cursor pagination is not valid together with order_by, aggregate, or ghost routing.

Parameter binding (injection-safe)

Never interpolate untrusted input into a statement. Pass it through bound $name placeholders — substituted server-side, so the text never becomes syntax:

db.execute('FIND "clients" WHERE name = $n', {"n": user_input})

execute() is also the escape hatch for any xyTalk the fluent API doesn't cover.

Result types

• Record — one row. Access fields by attribute (rec.name) or key (rec["name"]). .to_dict(), .to_json(), .to_compact(), .get(k, default). For PULL results, iterate the record to walk linked children.
• AggregateResult — metrics by attribute or key; parenthesised keys are normalised (sum(monto) → result.sum_monto).

Errors

from xyzdb import XyzDBError, ConnectionError

try:
    db.put("clients", {"rfc": "dupe"})
except XyzDBError as e:
    print(e.code, e.message)        # server-side rejection
except ConnectionError as e:
    print("transport:", e)          # connect/send/recv failure

ConnectionError is a subclass of XyzDBError.

Tests

Integration tests need a running server on localhost:2505 (override with XYZDB_PORT):

pip install -e './python[test]'
pytest python/tests

Versioning & license

Released independently under a py-X.Y.Z SemVer tag; the tip always speaks the current xyTalk v1 wire surface. SemVer is from the client API's point of view — MAJOR breaks the client API, MINOR is additive, PATCH is fixes.

This package is Apache-2.0. The linked repository, github.com/Tunolabs/xyzdb, is the xyzDB engine, published under the Business Source License 1.1 — a different license from this client. The clients are permissive; only the engine is BUSL-1.1.

The wire protocol, PROTOCOL.md, may be implemented by anyone, under any license: if the engine's license gives you pause, you can write your own client without asking.

For the full language reference and operator docs, see the engine repo's docs/ and the MCP integration guide.