Metadata-Version: 2.5
Name: kinedb-embedded
Version: 0.0.0.dev9658+g3334047b4cb0
Summary: The embedded kinedb engine in the process. The native library through ctypes.
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-embedded

The kinedb engine in the Python process. Import it as `kinedb.embedded`.

The package loads the native library `libkinedb_wasm.so` (the crate
`crates/kinedb-wasm`) with the stdlib `ctypes`. It has no runtime dependency.

```python
import asyncio
from kinedb.embedded import KineDB   # the same lines run with kinedb.client

async def main():
    async with KineDB("shop", dir="data") as db:   # KineDB() is in memory
        await db.sql("CREATE TABLE orders (id INT, item TEXT, PRIMARY KEY (id))")
        await db.sql("INSERT INTO orders (id, item) VALUES (1, 'apple')")
        print(await db.sql("SELECT * FROM orders"))
        await db.put_file("files", "a.bin", open("a.bin", "rb"))
        async for chunk in await db.get_file_stream("files", "a.bin"):
            ...
        # each commit is durable; durability="manual" needs `await db.flush()`

asyncio.run(main())
```

The surface is async, as in `kinedb.client`. The engine runs on the thread of
the event loop.

## Parameters

`sql(text, params)` takes the values of `$1`, `$2`, ... apart from the text.
The engine binds them, so a value is never parsed as SQL. The same call works
on a synced table.

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

The type table is the one of `kinedb.client`: `bool` -> BOOL, `int` -> INT,
`float` -> FLOAT, `str` -> TEXT, an aware `datetime` -> TIMESTAMP
(microseconds), `bytes` or `bytearray` -> BYTES, `None` -> NULL. Any other
value, an `int` outside the i64 range, a `float` that is not finite and a
naive `datetime` raise `TypeError` or `ValueError` before the statement runs.
`_params.py` is a verbatim copy of the client module, and
`tests/test_params_copy.py` fails when the two copies differ.

## Transactions

A transaction is ONE call (B2174). There is no open transaction:
`sql("BEGIN")`, `sql("COMMIT")` and `sql("ROLLBACK")` raise `KineDBError`, and
the error names `transaction()`. The same call works on `kinedb.client`.

```python
done = await db.transaction(
    [
        {"put_file": ["files", "a.bin", data]},
        ["INSERT INTO meta (id, fname) VALUES ($1, $2)", [1, "a.bin"]],
    ],
    token="save-1",
)
# {"type": "created", "files": [{"size": ..., "hash": ..., "chunks": [...]}]}
```

An item is a statement (a str, `[sql]` or `[sql, params]`) or a file
(`{"put_file": [collection, name, data]}`, the arguments of `put_file`). Every
item commits in one commit, or none of them; a file and a row commit together.
`token` makes a retry safe: a call whose token committed before applies nothing.
On a synced database the op-log entries of one call form one group, and `push`
sends the group whole.

## Functions: `call()`

`call(name, args, offline="preview")` runs a SERVER function (B2173). The
rules of `args` and of the result are the ones of `kinedb.client`
(`_call.py` is a verbatim copy of its module, and `tests/test_params_copy.py`
checks it too).

```python
await db.sync("orders", url="ws://127.0.0.1:8080/ws", user="app", password=pw)
await db.call("add_points", [1, 50])            # 51, from the server
await db.call("add_points", [1, 50], offline="refuse")
```

ONLINE is an open sync session whose socket is up. The call then goes to the
server on a second socket to the url of the session, with its credential.
The sync socket cannot carry it: the sync pump drops every frame that is not
a change of a synced table.

OFFLINE (no session, or its socket went away) `offline` decides. `"refuse"`
raises `KineDBError` "needs the server". `"preview"` (the default) and
`"queue"` raise `KineDBError` "needs the server" too: the embedded database
cannot run or queue a function yet (B2175). Any other value raises `ValueError`.

An ONLINE call that gets a typed hold refusal (`refusal["kind"]` `write_hold`
or `not_ready`) is sent again for you (B2298). The server sent the refusal
before the write took effect, so nothing was written. The link waits the
`retry_ms` of the server between two attempts, for up to 20 s from the first
send. Every other error raises at once, as before. `KineDB(..., retry=False)`
turns the hold retry off.

