kinedb-client (0.0.0.dev8363+geb695ea16dbd)
Installation
pip install --index-url https://git.kinedb.com/api/packages/kinedb/pypi/simple/ --extra-index-url https://pypi.org/simple kinedb-clientAbout this package
The direct client of kinedb. JSON over HTTP, binary TLV over WebSocket.
kinedb-client
The direct client of kinedb: JSON over HTTP, binary TLV over a WebSocket. It
speaks to a running kinedb-server.
It is the Python twin of @kinedb/client. The wire is the same, the API surface
is the same, and the names are the same in snake case.
The two shapes of every kinedb SDK
kinedb ships each SDK twice, in every language. The names are fixed, and the API surface is identical, so an application changes ONE import line and nothing else.
| role | Rust crate | npm | distribution (pip install) |
import | status |
|---|---|---|---|---|---|
| wire types + TLV/frame codec | kinedb-protocol |
@kinedb/protocol |
kinedb-protocol |
kinedb.protocol |
not split out yet |
| direct client (remote server) | (none yet) | @kinedb/client |
kinedb-client |
kinedb.client |
this package |
| embedded engine + sync | kinedb |
@kinedb/embedded |
kinedb-embedded |
kinedb.embedded |
built (B1862), Linux x86_64 |
from kinedb.client import KineDB # talks to a remote kinedb-server
from kinedb.embedded import KineDB # talks to a local engine that syncs
kinedb-embedded bundles the kinedb engine (B1862). The SDK talks to that local
embedded database, and the embedded database syncs with a real kinedb elsewhere.
It runs on Linux x86_64 for now (sdk/python/embedded).
Two names, one namespace
Python gives every package two names, and PyPI does not link them. The
distribution name is what pip install takes, and it mirrors the crate name
character for character. PEP 503 treats kinedb-client, kinedb_client and
kinedb.client as one name, so any of the three installs this package; the
hyphen is the form we write.
The import name lives under one namespace package, kinedb, which is the
Python form of the @kinedb/ npm scope.
No kinedb distribution may ever ship kinedb/__init__.py. kinedb is a
PEP 420 namespace package: a directory with NO __init__.py. A regular package
there shadows the namespace and breaks every other kinedb-* install in the
same environment. tests/test_namespace.py enforces the rule.
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.
Publishing: the forge publishes on every merge, like the image. By hand, from the repository root:
bun scripts/sdk-publish.ts sdk/python/client
It refuses a dirty tree, because a commit-named build must name a commit that actually holds the published bytes. The newest 10 versions are kept.
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_idis 64 hex characters: the SHA-256 of the 65-byte uncompressed public point.key_id_from_public_point(point)calculates it.sign(message: bytes)returnsbytes, or an awaitable ofbytes. It signsmessagewith 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
signthat raises givesKineDBError("the key did not sign"). The error ofsignis 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)
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 ofsql(). - No transaction stays open on the server.
BEGIN,COMMITandROLLBACKare not statements of a transaction, andsql()with one of them is refused. Twotransaction()calls on one socket are two separate transactions. - The answer is
created, as for a write withsql(). When a statement fails, the server writes nothing and the call raises aKineDBError. - A
SELECTinside a transaction gives no rows. For a read-modify-write, put the read and the write in a function and run it withCALL(see the SQL page, § Functions and tickers). tokenis the commit token. The server stores it in the same commit as the rows. A client that lost its connection asksGET 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
retryargument has the shape of the constructor option and wins over it (retry=Falseturns it off). A retry sends the WHOLE transaction again, never a part of it.dbandbranchwin over the default of the client. Over HTTP a branch other thanmainis refused, as forsql(): use the socket. - The transaction holds no lock. At the commit the server checks each row that
an
UPDATE, aDELETEor anINSERTover an existing key started from. When another client wrote that row first, the server writes nothing and answerscalvin conflict: … — nothing applied, retry, with the retry flag set. AnINSERTof a new key is not checked: when two transactions insert the same key, the later write wins. To claim a key, useINSERT … IF NOT EXISTS, also inside a transaction: a key that exists refuses the whole transaction withduplicate key:. A blindUPDATE(WHERE <primary key> = <value>, only literalSETs, 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 is0x1E.
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 ONEINSERT … IF NOT EXISTS: one present key refuses the whole statement. A create with a key that the ORM made (AUTO) is a plainINSERTin the transaction. A create with your key beside another change raisesBackendUnsupportedError(create_in_transaction): the ORM does not sendIF NOT EXISTSinside a transaction yet. The engine accepts it there since B3168.mode="upsert"is a plainINSERT: it replaces the row at the key. A model with a counter refuses it (counter_upsert).mode="ignore"raisesBackendUnsupportedError(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.
Test
cd sdk/python/client && KINEDB_SERVER_BIN=<path of the kinedb server> uv run --with pytest pytest
make sdk-py-test builds the server of the tree and sets KINEDB_SERVER_BIN.
The QueryAs test (tests/test_b2886_query_as.py) starts that server. Without
KINEDB_SERVER_BIN the test fails and names the build; it never skips.
pytest is the only development dependency. There is no pytest-asyncio: an
async case runs under asyncio.run(...) through the async_test helper in
tests/conftest.py.
A live test runs only when a server is named:
KINEDB_URL=http://127.0.0.1:4820 uv run --with pytest pytest tests/test_live.py -v
The wire format this package mirrors is the Rust one,
crates/kinedb-protocol/src/{frame,tlv,codec,messages}.rs. That is the truth;
this package follows it. The JavaScript client, sdk/js/client/src/index.js, is
the validated reference implementation of the same wire.