@livendb/liven-client (Python)

Python client SDK for LivenDB

from liven_py import LivenClient, Pipeline, Filter

client = LivenClient.connect("127.0.0.1:43121")
client.insert("users", "u1", {"name": "Alice", "role": "admin"})
user = client.get("users", "u1")
print(user[0].field("name"))
# → Alice

results = client.run(
    Pipeline.from_stream("users")
        .filter(Filter.field("role").eq("admin"))
        .limit(10)
)
client.close()

Installation

pip install liven-client

Requires Python 3.9+.

Quick Start

1. Connect

from liven_py import LivenClient

# Default port is 43121
client = LivenClient.connect("localhost")

# With authentication key
client = LivenClient.connect("localhost:43121?auth_key=your-api-key")

2. CRUD Operations

# Insert
client.insert("events", "e1", {"type": "click", "value": 42})

# Upsert (insert or replace)
client.upsert("events", "e1", {"type": "click", "value": 99})

# Update (merge)
client.update("users", "u1", {"last_login": 1234567890})

# Get by key
record = client.get("users", "u1")

# Delete
client.delete("events", "e1")

# Clear all records from a stream
client.clear("sessions")

# Drop an entire stream
client.drop_stream("old_data")

3. Batch Operations

client.insert_many("events", [
    ("e1", {"type": "click"}),
    ("e2", {"type": "view"}),
])

4. Pipeline Queries

from liven_py import Pipeline, Filter

results = client.run(
    Pipeline.from_stream("orders")
        .filter(Filter.field("amount").gte(100))
        .filter(Filter.field("status").eq("completed"))
        .sort("amount", descending=True)
        .limit(10)
)

5. Shorthand Pipeline Methods

client.filter("events", Filter.field("type").eq("click"))
client.limit("events", 50)
client.count("events")
client.sort("orders", "total", descending=True)
client.page("events", 1, 25)
client.page_cursor("feed", "cursor_abc", 25)
client.map("users", ["name", "email"])
client.window("pageviews", 60000, "count")
client.group("events", "type", ["count", "sum(value)"])
client.distinct("visitors", "ip_address")
client.vector_filter("documents", "embedding", [1, 0, -1], 0.75)

6. Joins

client.enrich("orders", "users", "user_id")
client.correlate("events", "sessions", "session_id", 5000)
client.chain("reviews", "orders", "order_id")
client.sequence("events", [
    Filter.field("type").eq("login"),
    Filter.field("type").eq("purchase"),
], 300000)

7. Pipeline Mutations

client.run_update(
    Pipeline.from_stream("events").filter(Filter.field("status").eq("pending")),
    {"status": "processed"}
)
client.run_delete(
    Pipeline.from_stream("events").filter(Filter.field("type").eq("temp"))
)

8. Live Subscriptions

client.run_listen(
    Pipeline.from_stream("alerts").filter(Filter.field("priority").eq("critical"))
)

9. Metadata

streams = client.streams()
status = client.status()

10. Raw DSL

results = client.query('from("users") | count()')

Development

# Install dependencies
pip install -e .

# Run the demo
python examples/demo.py
python examples/demo.py --key your-api-key

# Run tests
python -m pytest

License

MIT