Python Client

The Python client is a pure standard-library driver (socket and struct only, no third-party dependencies) that speaks the VeltrixDB binary protocol via a single connection or a thread-safe pool.

Install

bash
pip install veltrixdb

Requires Python 3.8+. No external dependencies.

Quick start

python
import veltrixdb
db = veltrixdb.Client(host="localhost", port=9000)
db.put("user:1001", "alice")
print(db.get("user:1001"))  # alice

The library's own example (python/example.py) uses the class directly as a context manager, which is the recommended pattern since it guarantees the socket is closed:

python
from veltrixdb import VeltrixDBClient

with VeltrixDBClient('127.0.0.1', 9000) as c:
    c.ping()

    c.put('hello', b'world')
    val = c.get('hello')
    print(f"PUT+GET  hello = {val.decode()}")

    c.delete('hello')
    gone = c.get('hello')
    print(f"GET after DEL  hello = {gone!r}")  # None

put requires bytes for the value — encode strings yourself (.encode()) before writing.

Config

python
VeltrixDBClient(
    host='127.0.0.1',
    port=9000,
    timeout=10.0,          # per-operation timeout
    connect_timeout=5.0,   # connection timeout
    ssl_context=None,      # None = no TLS
)

API surface

Core key-value

Batch operations

python
# Batch write — single round trip for N keys
entries = [
    ('user:1', b'{"name":"Alice","age":30}'),
    ('user:2', b'{"name":"Bob","age":25}'),
    ('user:3', b'{"name":"Carol","age":35}'),
]
errors = c.multi_put(entries)
failed = [i for i, e in enumerate(errors) if e is not None]

# With per-entry TTL (seconds)
errors = c.multi_put(
    [('session:abc', b'token-xyz')],
    ttls=[60],   # 60-second expiry
)

# Batch read — single round trip for N keys
keys = ['user:1', 'user:2', 'user:3', 'user:missing']
values = c.multi_get(keys)
for key, val in zip(keys, values):
    if val is not None:
        print(f"MultiGet  {key} = {val.decode()}")
    else:
        print(f"MultiGet  {key} → not found")

Namespaces

Hash fields (per-field TTL)

Atomic operations

New in v1.1.0 — typed atomic ops (previously Rust/C++ only). Every method takes an optional ttl in seconds for the stored value (-1 = no expiry):

python
from veltrixdb import KeyNotFoundError

# Counter — missing key starts from 0
n = c.incr('counter', 5)          # 5
n = c.decr('counter', 2)          # 3

# Compare-and-swap (values are bytes)
try:
    if c.cas('hello', b'world', b'earth'):
        print('swapped')
    else:
        print('mismatch — current value was not b"world"')
except KeyNotFoundError:
    print('key does not exist')

# Set-if-absent with a 30-second TTL (lock-style)
if c.setnx('lock', b'owner-A', ttl=30):
    print('lock acquired')

Connection pool (thread-safe)

example.py demonstrates the pool under real concurrent load — 20 threads sharing 8 connections:

python
from veltrixdb import VeltrixDBPool
import threading

pool = VeltrixDBPool(
    host='127.0.0.1',
    port=9000,
    pool_size=8,         # 8 connections shared across all threads
    acquire_timeout=5.0,
)

def worker(i: int) -> None:
    with pool.acquire() as c:
        c.put(f'thread:{i}', f'value-{i}'.encode())

threads = [threading.Thread(target=worker, args=(i,)) for i in range(20)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# Functional style
val = pool.execute(lambda c: c.get('user:1'))
pool.execute(lambda c: c.put('pool_test', b'hello from pool'))

pool.close()

pool.acquire() is a context manager: on normal exit the connection returns to the pool, and on an exception the broken connection is closed and a replacement is spawned before the exception propagates.

TLS

python
import ssl

ctx = ssl.create_default_context()
ctx.load_verify_locations('ca.crt')

with VeltrixDBClient('veltrix.example.com', 9443, ssl_context=ctx) as c:
    c.ping()

# mTLS (client certificate)
ctx.load_cert_chain('client.crt', 'client.key')

Auth

python
with VeltrixDBClient('127.0.0.1', 9000) as c:
    c.auth('alice', 'secret')
    c.put('secure_key', b'secure_value')

The pool also supports auto-auth: pass username/password to VeltrixDBPool and every new connection authenticates itself before entering the pool.

Error handling

python
from veltrixdb import VeltrixDBError

val = c.get('missing-key')
if val is None:
    print('not found')

try:
    c.put('key', b'value')
except VeltrixDBError as e:
    print(f'server error: {e}')
except TimeoutError:
    print('operation timed out')

VeltrixDBError is raised for a STATUS_ERR response from the server or for a connection dropped mid-read. Missing keys are represented as None, never an exception — check for None rather than wrapping every get in a try/except.