Metadata-Version: 2.5
Name: kinedb-client
Version: 0.0.0.dev10044+g26b670e8a40f
Summary: The direct client of kinedb. JSON over HTTP, binary TLV over WebSocket.
Project-URL: Homepage, https://git.kinedb.com/kinedb/kinedb
Project-URL: Source, https://git.kinedb.com/kinedb/kinedb
License: UNLICENSED
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# 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 |

```python
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
<!-- #region 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:

```toml
[[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`.

<!-- #endregion install -->

<!-- #region versions -->
### 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.

<!-- #endregion versions -->

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
<!-- #region use -->

```python
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:

```python
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:

```python
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:

```python
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"}`.

```python
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:

```python
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:

```python
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.

```python
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:

```python
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`.

```python
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:

```python
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

```python
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.

```python
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.

```python
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.

```python
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:

```python
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 `SET`s, 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.

```python
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").

```python
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)`:

```python
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:

```python
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):

```python
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.

<!-- #endregion use -->

## 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.
