Python

kinedb has two Python packages with the same API shape:

• kinedb-client sends SQL to a running kinedb server, over HTTP and over a WebSocket.
• kinedb-embedded runs the kinedb engine inside your Python process, with no server.

kinedb-client and kinedb-embedded both export the name KineDB. The two classes share sql(), health(), rows_as_objects() and the result shapes, so the lines that use only these move between the packages with ONE import line:

from kinedb.client import KineDB      # a remote kinedb server
from kinedb.embedded import KineDB    # a local engine, in this process

The other calls are not the same in the two packages:

	kinedb-client	kinedb-embedded
Open	KineDB("http://host:4820")	KineDB("name", dir=...) or KineDB()
A socket	await db.connect()	none: the database is local
Files	put_file, get_file, file_info on the socket	put_file, get_file, get_file_stream on db
Live changes	sock.watch()	db.sync() with on_change
Default database	sock.set_context()	none: the engine is ONE database
Offline writes	none	push(), refused(), discard_refused(), freshness(), flush()

put_file and get_file take the same arguments in the two packages.

A section below says which package it is for. For JavaScript, see JavaScript.

The client: a remote server

kinedb-client is the direct client of kinedb for Python, with the same API as the JavaScript client.

Install

The packages live on the forge, under the kinedb organisation: https://git.kinedb.com/kinedb/-/packages. There is no pypi.org copy.

pip:

pip install --index-url https://git.kinedb.com/api/packages/kinedb/pypi/simple kinedb-client

uv, in the pyproject.toml of your application:

[[tool.uv.index]]
name = "kinedb"
url = "https://git.kinedb.com/api/packages/kinedb/pypi/simple"
explicit = true

[tool.uv.sources]
kinedb-client = { index = "kinedb" }

Use the forge as an EXPLICIT index, never --extra-index-url alone. All four kinedb names are free on pypi.org, so a second index may answer first and hand your application a stranger's package.