An ONLINE call that gets the refusal `outcome_unknown` raises
`KineDBOutcomeUnknownError`, a subclass of `KineDBError` (B2304). The server
sent the call on, but the reply was lost, so the write MAY have applied. The
link never sends the call again. Read the data back before you call again.

## Describe a table: `describe()`

`describe(table)` reads the structure of a table (B3096). It answers one dict
with the keys `table`, `schema_version` (an int), `primary_key`, `columns` and
`indexes`. The shape and the rules are the ones of `kinedb.client`
(`_describe.py` is a verbatim copy of its module, and `tests/test_params_copy.py`
checks it too).

```python
d = await db.describe("users")
d["primary_key"]                         # ["id"]
d["columns"][1]["unique"]                # True: a UNIQUE index of this one column covers it
```

The call runs `DESCRIBE`, `SHOW INDEXES FROM` and `DESCRIBE` again on the local
database through `sql()`. It works offline. When the `schema_version` of the two
`DESCRIBE` replies differs, the call starts again. After 3 reads it raises
`KineDBError`.

## The sync

A local table follows a table of a remote kinedb. The engine runs the sync
protocol; the package owns the WebSocket (asyncio, `wss://` through the stdlib
`ssl`). The library must be the remote build (`--features remote`).

```python
async with KineDB("tenant", dir="data") as db:
    creds = dict(url="ws://kinedb:4820/ws", user="app", password="...")
    await db.sync("orders", **creds, on_change=lambda table, report, error: ...)
    await db.sql("INSERT INTO orders (id, item) VALUES (10, 'pen')")   # works offline
    await db.push("orders", **creds)        # after a lost socket: the Online switch
    print(await db.freshness("orders"))     # never_synced / filling / complete_as_of / offline_stale / gapped
    print(await db.refused("orders"))       # the writes the server refused, with its reason
```

`sync()` reads the server database `default`. Pass `database="name"` to sync another
one. All five methods take it. One local database syncs with ONE server database: a call
that asks for another one is refused with `sync target mismatch: …` (B3030).

`sync()` keeps a LIVE tail by default: a change on the server lands locally
and `on_change` hears it. A push of a stale write is refused (the conflict
policy `REJECT`) and stays in `refused()` until `discard_refused()`.

### The first fill is a normal fill

The README of the JS package names a PACK (B3431) as a first fill of an empty device.
This 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 no query is refused as "fresh pack", and there is no
`.pack-incomplete` marker to handle. The pack here is a later issue (#6470).
`tests/test_b3472_no_pack.py` fails when the package starts to send a pack request.

### Log in with a key pair

A user can log in with a P-256 key pair instead of a password (B2813). The
server knows the public key: `ALTER USER app ADD PUBLIC KEY '<hex point>'`.
Give `user`, `key_id` and `sign` to `sync`, `push`, `freshness`, `refused` and
`discard_refused`. The same keywords log in the statement link of `call()`.

- `key_id` is 64 hex characters: the 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 awaits an awaitable answer.
- The package holds no key bytes. `sign` holds the key, for example in a
  hardware token or a key service.

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

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

The login, step by step: the package sends a challenge request with a new
32-byte nonce from `os.urandom`. The server answers with its own nonce. The
package calls `sign` once with the bytes of that challenge, and sends the
signature once. A signature never goes out for an old challenge: 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__`: its text can hold a secret, so 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 calls the same `sign`, and a hardware token can lock after
  some failed attempts. A `sync()` or `push()` with the credential starts it
  again.

A reconnect (the `reconnect` option) keeps `sign`, never a signature. Each
reconnect signs a NEW challenge.

## The library

1. `KINEDB_EMBEDDED_LIB` names the file explicitly.
2. Else the package loads the copy that a platform wheel bundles at
   `kinedb/embedded/_lib/libkinedb_wasm.so`.
3. Else the open raises `KineDBError` and names both places.

## The thread rule

The registries of the native library are thread-local. A database belongs to
the thread that opened it: the thread of its event loop. A call from another
thread raises `KineDBError`. Open one database on each thread that needs one.

## The tests

```sh
cargo build --release -p kinedb-wasm      # through scripts/box.sh in this repo
cd sdk/python/embedded && uv run --with pytest pytest
```

The tests set `KINEDB_EMBEDDED_LIB` to `$CARGO_TARGET_DIR/release/` or
`<repo>/target/release/`.