The index needs no token: the kinedb organisation is public. A token without the package scope gets 401 Unauthorized, so send no token, or a token with the package read scope. pip takes a token in the URL (https://<user>:<token>@git.kinedb.com/...); uv takes UV_INDEX_KINEDB_USERNAME and UV_INDEX_KINEDB_PASSWORD.

Versions are commits, not releases

kinedb has released no SDK, so no package carries a version number that promises compatibility. A build is named by the commit that produced it:

0.0.0.dev<N>+g<12-char sha>        N = git rev-list --count HEAD

0.0.0 and .dev both say that nothing is released. +g<sha> is the PEP 440 local version label, and it is the commit: kinedb-client==0.0.0.dev1234+g77df86eaede3 is the client of the image kinedb:77df86eaede3.

PyPI has no dist-tags, unlike npm. There is no latest to move and no g<sha> to install by. An unpinned install means "the highest version", and a hex sha does not order by time, so the ordering rides on <N>: the commit count only grows on main, so the newest merge is always the highest version. That is what latest does for the npm package.

Use

import asyncio
from kinedb.client import KineDB, rows_as_objects

async def main():
    db = KineDB("http://localhost:4820")

    print(await db.health())
    print(await db.sql("SHOW DATABASES"))    # one statement over HTTP

    sock = await db.connect()                # a persistent WebSocket
    await sock.authenticate("root", "hunter2")
    rows = await sock.sql("SELECT * FROM users")
    print(rows_as_objects(rows))

    sub = await sock.watch("users", lambda notify: print(notify))
    await sub.cancel()
    # A filter value goes in `params`, never in the SQL text (B3111).
    mine = await sock.watch("orders", print, where="owner = $1", params=["alice"])
    await mine.cancel()
    sock.close()

asyncio.run(main())

base_url is required. Unlike the browser client, this one has no page to derive a URL from. ws_url is derived from base_url when it is not given: http becomes ws, https becomes wss, and /ws is appended to the path.

A result is a plain dict, tagged the way the server tags its own JSON: resp["type"] is rows, created, inserted, backoff, health and so on. A typed result set also carries resp["types"] and resp["schema_id"].

On integers. The JavaScript client hands back a BigInt above 2^53, because a JavaScript number cannot hold an integer exactly past that point. A Python int is unbounded, so that distinction has no Python half: an Int column decodes to a plain int and keeps every bit, always.

Credentials

The client holds no session. An application installs two hooks once, and every KineDB(...) in that application picks them up:

from kinedb.client import set_auth_hooks

set_auth_hooks(
    get_token=lambda: store.token,           # read on EVERY request, never cached
    on_unauthorized=lambda: store.logout(),  # runs on a 401, BEFORE the raise
)

Either hook may be a coroutine function; the client awaits the result.

One client can override them, which is what a program that talks to a second server with a different credential needs:

KineDB(url, get_token=lambda: other_token)

With neither, the client sends no bearer and bounces nobody. That is the right default for a consumer that never logs in. GET /health stays bare in every case, because it is open by design and a liveness probe has no credentials.

set_auth_hooks returns the hooks it replaced, so a test or a one-off task can put them back:

previous = set_auth_hooks(get_token=borrowed)
...
set_auth_hooks(**previous)

Login with an API secret

A WebSocket connection logs in with one of four calls. Each call returns {"type": "auth_result", "username", "is_superuser"}.

await sock.authenticate("root", password)
await sock.authenticate_with_token(token)
await sock.authenticate_with_api_secret("root", os.environ["KINEDB_API_SECRET"])  # kdsk_...
await sock.authenticate_with_key("alice", key_id, sign)  # see "Log in with a key pair"

The API-secret login names the user and sends the secret in the login frame. The secret must be a non-empty string. No error message contains the secret.

Every login carries CLIENT_PROTOCOL_VERSION. The version is 6 since the token query (QueryAs, B2843). A server refuses a client of a lower version with an "upgrade the SDK" error, so upgrade the SDK and the server together.

Log in with a key pair

A user with a public key (ALTER USER alice ADD PUBLIC KEY '<point hex>') logs in with the private key. The private key never goes to the server, and the SDK never holds it. You give a sign function, and the SDK calls it once for each login:

await sock.authenticate_with_key(username, key_id, sign, client_name=None)

• key_id is 64 hex characters: the SHA-256 of the 65-byte uncompressed public point. key_id_from_public_point(point) calculates it.
• sign(message: bytes) returns bytes, or an awaitable of bytes. It signs message with ECDSA P-256 and SHA-256 and returns EXACTLY 64 raw bytes (r||s). The SDK refuses any other length and then sends no login frame.
• A DER signature (72 bytes or fewer, 0x30 ...) is not the wire form. der_signature_to_raw(der) changes it into the 64 raw bytes.

The SDK bounds an awaitable sign by the valid_ms of the challenge (30 seconds). After that, the login fails with the key did not sign in time, and a late signature is dropped. A sync sign runs on the event loop thread: the SDK cannot bound it, so keep it fast.

The login is three frames: a challenge request with a fresh client nonce (os.urandom(32)), the challenge of the server, then the signed login. Each call signs a new challenge. After a reconnect, call authenticate_with_key again with the same sign. Never store a signature: the server accepts it only once.

The errors:

• a sign that raises gives KineDBError("the key did not sign"). The error of sign is only its __cause__, so its text never reaches a log of the message;
• a wrong key and an unknown user both give the server text invalid username or password;
• a server that is too old gives SERVER_TOO_OLD_FOR_KEY_LOGIN.

The SDK keeps zero runtime dependencies. This example uses the package cryptography. It is your choice, not a dependency of kinedb-client. It loads the .pk8.pem file that the web console gives:

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec

from kinedb.client import der_signature_to_raw, key_id_from_public_point

with open("alice.pk8.pem", "rb") as f:
    key = serialization.load_pem_private_key(f.read(), None)
point = key.public_key().public_bytes(
    serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint
)
key_id = key_id_from_public_point(point)
sign = lambda m: der_signature_to_raw(key.sign(m, ec.ECDSA(hashes.SHA256())))

await sock.authenticate_with_key("alice", key_id, sign)

Get a bearer token with a key pair (HTTP)

login_with_key logs in over HTTP and returns a bearer token (B3672). It needs no socket.

from kinedb.client import KineDB, login_with_key

out = await login_with_key(
    "https://db.example.com", "alice", key_id, sign, "my-app", timeout_ms=10_000
)
db = KineDB("https://db.example.com", get_token=lambda: out["token"])

• One call sends POST /login/challenge, signs the same bytes as authenticate_with_key, and sends the proof to POST /login. It returns {"token": ..., "expires_at_ms": ...}. The function calls sign(bytes) only: it never sees the private key, and it sends the key id, the signature and the nonces.
• Each call asks for a new challenge with a new random nonce. A proof is never sent twice. The function does not retry: a 503 or a 429 sets err.retryable (a 429 also err.retry_after_sec). Call login_with_key again to retry with a new challenge.
• timeout_ms (default 10 s) bounds each of the two requests. A timeout is retryable.
• The errors are those of authenticate_with_key: invalid username or password, the key did not sign, the key did not sign in time, and SERVER_TOO_OLD_FOR_KEY_LOGIN for a server with no HTTP key login.
• It is a function, not a method of KineDB: the embedded KineDB has no base URL.

Retrying

With no retry option, the client retries ONE kind of error: a typed hold refusal (B2298). Every other server-flagged transient rejection raises at once, with err.retryable set, so a caller can build its own loop. The CLIENT retries; the server never does.

A hold refusal has refusal["kind"] write_hold or not_ready (B495). A split, a merge or a leadership transfer holds the writes of a shard for a short time, and the server refuses a write with refusal: {"kind": "write_hold", "retry_ms": N}. The server sends the refusal before the write takes effect, so nothing was written and a resend is safe. The client sends the request again after retry_ms until hold_budget_ms (20 s, HOLD_RETRY_BUDGET_MS) from the first send. After the budget, the client raises the last refusal. This is the default for sql(), transaction(), a socket from connect() and put_file(). For a file, the client sends again only the refused chunk frame or manifest frame, never the whole upload.

retry=False turns off every retry, the hold retry too:

db = KineDB(url, retry=False)

retry=True also retries the other transient rejections. The client waits a random full-jitter time between two attempts, and honours the server's own retry_ms pacing hint on a backoff response. A hold refusal does not count against max_retries. Every other retryable error keeps max_retries.

db = KineDB(url, retry=True)
db = KineDB(url, retry={"max_retries": 5, "hold_budget_ms": 20000})

The outcome is unknown

A KineDBOutcomeUnknownError (a subclass of KineDBError) tells you that the server sent your request on, but the reply was lost (B2304). The write MAY have applied. The error has outcome_unknown True, retryable False and response["refusal"]["kind"] outcome_unknown (TLV kind code 5). No retry option sends the request again, also not retry=True. Read the data back before you send the request again:

from kinedb.client import KineDBOutcomeUnknownError

try:
    await db.sql("INSERT INTO orders VALUES (42, 'paid')")
except KineDBOutcomeUnknownError:
    r = await db.sql("SELECT id FROM orders WHERE id = 42")
    # No row: the insert did not apply, and you can send it again.

A default database

db = KineDB(url, database="app")              # branch main
db = KineDB(url, database="app", branch="dev")
sock.set_context("app", "dev")                # one socket; set_context() removes it

Every statement of a socket carries the default. db.sql() sends ?db=, and it raises KineDBError for a default branch other than main: POST /sql has no branch selector. A qualified table name (default.t) wins over the default. A socket with a default refuses USE.

Parameters

sql(text, params) sends the values of $1, $2, ... apart from the text. The server binds them, so a value is never parsed as SQL. Use parameters for every value that comes from a user.

name = "it's'); DELETE FROM t; --"
await sock.sql("INSERT INTO t (id, name) VALUES ($1, $2)", [2, name])  # one row, a string
await sock.sql("SELECT id FROM t WHERE name = $1", [name])
await db.sql("SELECT id FROM t WHERE id = $1", [2])                    # HTTP too

Python value	SQL type
bool	BOOL
int	INT
float	FLOAT
str	TEXT
an aware datetime	TIMESTAMP (microseconds)
bytes, bytearray	BYTES
Vector([1.0, 0.5])	VECTOR (the items go as f32)
None	NULL

params is a list or a tuple. The client raises TypeError or ValueError before it sends anything for: any other type, an int outside the i64 range, a float that is not finite, a naive datetime (its zone is unknown), and a Vector with an item that is not a finite f32 number. A plain list is an ARRAY, and the server refuses an ARRAY for a VECTOR column: use Vector. No params (or []) sends the old bytes. Over HTTP the body is the TLV [sql, P] with the content type application/vnd.kinedb.sql-params.

Functions: call()

call(name, args) runs a server function (B2173). It is on KineDB (HTTP) and on KineDBSocket. args is a list in the order of the parameters, or a dict by parameter name. A missing NULL parameter is NULL.

await sock.call("add_points", [1, 50])                      # 51: RETURNS INT
await sock.call("add_points", {"player": 1, "points": 50})  # by name
await sock.call("top", [2])                                 # RETURNS TABLE: a list of dicts

The connection reads SHOW FUNCTIONS on its first call and keeps the typed signatures. Each argument goes with the type of its PARAMETER, not the Python type: a FLOAT parameter takes 2, a TIMESTAMP takes an aware datetime, an int of microseconds or an ISO text with Z, a VECTOR(n) takes a list of n numbers, and a ROW takes a dict (sent as JSON). A value that does not fit raises TypeError or ValueError with the function and the parameter name, and nothing goes on the wire. The statement is CALL name($1, ..., $n) with typed parameters, so a value is never parsed as SQL.

A function that changed on the server after the read (CREATE OR REPLACE): the client reads SHOW FUNCTIONS again one time, converts again and sends again one time. A function the read does not list, and a function with no parameter types (made before B2172), raise KineDBError with no CALL.

Describe a table: describe()

describe(table) reads the structure of a table (B3096). It is on KineDB (HTTP) and on KineDBSocket. It answers one dict.

d = await sock.describe("users")
d["schema_version"]                      # an int. 0 means unknown.
d["primary_key"]                         # ["id"]: the column names in key order
d["columns"][1]                          # {"name": "email", "type": "Text", "not_null": True,
                                         #  "default": None, "primary_key": False,
                                         #  "unique": True, "indexes": ["users_email"]}
d["indexes"]                             # [{"name": "users_email", "columns": ["email"], "unique": True}]

The call runs DESCRIBE, SHOW INDEXES FROM and DESCRIBE again. It uses the database and the branch of the connection, as sql() does. A column is unique only when a UNIQUE index of that one column covers it. The three statements are not one snapshot. When the schema_version of the two DESCRIBE replies differs, the call starts again. After 3 reads it raises KineDBError. The shared test vector is sdk/test-vectors/describe-v1.json.

Transactions

transaction(statements, ...) sends ONE complete transaction in ONE request. The server writes all the statements in one atomic write, or it writes none of them. KineDB (POST /sql) and KineDBSocket have the same method:

sock = await db.connect()
await sock.authenticate("root", "hunter2")
await sock.transaction(
    [
        ["INSERT INTO orders (id, item) VALUES ($1, $2)", [1, "pen"]],
        ["INSERT INTO orders (id, item) VALUES ($1, $2)", [2, "ink"]],
        "DELETE FROM carts WHERE id = 7",
    ],
    token="order-1-2",
)   # both rows and the delete are written, or none

• A statement is a str, [sql] or [sql, params] (a tuple is correct too). The params are the values of $1, $2, ..., with the type table of sql().
• No transaction stays open on the server. BEGIN, COMMIT and ROLLBACK are not statements of a transaction, and sql() with one of them is refused. Two transaction() calls on one socket are two separate transactions.
• The answer is created, as for a write with sql(). When a statement fails, the server writes nothing and the call raises a KineDBError.
• A SELECT inside a transaction gives no rows. For a read-modify-write, put the read and the write in a function and run it with CALL (see the SQL page, § Functions and tickers).
• token is the commit token. The server stores it in the same commit as the rows. A client that lost its connection asks GET TRANSACTION STATUS '<token>' if the commit happened, and it sends the transaction again only when it did not. A failed transaction stores no token.
• The retry argument has the shape of the constructor option and wins over it (retry=False turns it off). A retry sends the WHOLE transaction again, never a part of it. db and branch win over the default of the client. Over HTTP a branch other than main is refused, as for sql(): use the socket.
• The transaction holds no lock. At the commit the server checks each row that an UPDATE, a DELETE or an INSERT over an existing key started from. When another client wrote that row first, the server writes nothing and answers calvin conflict: … — nothing applied, retry, with the retry flag set. An INSERT of a new key is not checked: when two transactions insert the same key, the later write wins. To claim a key, use INSERT … IF NOT EXISTS, also inside a transaction: a key that exists refuses the whole transaction with duplicate key:. A blind UPDATE (WHERE <primary key> = <value>, only literal SETs, a table with no index and no policy, and no read of that row earlier in the transaction) is not checked either: it writes only its columns, and the later write of a column wins.
• The client puts consecutive statements with the same text and the same parameter types in one group on the wire. The order of the statements never changes. Over HTTP the body is the TLV payload with the content type application/vnd.kinedb.sql-txn; over the socket the frame type is 0x1E.

The SQL page, § Transactions, has the full rules.

Run a query as a token user

query_as(token, sql, params=None, *, db=None, branch=None, timeout_ms=None) runs ONE statement with the rights of the user of the bearer token (kdt1_..., from POST /login). The socket sends the frame QueryAs (0xF2, B2843). Only KineDBSocket has the method: HTTP has no QueryAs.

sock = await db.connect()
await sock.authenticate("webapp", password)   # webapp holds GRANT QUERY AS
rows = await sock.query_as(user_token, "SELECT id, total FROM orders WHERE id = $1", [42])

The reply is the reply of sql(). Each refusal is a KineDBError. Compare its message with the constants SESSION_REJECTED, QUERY_AS_NO_SUBSCRIPTION, QUERY_AS_NO_USE, QUERY_AS_NO_TXN_CONTROL, QUERY_AS_CLIENT_CONNECTION_ONLY and SERVER_TOO_OLD_FOR_QUERY_AS. The client does not retry a QueryAs. No error text, log line or repr holds the token.

The ORM: kinedb.client.orm

The package carries the kinedb ORM (B2387, B3099) as the module kinedb.client.orm. A model is a class, and ONE model maps to ONE table. The specification is docs/orm/spec.md. Each object is live (P3, B3100): when another client changes its row, the fields of the object change too. Each awaited result is live too: its list changes when the rows change (see "Live results").

from datetime import datetime
from kinedb.client.orm import AUTO, Configuration, Credentials, Database, Field, Model, field

class Task(Model, table="tasks"):
    id: Field[str] = field(primary_key=True, default=AUTO)  # AUTO: the ORM makes a UUID
    title: Field[str]                                       # str: TEXT NOT NULL
    done: Field[bool] = field(default=False)
    priority: Field[int | None] = field(indexed=True)       # `| None`: NULL is allowed
    views: Field[int] = field(counter=True, default=0)      # a COUNTER column
    due: Field[datetime | None]                             # an aware datetime

config = Configuration(
    server_url="http://localhost:4820",
    credentials=Credentials.username_password(username="ana", password="..."),
    schema=[Task],
)

async with Database.open(config) as db:
    tasks = await (db.objects(Task)
                   .where(Task.priority >= 2, done=False)
                   .order_by(Task.priority.desc())
                   .limit(10))
    one = await db.get(Task, "a")                # the object, or None

    async with db.write() as tx:                 # ONE transaction at the end of the block
        tasks[0].done = True                     # an update writes only the changed columns
        tasks[0].views += 1                      # a counter writes a delta: views = views + 1
        tx.create(Task(title="Write docs"))
        tx.delete(one)

Models. The annotation Field[T] names the value type: str, int, float, bool, bytes, datetime (an aware datetime) or Vector (with field(dimensions=n)). A read gives a vector value, and a write of a vector value goes as a Vector parameter (B3171). field() takes primary_key, default, unique, indexed, counter, column (the column name, when it is not the field name) and renamed_from (the old column name of a rename). A type checker sees the keyword __init__ of each model (typing.dataclass_transform).

The open. Database.open() connects, logs in and compares each model with its table. A missing table gets CREATE TABLE, a new optional field gets ADD COLUMN, and an index gets CREATE INDEX. The open checks and plans every model first. A refused model, a change that the ORM cannot plan (for example a new required field) or a destructive statement (DROP COLUMN) refuses the open with an OpenError, and NOTHING runs. e.code is the code of the first problem, and e.problems names the table, the field or the column, and the code of each one. A difference that needs no statement (NOT NULL, DEFAULT) is a notice in db.notices. Credentials also has token(...) and anonymous(). Database.open_backend(backend, schema) opens on an adapter of the Backend port that you made.

Queries. where(cond, ..., field=value) takes the conditions and the keyword equalities as one AND. A condition is Task.priority >= 2, ==, !=, <, <=, >, .in_([...]), .is_none(), .is_not_none(), .startswith(p) or .contains(p), and &, | and ~ combine them. Each value goes to the server as a typed $n parameter, never as SQL text. NULL follows SQL: Task.priority != 2 does not select a row whose priority is NULL. The sort always ends with the primary key, so the order is total. A NULL in a required field is a read error (required) that names the key: the ORM never invents a value.

The write block. Only assignments inside async with db.write() are valid; an assignment outside raises OrmError with the code outside_write. At the end of the block the ORM sends every change as ONE transaction(). An exception in the block, or a refusal of the server, writes nothing, and each object goes back to its values before the block. tx.create(obj, mode=...):

• mode="error" (the default) never replaces a row. A block of only these creates of ONE model is ONE INSERT … IF NOT EXISTS: one present key refuses the whole statement. A create with a key that the ORM made (AUTO) is a plain INSERT in the transaction. A create with your key beside another change raises BackendUnsupportedError (create_in_transaction): the ORM does not send IF NOT EXISTS inside a transaction yet. The engine accepts it there since B3168.
• mode="upsert" is a plain INSERT: it replaces the row at the key. A model with a counter refuses it (counter_upsert).
• mode="ignore" raises BackendUnsupportedError (create_ignore, B3168).

Every refusal of the ORM is an OrmError with a code, and nothing is sent.

Live objects. obj.subscribe(listener) gives a callable that stops the listener. obj.changes() is an async iterator of the same events, and it ends after the delete. An event is ObjectChange(deleted, changed_fields):

task = await db.get(Task, "a")
stop = task.subscribe(lambda change: print(change.changed_fields))  # ('title',)
async for change in task.changes():
    if change.deleted:   # ObjectChange(deleted=True, changed_fields=())
        break
stop()

An update with the same values gives no event. A delete, or a row that you may no longer read, makes the object invalid (task.is_valid is False), and then no more events come. The ORM keeps ONE watch for each table on the socket.

Live results. A result is ONE predicate of the watch of its table. Its list changes in place, and each change gives ONE ResultsChange with the indices of the list:

from kinedb.client.orm import ResultsChange

open_tasks = await db.objects(Task).where(done=False).order_by(Task.priority.desc()).limit(20)
stop = open_tasks.subscribe(lambda change: print(change))
# ResultsChange(deletions=(0,), insertions=(3,), modifications=(1,), moves=(Move(from_=4, to=2),), reset=False)
async for change in open_tasks.changes():
    ...
stop()

Replay an event on a UI list: delete at deletions and moves[].from_ (old indices), insert at insertions and moves[].to (new indices), then reload modifications. A gap of the watch gives ONE event with reset=True: reload the whole list. A result with a limit watches about 2 × limit rows and moves its bound when the window gets short, with no short list in between. A result shows the commits of this database that came before it. close() makes the result invalid (is_valid is False). The spec is docs/orm/spec.md § 7.

Relations. A ref holds the key of a row of another model. A backlink reads the rows that point to an object. A many-to-many relation stores one link row for each pair (B3102):

from kinedb.client.orm import Backlink, ManyToMany, Ref, backlink, many_to_many, ref, ref_key

class Project(Model, table="projects"):
    id: Field[str] = field(primary_key=True)
    tasks: Backlink[Task] = backlink(lambda: Task, "project")

class Tag(Model, table="tags"):
    id: Field[str] = field(primary_key=True)
    tasks: ManyToMany[Task] = many_to_many(lambda: Task, inverse="tags")

class Task(Model, table="tasks"):
    id: Field[str] = field(primary_key=True, default=AUTO)
    project: Ref[Project] = ref(Project, optional=True, on_delete="nullify")
    tags: ManyToMany[Tag] = many_to_many(Tag)  # the link table tasks_tags

task = await db.get(Task, "a")
task.project                     # the Project object, or None (absent, deleted, hidden)
ref_key(task, "project")         # the stored key, always
tasks = await task.project.tasks # a live result of the tasks of the project
tags = await task.tags           # a live result of the tags of the task
async with db.write() as tx:
    task.tags.add(tag)           # one link row, key "<task>:<tag>"
    task.tags.remove("old")      # deletes every link of the pair
    tx.delete(project)           # nullify: the tasks of the project keep no key

A ref takes an object or a key. A ref is null while its target is absent, and it changes to the object (with ONE event) when the target arrives. All the refs of the database to one table cost ONE predicate of its watch, and all the backlinks with no limit of one field cost ONE predicate. on_delete is none (the default), nullify or cascade; the rules run in the transaction of the delete. where() on a many-to-many result raises BackendUnsupportedError (many_to_many_where, B3289). The spec is docs/orm/spec.md § 5.

A known gap of the type check: mypy wants each ref as an argument of the model constructor, also an optional ref. Write Task(project=None) for mypy; at runtime an optional ref that you do not give is None.

No runtime dependencies

The package installs nothing else. The HTTP transport is urllib.request on a worker thread, and the WebSocket client is our own RFC 6455 codec over asyncio streams (kinedb.client.ws). An SDK that pulled websockets or httpx would force a version range on every application that installs it.

Pass http_request= or ws_connect= to KineDB(...) to drive it from a test with no server.

Login and the protocol version

The ways to log in (password, API secret, key pair, token) are on the page Authentication.

sock.authenticate(username, password), sock.authenticate_with_token(token), sock.authenticate_with_api_secret(username, secret) and sock.authenticate_with_key(username, key_id, sign) send the protocol version of the client (CLIENT_PROTOCOL_VERSION, now 6) in the login frame. The API-secret login (B2499) sends a secret that starts with kdsk_. The login frame has 6 slots. A key login sends 8 slots: the key id is in slot 7 and the signature is in slot 8. It uses the challenge frames before the login frame (see "Log in with a key pair" above). A server refuses a login with no version, or with a version below its minimum. The error names the client version, the server minimum and the fix: "upgrade the SDK".

A key login over HTTP (B3672) uses login_with_key(base_url, username, key_id, sign, client_name=None, *, timeout_ms=None). It sends POST /login/challenge, signs the same bytes as the WebSocket key login, and sends the proof to POST /login. It returns {"token": ..., "expires_at_ms": ...}: give the token to get_token of KineDB or to sock.authenticate_with_token(token). The function calls sign(bytes) only and never sees the private key. Each call asks for a new challenge, and the function does not retry. A 503 or a 429 sets err.retryable; call again to get a new challenge.

A login succeeds only on an AuthResult reply. Every other reply raises a KineDBErrorunexpected reply to a login: <type> (B2857). The Rust client does the same. The message names the reply type, never a credential.

A login that gets no reply in its timeout_ms (default 10 seconds) raises WS login timed out after <n>ms and closes the connection (B2891). Open a new connection before you try again. The client matches each reply to its request by order, so a late reply on the old connection could answer the wrong request. The Rust client does the same.

After a login, every reply has typed cells: rows results and VERBOSE watch rows. The client asks for no typed option at login. The typed_rows keyword of KineDBSocket is removed: a call that passes it gets a TypeError. A socket with no login asks for typed rows on each statement, so a guest also gets typed cells. A guest reply always carries the full column header.

Parameters

Give a value from a caller as a parameter, never as joined SQL text:

await sock.sql("SELECT id FROM people WHERE name = $1", [name])
await db.sql("INSERT INTO people (id, name) VALUES ($1, $2)", [7, name])

sql(text, params) sends each value with its type on the binary wire: bool is BOOL, int is INT (64 bits), float is FLOAT, str is TEXT, datetime with a time zone is TIMESTAMP, bytes and bytearray are bytes, and None is NULL. A list or a tuple of scalars of one type is an ARRAY value (see below). Any other type is refused. See Parameters.

An array parameter takes bool, int, float, str, bytes and bytearray, and datetime values with a time zone, and all elements must be of one type. A TypeError refuses, before the client sends anything, a None element, a nested list or tuple, an element of another type and a mix of types (an int next to a float is a mix). The empty list is valid. A column of the type ARRAY(<type>) in a result is a Python list. See Array columns.

Default database

A client can name a default database, and a default branch, for its statements:

db = KineDB("http://localhost:4820", database="app")
await db.sql("SELECT * FROM orders")          # POST /sql?db=app
sock = await db.connect()
await sock.sql("SELECT * FROM orders")        # app.orders on main
sock.set_context("app", "dev")                # app@dev from now on
sock.set_context()                            # no default

• A socket sends the default in every statement frame. The server keeps no default for the connection.
• A default database with no branch uses the branch main. A branch with no database is an error (ValueError).
• sock.set_context(db, branch) changes the default of one socket. With no argument, it removes the default.
• A qualified table name in the SQL wins over the default, for example SELECT * FROM default.orders (see qualified table names).
• USE <db> and USE BRANCH <b> on a socket with a default get an error that names set_context: the default wins over USE, so USE changes nothing.
• db.sql() sends the database as POST /sql?db=<name>. POST /sql has no branch selector, so db.sql() with a default branch other than main raises KineDBError. Use a socket, or name the branch in the table name (app@dev.orders).
• A database that does not exist gets database not found: <name>. A branch that does not exist gets branch not found: <name>.
• put_file, get_file and file_info with no db and no branch use the default too.
• transaction() on a default branch other than main is refused. Transactions on a branch are not supported yet.

Retrying

With no retry option, the client retries a typed hold refusal only (B2298). The refusal kind is write_hold or not_ready, and the server wrote nothing. The client sends the request again after the retry_ms of the server, for up to 20 s (hold_budget_ms). This applies to sql(), transaction(), a socket and put_file(). A file upload sends again only the refused chunk frame or manifest frame. Every other retryable error raises at once with err.retryable set.

• retry=False turns off every retry, the hold retry too.
• retry=True, or a dict of options, also retries the other retryable errors.

See A write fails while a shard splits, merges or moves its leader and the README of kinedb.client.

The outcome is unknown

A KineDBOutcomeUnknownError (a subclass of KineDBError) tells you that the server sent the request on, but the reply was lost (B2304). The write MAY have applied. The error has outcome_unknown True, retryable False and response["refusal"]["kind"]outcome_unknown. No retry option sends the request again, also not retry=True. Read the data back before you send the request again. kinedb.embedded raises the same error on a call() that goes to the server.

from kinedb.client import KineDBOutcomeUnknownError

try:
    await db.sql("INSERT INTO orders VALUES (42, 'paid')")
except KineDBOutcomeUnknownError:
    # Read the row back. No row: the insert did not apply, so send it again.
    ...

See A forwarded write fails: the outcome is unknown.

The schema version of a table

A rows result has table_versions: a list of [table, version], one pair for each table that the statement read. A watch_fill, a watch_tail and a watch_data frame have schema_version: the version of the watched table.

r = await sock.sql("SELECT * FROM orders")
# r["table_versions"]  ->  [["orders", 57]]

• The version is the stamp of the last DDL on the table. It is an int.
• The version changes at each DDL of the table: ALTER TABLE, CREATE INDEX, DROP INDEX, SET CONFLICT POLICY, also a GRANT or a change of a policy. DROP TABLE and then CREATE TABLE gives a later version.
• The version never changes at a row write. An INSERT, an UPDATE and a DELETE leave it as it was.
• The version never goes down. Compare two versions only for equality: a different value means that a DDL ran. The change can be one that does not touch the columns, so read DESCRIBE again to see the new shape.
• 0 or an absent value means that the version is not known. table_versions is [] when the server sent no version, and schema_version is 0.
• DESCRIBE <table> has the same value in its last column, schema_version.
• Do not compare the version of a server with the version of an embedded database (the embedded package). The two counters are not the same.

Describe a table

await describe(table) returns the schema of one table as a dict. Both packages have it, on every connection object.

d = await sock.describe('users')
# d['schema_version']  →  57
# d['primary_key']     →  ['id']
# d['columns'][1]      →  {'name': 'email', 'type': 'Text', 'not_null': True, 'default': None,
#                          'primary_key': False, 'unique': True, 'indexes': ['users_email']}
# d['indexes'][0]      →  {'name': 'users_email', 'columns': ['email'], 'unique': True}

• The keys are the same as in the JavaScript packages.
• A column is unique only when an index of that one column is UNIQUE. A composite unique index does not make its columns unique.
• default is the SQL literal of the default, or None.
• The call reads DESCRIBE, SHOW INDEXES FROM and DESCRIBE again. When the version moves between the two DESCRIBE reads, it starts again. After 3 reads it raises.

Transactions

transaction(statements, ...) sends ONE complete transaction in ONE request. The server writes all the statements, or none of them:

await sock.transaction(
    [
        ["INSERT INTO orders (id, item) VALUES ($1, $2)", [7, "pen"]],
        ["UPDATE stock SET n = n - 1 WHERE item = $1", ["pen"]],
    ],
    token="order-7",
    retry=True,
)
await db.transaction(["DELETE FROM carts WHERE id = 3"])   # POST /sql

• A statement is a str, [sql] or [sql, params]. The params are the parameters of sql(text, params).
• The socket sends the Transaction frame (0x1E). db.transaction() sends the same payload as POST /sql with content-type: application/vnd.kinedb.sql-txn.
• The answer is {"type": "created"}. An error raises KineDBError, and nothing of the transaction is written. The answer gives no row counts.
• No transaction stays open on the server. sql("BEGIN"), sql("COMMIT") and sql("ROLLBACK") are refused. Two transaction() calls on one socket are two separate transactions.
• A SELECT in the list gives no rows. For a read, a decision and a write in one step, CALL a function.
• token is the commit token. A second transaction with a token that committed applies nothing and answers created.
• retry has the shape of the retry option of the client and wins over it (False turns it off). A retry sends the WHOLE transaction again, never a part.
• db and branch win over the default of the client.

See Transactions.

Run a query as a token user

A web server can keep one socket for all its requests. Each request has the bearer token of its own user. query_as runs ONE statement as the user of that token, on the shared socket. The statement has the rights of that user only. It never has the rights of the socket user.

1. Give the user of the socket the right: GRANT QUERY AS TO 'webapp' (GRANT QUERY AS). root needs the grant too.
2. Log in on the socket as that user.
3. Call query_as with the token of the request user. The user got the token from POST /login.

from kinedb.client import KineDB, KineDBError, SESSION_REJECTED

sock = await KineDB("http://localhost:4820").connect()
await sock.authenticate("webapp", password)

rows = await sock.query_as(user_token, "SELECT id, total FROM orders")
rows = await sock.query_as(
    user_token, "SELECT * FROM orders WHERE id = $1", [42], db="shop", branch="main"
)

try:
    await sock.query_as(user_token, "SELECT * FROM orders")
except KineDBError as err:
    if err.message == SESSION_REJECTED:
        ...  # the token is not valid: the user logs in again

The rules:

• The signature is query_as(token, sql, params=None, *, db=None, branch=None, timeout_ms=None). params are the values of $1, $2, ..., as in sql.

• With no db and no branch, the frame carries the default database of the socket. Else the frame carries the db and the branch that you give.

• The socket sends the frame QueryAs (0xF2). Only the socket has query_as: HTTP has no QueryAs frame.

• The reply is the reply of sql, with typed rows. A logged-in socket asks for the column header on each query_as, as the Rust client does.

• Each refusal raises a KineDBError. Its message is the text of the server. Compare it with a constant of kinedb.client:

  Constant	When
  SESSION_REJECTED	No grant, a bad token, an expired or a revoked token. The error does not tell which. Do not retry it.
  QUERY_AS_NO_SUBSCRIPTION	WATCH, ALTER WATCH or UNWATCH.
  QUERY_AS_NO_USE	USE. Give db and branch to query_as.
  QUERY_AS_NO_TXN_CONTROL	BEGIN, COMMIT or ROLLBACK.
  QUERY_AS_CLIENT_CONNECTION_ONLY	The connection is on the peer plane.
  SERVER_TOO_OLD_FOR_QUERY_AS	The server does not know the frame. Upgrade the server.

• The client does not retry a query_as. A Backoff reply, or an error with the retry flag, raises a KineDBError with retryable set to True. You decide to send it again.

• With no reply in timeout_ms (default QUERY_AS_TIMEOUT_MS, 30 seconds), the client closes the socket. A late reply must not answer a later request. Open a new socket.

• No error text, log line or repr of the client holds the token.

See Run a query as a user of a token.

Watch

sock.watch(table, callback) sends WATCH <table> and calls callback with each frame that the server pushes for that table. A pushed frame never answers a statement: a statement in flight keeps its own reply.

The keyword options are the options of the JS client (B3097): verbose=True, fast=True, where="<SQL>", since="START" or since="<token>" (a fill), and live=True (a live tail after the fill; it needs since). So sock.watch("t", cb, since="START", live=True, where="n > 2") sends WATCH t SINCE START LIVE WHERE n > 2. The client adds the callback before it sends the WATCH, so the callback gets the first fill page.

sock.watch(table, callback, where=..., params=[...]) adds WHERE <where> to the statement. params holds the values of the $1, $2, ... of where, with the types of sql(text, params). The server binds them, so a value is never parsed as SQL. Put each filter value in params, not in the text of where (B3111). params without where raises ValueError.

sub = await sock.watch("orders", print, where="owner = $1", params=[user_id])

table goes into the text of the statement, so it must be a table name. A name has one to three parts with . between them, for example orders or shop.orders. A part is a word ([A-Za-z_][A-Za-z0-9_]*) or a bracket-quoted name such as [order lines]. A part can have an @branch, as in shop@dev.orders. Any other text, for example orders WHERE 1=1, raises ValueError, and a table that is not a str raises TypeError. Every check runs before the client sends a frame.

n["type"] tells the frames apart:

n["type"]	The frame	What it carries
watch_notify	WatchNotify (0x22)	table, changed: the count of a plain watch, and operation when the server names it (insert, update, delete, or rights with changed: 0 after a rights change, B3043)
watch_data	WatchTypedData (0x27)	columns, types, data: the typed rows of a VERBOSE watch, with typed: True. It also has schema_version (B2480). The server no longer sends the old WatchData (0x23); the client fails the connection if it gets one
watch_fill	WatchFill (0x24)	the fill of a WATCH … SINCE subscription: upserts, removes, token, full, more, upsert_versions, upsert_manifests, upsert_predicates, error, gap (B2201: the gap answer of a watch with ALLOW GAP, or None; it holds path only when the server names the read path of a server_limit gap, B3233), schema_version (B2480)
watch_tail	WatchTail (0x28)	one diff of a WATCH … SINCE … LIVE subscription: the watch_fill fields plus base, the token the diff starts from

The rows of upserts and removes are typed cells, as in a typed rows result. A non-empty error marks a fill that failed: it has no rows and no token. upsert_predicates has one list for each upsert row: the ids of the predicates of the watch that match the row. It is empty for a watch with a bare WHERE. See A set of predicates.

A fill comes in pages (B2200). A watch_fill or watch_tail frame with more true is one page. Its token is a resume token: a cursor (c1:…) or a commit hash. Do not parse it. The server sends the next page only when the client asks for it with WatchFillNext (0x29). The client asks after every callback returned. When a callback returns an awaitable (an async def callback), the client asks when the awaitable is done. Store the page rows and the token together in that callback.

When a callback raises, the client does not ask. After 120 s with no ask, the server ends the fill with a failure frame (retry true). Then subscribe again with sock.watch("t", cb, since="<the stored token>"). A page can have zero rows: store its token too. The last frame (more false) carries the plain commit token.

Files

A socket stores and reads a file of any size. A file has no size limit by design; the client sends chunks of at most 128 KiB, content-defined, so two versions of a file share their unchanged chunks.

The three calls are put_file, get_file and file_info. Each one takes an optional db, branch and timeout_ms. The timeout is the budget of ONE frame, not of the whole file, and it defaults to 60 seconds. A frame with no reply in its budget raises no reply to <frame> within <n>ms and closes the connection (B2891). A SQL statement names its database and branch in the table name instead, for example SELECT * FROM app@dev.orders (see qualified table names).

put_file

put_file cuts the bytes at content boundaries, hashes each chunk with BLAKE3 and sends one chunk per frame, at most four unanswered. It then sends the manifest: the size, the BLAKE3 of the whole file and the chunk hash list. It returns that manifest, plus chunk_bytes_sent: the chunk bytes the call really sent.

The client does not send a chunk again when the shard already holds it. The chunk can come from any file of the same table. Before the upload, the client asks the server which chunk hashes it holds, and it sends only the other chunks. The server decides: when it refuses the manifest, the client sends the skipped chunks and the manifest once more. A file that the shard already holds reports chunk_bytes_sent 0. A server without this feature gets every chunk, as before.

with open("report.pdf", "rb") as handle:
    stored = await sock.put_file("files", "report.pdf", handle.read())

print(stored["size"])        # 4_182_003
print(stored["hash"])        # the blake3 of the whole file, 64 hex characters
print(len(stored["chunks"]))  # 64, each one 64 hex characters

put_file takes an optional fields dict. A table takes it as the column values of the file row. A collection adds it to the manifest document. A user with only a MATCH permission writes a file this way, because the policy judges the row that the write leaves. The server refuses the keys name, path, size, hash, chunks, chunk_count and _id. An empty dict counts as no map. A value that is not a dict raises TypeError before the client sends any byte. The server must be of a build that knows the map: an older server ignores it.

await sock.put_file("files", "report.pdf", data, fields={"tenant": "acme"})

get_file

get_file reads the manifest, then fetches every chunk and verifies it. A chunk whose bytes do not hash to the name the manifest gave raises, and so does an assembled file that does not match the whole-file hash. No bytes come back from a failed read: a caller never sees a file that was only partly verified.

data = await sock.get_file("files", "report.pdf")
open("copy.pdf", "wb").write(data)

file_info

file_info reads the manifest alone, with no chunk transfer. Use it to learn the size of a file, or to compare two versions by their chunk lists.

info = await sock.file_info("files", "report.pdf")
print(info["size"], info["hash"], len(info["chunks"]))

The server must support files

The file calls need a server build with file support. Every server mode stores files: one server without --calvin, a one-node cluster, and a cluster of several nodes. The client asks for file support on the login frame, and the server echoes the grant; sock.files_granted reports the answer. On a logged-in socket with no echo, every file call raises this server has no file support (FLAG_FILES not granted). A socket that never logged in is not checked, because an open server answers without a login and then the grant was never on the wire.

await sock.authenticate("root", "hunter2")
assert sock.files_granted

The embedded package: the engine in your process

kinedb-embedded is the embedded shape of kinedb for Python. The package carries the kinedb engine as a native library, and it runs the database inside your Python process. It needs no server and no port to read and write its own data. It can also keep a local copy of a table of a running kinedb server: the sync.

It is the Python twin of the JavaScript package (see JavaScript).

Platforms

• Linux x86_64 and macOS arm64 (Apple silicon, macOS 11 or newer). Each wheel carries the native library libkinedb_wasm.so. The Linux wheel has a manylinux tag. The macOS wheel has the tag macosx_11_0_arm64.
• Not yet: Linux aarch64 and macOS x86_64. On these platforms pip install finds no wheel.
• Python 3.10 or newer.
• No runtime dependency. The package uses the standard library only (ctypes, asyncio, ssl).

Install

The packages live on the forge, under the kinedb organisation: https://git.kinedb.com/kinedb/-/packages. There is no pypi.org copy.

pip:

pip install --index-url https://git.kinedb.com/api/packages/kinedb/pypi/simple kinedb-embedded

--index-url REPLACES pypi.org for this command, so pip reads the forge only. That is the explicit form for pip.

uv, in the pyproject.toml of your application:

[[tool.uv.index]]
name = "kinedb"
url = "https://git.kinedb.com/api/packages/kinedb/pypi/simple"
explicit = true

[tool.uv.sources]
kinedb-embedded = { index = "kinedb" }

Then add the dependency:

uv add kinedb-embedded

Use the forge as an EXPLICIT index, never --extra-index-url alone. The kinedb names are free on pypi.org, so a second index can answer first with a package that is not ours.

The index needs no token: the kinedb organisation is public. A token without the package scope gets 401 Unauthorized, so send no token, or a token with the package read scope. pip takes a token in the URL: https://<user>:<token>@git.kinedb.com/api/packages/kinedb/pypi/simple. uv takes the environment variables UV_INDEX_KINEDB_USERNAME and UV_INDEX_KINEDB_PASSWORD.

The version shape is the one of kinedb-client: 0.0.0.dev<N>+g<sha>, the commit that built it (see Versions are commits, not releases). No version promises compatibility yet, and no table lists the server versions that a sync supports. Use a kinedb-embedded and a kinedb server of the same commit: the image kinedb:<sha> and the package 0.0.0.dev<N>+g<sha>.

Open a database

import asyncio
from kinedb.embedded import KineDB

async def main():
    async with KineDB("shop", dir="data") as db:    # the files are in ./data/shop
        await db.sql("CREATE TABLE IF NOT EXISTS orders (id INT PRIMARY KEY, item TEXT)")
        await db.sql("INSERT INTO orders (id, item) VALUES (1, 'pen')")
        print(await db.sql("SELECT id, item FROM orders"))

asyncio.run(main())

• KineDB() with no name opens an IN-MEMORY database. It writes no file.
• KineDB(name, dir=...) opens a persistent database in the directory dir/name. The process must have the permission to write to dir. With no dir, a relative name is relative to the working directory.
• CREATE TABLE IF NOT EXISTS does nothing when the table is there, so the sample can run again.
• memory_budget= sets the memory the engine may use, in bytes. The minimum is 4 MiB. A persistent database uses 32 MiB when you give no budget.
• durability= is "commit" (the default) or "manual". See Durability.
• ssl= is the ssl.SSLContext of the sync socket (see The sync).
• retry= controls the hold retry of a call() that goes to the server (B2298). With no option, a typed hold refusal (write_hold or not_ready: nothing was written) is sent again after the retry_ms of the server, for up to 20 s. Every other error raises at once. retry=False turns the hold retry off. The refusal outcome_unknown raises KineDBOutcomeUnknownError and is never sent again: the write may have applied (B2304).
• reconnect= controls the automatic reconnect of the sync (B2208). See When the connection closes.
• async with closes the database at the end of the block. db.close() is a plain call. close() does NOT flush.

Durability

• In "commit" mode (the default), a statement that commits answers only after its data is on disk. A crash of the process or a loss of power after the answer loses nothing.
• In "manual" mode, a write is durable only after await db.flush(). A loss of power or a crash of the operating system can lose every write since the last flush. It is faster for many small writes.
• An offline write to a synced table keeps its record for push() on disk too. A new process that opens the database sends it with push() and the url; it needs no sync() first.
• await db.flush() makes every write durable. In "commit" mode it has no work.
• A failed fsync of the data files stops the database (B2960). The disk can lose the data of a failed fsync, and a second fsync can still report success, so kinedb does not try again. The failed call raises KineDBError. Its text starts with panic: and holds [B2960_SEGMENT_FAIL_STOP]. After a panic, the library stops the thread (B1862). Every later call of the library on that thread raises KineDBError, also a read and a new open. close() does not raise, but it frees nothing: the database keeps its lock. So a second open in the same process also fails. Restart the process. The open replays the log up to the last durable commit. Before it reads a file, the open syncs the file and drops its page cache (B2966), so it reads the bytes that the disk holds.

One process for one database

Two processes cannot open the same database. While a KineDB holds dir/name, a second open of dir/name raises KineDBError, from another process and from the same process. The error names the directory and the pid of the holder:

B2150: the database directory /var/lib/app/shop is already open (the lock /var/lib/app/shop/LOCK is held by process 4242). Close the other opener first; two openers on one directory corrupt it.

Close the first KineDB, then open again. When the holder process dies, also after a crash or a kill -9, the operating system frees the lock. Do not delete the LOCK file.

sql

await db.sql(statement) runs ONE statement. The result is the dict of kinedb-client:

{"type": "rows", "columns": ["id", "item"],
 "types": [{"code": 2, "name": "Int", "param": 0}, {"code": 4, "name": "Text", "param": 0}],
 "data": [[1, "pen"]], "count": 1, "schema_id": None, "table_versions": [["orders", 3]]}
{"type": "created"}
{"type": "inserted", "count": 1}      # also "updated" and "deleted"

A failed statement raises KineDBError with the message of the engine. rows_as_objects(result) gives the rows as dicts. sql("BEGIN"), sql("COMMIT") and sql("ROLLBACK") are refused: a transaction is ONE transaction() call (see transaction).

WATCH and UNWATCH raise KineDBError (B2198). A watch on the embedded database is not available yet (B2196). To receive the changes of a server table, use db.sync() with on_change.

Give a value from outside your program as a parameter, never as joined text. sql(statement, params) binds $1, $2, … with the binder of the server:

await db.sql("SELECT id, item FROM orders WHERE item = $1", [item])

bool is BOOL, int is INT (64 bits), float is FLOAT, str is TEXT, datetime with a time zone is TIMESTAMP, bytes and bytearray are bytes, and None is NULL. A list or a tuple of scalars of one type is an ARRAY value. It follows the rules of the direct client. A ? placeholder is not a parameter. See Parameters.

The schema version of a table in the embedded engine

A rows result has table_versions: a list of [table, version], one pair for each table that the statement read. The name is the name of kinedb-client.

r = await db.sql("SELECT * FROM orders")
# r["table_versions"]  ->  [["orders", 3]]

• The version is the stamp of the last DDL on the table in this database. It is an int.
• The version changes at each DDL of the table, also a GRANT or a change of a policy. It never changes at a row write.
• 0 or an absent value means that the version is not known. table_versions is [] then.
• The version of an embedded database is not the version of a server. The two counters are separate. Do not compare an embedded value with a server value of the same table.

transaction

await db.transaction(items, token=None) runs ONE complete transaction. Every item commits, or none of them. The same call works on kinedb.client.

await db.transaction(
    [
        ["INSERT INTO orders (id, item) VALUES ($1, $2)", [7, "pen"]],
        {"put_file": ["files", "receipt-7.pdf", data]},
        "UPDATE stock SET n = n - 1 WHERE item = 'pen'",
    ],
    token="order-7",
)

• An item is a statement (a str, [sql] or [sql, params]) or a file ({"put_file": [collection, name, data]}, the arguments of put_file()). A file and a row of one call commit together.
• The answer is {"type": "created"}. A call with file items also answers "files": the record of each file item, in list order, the shape that put_file() answers.
• When one item fails, the call raises KineDBError with its error, and nothing of the call is written. An item of a wrong shape raises TypeError or ValueError before anything runs.
• A SELECT in the list reads the committed rows only, and gives no rows back. DDL cannot be an item.
• token makes a retry safe: a call whose token committed before applies nothing and answers created.

See Transactions.

Files in the embedded engine

await db.sql("CREATE TABLE files (name TEXT PRIMARY KEY)")
with open("report.pdf", "rb") as f:
    await db.put_file("files", "report.pdf", f)      # the table, then the file name
data = await db.get_file("files", "report.pdf")

stream = await db.get_file_stream("files", "report.pdf")
if stream is not None:                               # None for a missing file
    async for chunk in stream:
        print(len(chunk))

• The first argument of put_file, get_file and get_file_stream is the table. It has a TEXT primary key, or it is a schemaless collection. The second argument is the file name, which is the primary key of the row.
• put_file takes bytes, a binary file object, an iterable or an async iterable of bytes. The engine takes the data in pieces of at most 1 MiB, so a file larger than the memory stores.
• get_file raises KineDBError for a missing file. get_file_stream gives None.
• get_file_stream gives the file one verified chunk at a time (at most 128 KiB). The engine checks each chunk and the whole file against their hashes.
• A file in a SYNCED table goes to the server with the next push(), as a row does. The report of push() counts its chunks.
• The keyword arguments db, branch and timeout_ms are the ones of kinedb-client. The embedded engine is ONE database, so db and branch must be None.

The sync: an offline-first copy of a table

sync() fills a local table from a running kinedb server, and it answers when the fill is COMPLETE. After that the local database answers reads on its own, also with the network gone. A write to the synced table works offline, and push() sends it to the server.

import os

creds = dict(url="ws://kinedb.example:4820/ws", user="alice",
             password=os.environ["KINEDB_PASSWORD"])

async def changed(table, report, error):
    print(table, report["upserts"], report["removes"], error)

async with KineDB("tenant", dir="data") as db:
    await db.sync("orders", **creds, on_change=changed)

    # The rows are local now. This read needs no server.
    await db.sql("SELECT id, item FROM orders")

    # A write while the server is away runs locally, and it is on disk with its record.
    await db.sql("INSERT INTO orders (id, item) VALUES (10, 'ink')")

    # Back online: send the recorded writes. Then look at what the server refused.
    report = await db.push("orders", **creds)
    if report["conflicts"] or report["rejected"]:
        for entry in await db.refused("orders"):
            print(entry["pk"], entry["reason"])

• The credential is one of four: user with password, or token, or user with key_id and sign (see Log in with a key pair), or anonymous=True.

• sync() keeps a LIVE subscription. Each later change on the server lands in the local table by itself, and on_change(table, report, error) hears it. on_change can be a function or a coroutine function, as in the sample. on_change hears the live changes and the fill of a table that a new session resumes. It never hears a push or a conflict: read the report of push() and refused().

• A cancel of a sync() call, or a wait_for timeout, stops only the wait of your code. The sync goes on in the database. When it ends with success, the table is live and on_change hears the later live changes, as if the call had returned. The fill report of the cancelled call reaches nobody: on_change hears only the later changes and resumes. When the sync fails, the table keeps the on_change that it had before the call. But a table that was live can lose its tail in the failed sync. Then the database owes the table a resume, and the next sync() or push(), or the automatic reconnect, resumes it. Until the cancelled sync ends, the session refuses each new sync with an "already running" error, also sync(table, live=False). Then try the call again.

• A cancel of a push() call, or of a sync() call while it resumes the live tables, also stops only the wait of your code. The push or the resume goes on. When it ends, connection_state follows the tables that the session still owes a resume: "online" when it owes none.

• sync() with live=False stops the live tail of the table, as live: false does in JavaScript (see Stop the live tail of a table). After it, a change on the server does not reach the local table, and on_change is not called for the table. The tail stops when the database starts the call, also when the call fails after that. A call that the database refuses before it starts keeps the tail and on_change, for example a call while another sync of the session runs. A table that holds a query window stays live and keeps its on_change: a window needs the tail. When the session closes during the call, the call fails, and the table loses its on_change, also a table with a window.

• The sync reads only the rows that the user may read: the grants of the table and its row policies apply.

• await db.freshness("orders") gives the state of the local copy: never_synced, filling, complete_as_of, offline_stale or gapped, and the token of the last complete fill. Read token together with state. gapped means that the server sent a gap answer (see WATCH limits): the sync is live from token, but the changes between gap["since"] and gap["token"] are missing. A row that the user lost in that time must not stay, so the sync never keeps the rows on a gap (B3443). It deletes the synced rows of the table and fills again from SINCE START, once. The pending writes stay. When that fill also gets a gap, the table stays EMPTY and gapped. A gap on the live tail deletes the rows too, and the next sync fills from SINCE START. The report of the sync names the same gap in report["gap"] (None when there is none). To get a complete copy again, call await db.sync("orders", full=True). That sync fills the table from SINCE START, replaces the local rows and clears the gap.

• After a restart of the process, no live watch keeps a table current. So the open of the sync session sets each current subscription to offline_stale, with its token (B3824). Until you call sync() for the table, freshness() answers offline_stale, and a query that the local rows cover answers the old rows as offline_stale, with no server call.

• The sync opens its OWN connection to the server.

• The fill comes in pages of up to 4 MiB. The sync asks the server for a window of 4 pages (B3033), as the JavaScript sync does. So at most 16 MiB of fill that is not committed is in memory. The client commits each page and its resume token in one local transaction, then it acknowledges that page. A crash after the commit loses no row: the next sync() resumes from the committed token. A server from before B3026 refuses the window. Then the session fills at window 1 and writes one warn line to stderr (see Troubleshooting).

• A wss:// url uses TLS through the standard library ssl. For a server whose certificate the default context does not trust, give a context:

  import ssl
  db = KineDB("tenant", dir="data", ssl=ssl.create_default_context(cafile="ca.pem"))

The first fill is a normal fill: Python takes no pack

The README of the JS package names a PACK (B3431) as a first fill of an empty device: one file stream of the server database. The Python package does NOT take a pack. An empty device asks for a normal fill: SHOW COLUMNS, then WATCH <table> SINCE START. It does this also when the server can answer a pack. The device sends no Pack request, so it sees no "fresh pack" refusal and it has no .pack-incomplete marker to handle. The pack in the Python package is a later issue (#6470). The test tests/test_b3472_no_pack.py fails when the package starts to send a pack request.

Sync a table from another server database

The sync reads the server database default. Pass database to sync a table from another server database:

await db.sync("orders", url="ws://127.0.0.1:8080/ws", token=token, database="scale_demo")

Each frame of the sync session names that database. The value must be a non-empty string. Another value is a TypeError, and no socket opens. All five methods take database: sync, push, freshness, refused and discard_refused.

One local database syncs with ONE server database. The first sync binds it, and a later call that asks for another one is refused with this text:

sync target mismatch: this local database syncs with server database 'scale_demo'; the call asks for 'default'. Use a new local database to sync 'default'.

• A call with a url or a database asks for database, else for default. So on a local database bound to scale_demo, pass database="scale_demo" with each url.
• A call with no url and no database uses the live session as it is.
• The automatic reconnect opens the new session with the same server database.
• To sync another server database, open a new local database for it. The rules are the same as in JavaScript.

Log in with a key pair

A user can log in with a P-256 key pair instead of a password (B2813). The server holds the public key. The statement ALTER USER app ADD PUBLIC KEY '<hex point>' adds it (see the key login in SQL). Give user, key_id and sign to sync, push, freshness, refused and discard_refused. The same keywords log in the statement link of call(). The web console shows the private key one time, as a .pk8.pem file.

• key_id is 64 hex characters: the lower-case SHA-256 of the 65-byte uncompressed public point (04 || x || y). It is NOT the hash of the SPKI.
• sign(message: bytes) answers the signature of message: ECDSA P-256 with SHA-256, as exactly 64 RAW bytes r || s (32 + 32, big-endian). It can be a plain function or an async def.
• The package holds no key bytes, and it never sends or logs the private key. sign holds the key, for example in a hardware token or a key service.

The package has no runtime dependency. This example uses the package cryptography, which signs in the DER form. decode_dss_signature changes DER into r and s:

import hashlib
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature

key = serialization.load_pem_private_key(pem_bytes, password=None)
point = key.public_key().public_bytes(
    serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint
)
key_id = hashlib.sha256(point).hexdigest()

def sign(message: bytes) -> bytes:
    r, s = decode_dss_signature(key.sign(message, ec.ECDSA(hashes.SHA256())))
    return r.to_bytes(32, "big") + s.to_bytes(32, "big")

await db.sync("orders", url="ws://kinedb:4820/ws", user="app", key_id=key_id, sign=sign)

Each login asks the server for a new challenge. The package calls sign once with the bytes of that challenge and sends the signature once. When the socket closed or a new login started while an async sign ran, the package drops the signature.

The errors:

• A wrong key and an unknown user get the same server text: invalid username or password.
• A sign that raises gives KineDBError("the key did not sign"). The error of sign is only its __cause__. It is not in str(), repr() or response.
• A sign that does not answer exactly 64 bytes gives a KineDBError that names the 64 bytes. A DER signature (about 70 bytes) gets this error.
• An async sign has 30 s to answer (the challenge validity of the server). After that, the login fails with KineDBError("the key did not sign in time"), and a later signature is dropped, never sent. A plain (sync) sign runs on the event loop thread, so the time limit cannot stop it and it blocks the loop. Use an async sign for a slow signer, for example async def sign(m): return await asyncio.to_thread(blocking_sign, m).
• A server that is too old for a key login gives the text of the engine.
• A failed key login stops the automatic reconnect, as a refused password does. A retry would call the same sign, and a hardware token can lock after some failed attempts. A new sync() or push() with the credential starts it again.

A reconnect keeps sign, never a signature. Each reconnect signs a NEW challenge.

When the connection closes

When the socket closes, the local table becomes offline_stale. A table whose fill or diff stopped after it wrote a part of its rows stays filling: its rows match no commit, so no query of it is covered. The local reads and writes still work.

The library connects again by itself (B2208). When the socket of a session with a live table closes, the database waits, then opens a new socket with the same url and credential, and it resumes every live table from its local state. The delay before an attempt is min(max_delay_ms, min_delay_ms * 2^n) * j: n is the count of failed attempts, and j is a random value from 0.5 to 1. The defaults are 500 ms and 30 s. A success sets n to 0 again.

• For each live table, the attempt first sends the recorded writes of the table (as push() does), then resumes the table. So an offline write reaches the server with no call from your application.
• A table that does not resume makes the attempt fail. The next attempt uses the open session and resumes only the tables that did not resume yet. The state is "online" only when every live table resumed.
• When the server refuses the credential (for example a changed password or an expired token), the attempts stop. The state is "offline", and detail holds stopped: True and the error. A later sync() or push() with a valid credential connects again.
• While an attempt runs, the other calls that use the server (sync(), push()) wait for it.

db = KineDB("tenant", dir="data", reconnect={"min_delay_ms": 250, "max_delay_ms": 10_000})
unsubscribe = db.on_connection_state(lambda state, detail: print(state, detail))
print(db.connection_state)  # "offline" before the first connect

• connection_state is "online" (a session is open, logged in, and its resume finished), "reconnecting" (an automatic attempt runs now) or "offline".
• on_connection_state(listener) calls listener(state, detail) at each change, never with the same state two times in a row. detail holds attempt, delay_ms and error when they apply. It answers a callable that removes the listener.
• reconnect=False turns the automatic reconnect off. close() stops it.
• open_timeout_ms in the reconnect dict limits the open of a session: the socket and the login. The default is 30 000 ms. At the limit, the half-open session closes, and the attempt fails and waits for the next delay. A sync() or push() that opens a session uses the same limit.
• A login error that the server marks as retryable (for example the root user is not seeded yet, or a fault of the node that it finds before it checks the password) does not stop the attempts. A fault after the password check gets the normal refusal, so that the answer never tells a correct password from a wrong one; that refusal stops the attempts.
• Known limits: the time limit applies to the open only. A push or a resume that gets no answer on an open socket blocks the attempt (B2548). A resync can start between two steps of an application sync() or push(), and the step then fails with "already running" (B2547); call it again.

Your application can also call sync() or push() with the url and the credential. That call cancels the pending automatic attempt, opens a new socket, sends the recorded writes, and resumes every live table from its local state. A call while the server is away raises KineDBError with retry set in its response: wait, then call again.

After a failed call or a failed attempt, the offline session stays. So freshness(), refused() and discard_refused() still answer with no url, and freshness() answers offline_stale (or filling for a table whose fill stopped after a part of its rows).

A local write is not a server decision

The local engine decides a local write. The server decides it only at push(). So the count of a local UPDATE … WHERE or INSERT tells you about the LOCAL row. A local UPDATE can answer count 1 for a row that the server changed already; the push then reports a conflict, and the server keeps its row. Do not start an action that you cannot undo on a local count. When the action needs the server's answer, call push() and read its report first.

The report of push()

push() answers a dict:

Key	The value
table	the table
sent	the recorded writes that this push sent
applied	the writes that the server applied
merged	the writes that a conflict function merged with the server's row (WITH CONFLICT POLICY FUNCTION)
conflicts	the writes over a row that changed on the server after the local copy read it
rejected	the writes that the server refused for a reason that is not a conflict; refused() gives the reason
outcomes	one dict for each write: status, seq, replayed, and the keys that the status has (version, server_version, reason). A conflict has no reason here: read it in refused()
chunks_total, chunks_held, chunks_sent, chunk_bytes_sent, chunk_errors	the file chunks of the push. chunk_errors holds the final refusals of the chunk upload; the write still goes out and the server decides it. A retryable refusal of the upload raises instead, and the next push() sends only the chunks that did not land

A second push() with nothing new sends 0 writes.

Refused writes

The server judges each pushed write by the conflict policy of the table (see Conflict policy). Under REJECT, the default, a write over a row that changed on the server after the local copy read it is a CONFLICT. The server keeps its own row.

A write that the server did not apply is never dropped in silence. It stays in refused(), with the server's reason, until you discard it:

for entry in await db.refused("orders"):
    print(entry["seq"], entry["pk"], entry["reason"])
    # Act on the entry here: read the server's row with sync(), then write again.
    await db.discard_refused("orders", entry["seq"])   # the ONLY call that removes one

An empty list runs no loop. seq, read_version and write_hlc are DECIMAL TEXT, because each one is a 64-bit number. A refused write stays in the local table: a refusal is not a rollback.

One thread for one database

The engine keeps its databases, its connections and its sync sessions per THREAD. A database belongs to the thread that opened it, which is the thread of its event loop. A call from another thread raises KineDBError. Open one database on each thread that needs one. The calls do their work on the thread of the event loop.

The library file

The package loads libkinedb_wasm.so from the wheel. The name comes from the crate kinedb-wasm, which builds the browser module and this native library from one source. The file is a native Linux library, not WebAssembly.

The sync needs the REMOTE build of the library: the build with the remote and sync exports (--features remote). The published wheel carries the remote build. This line prints True for it:

from kinedb.embedded._abi import load_library; print(load_library().has_remote)

With the other build, sync() and push() raise KineDBError. Its message contains needs the remote and sync ABI and It is the LOCAL build.

The environment variable KINEDB_EMBEDDED_LIB names another library file explicitly.