ekoDB Python Client

High-performance Python client for ekoDB, built with Rust for speed and safety.

This package wraps the ekodb_client Rust library using PyO3 to provide a native Python interface.

Features

• ✅ Fast: Built with Rust, leveraging the same client library as the Rust SDK
• ✅ Type-safe: Strong typing with Python type hints
• ✅ Async/await: Full async support using Python's asyncio
• ✅ Easy to use: Pythonic API that feels natural
• ✅ Complete: All ekoDB features supported
• ✅ Query Builder - Fluent API for complex queries with operators, sorting, and pagination
• ✅ Search - Full-text search, fuzzy search, and field-specific search with scoring
• ✅ Schema Management - Define and enforce data schemas with validation
• ✅ Join Operations - Single and multi-collection joins with queries
• ✅ Rate limiting with automatic retry (429, 503, network errors)
• ✅ Rate limit tracking (X-RateLimit-* headers)
• ✅ Configurable retry behavior
• ✅ Retry-After header support

Installation

pip install ekodb-client

Or install from source:

cd ekodb-py
pip install maturin
maturin develop

Python 3.10 or later is required. The native extension shares the Rust client's TLS dependencies; updating Python packages alone does not update TLS code embedded in an existing wheel. Rebuild source installations after refreshing ekodb-client-py/Cargo.lock, or install a wheel built with the patched lockfile.

For this repository's Python examples, install examples/requirements.txt in the project virtual environment. Its security minimums cover aiohttp 3.14.4, multidict 6.9.1, python-dotenv 1.2.2, requests 2.33.0, certifi 2026.10.19, and urllib3 2.8.0. These are example dependencies; the native SDK does not declare them as Python runtime dependencies.

Quick Start

import asyncio
from ekodb_client import Client, RateLimitError

async def main():
    # Create client with configuration
    client = Client.new(
        "http://localhost:8080",
        "your-api-key",
        should_retry=True,  # Enable automatic retries (default: True)
        max_retries=3,      # Maximum retry attempts (default: 3)
        timeout_secs=30     # Request timeout in seconds (default: 30)
    )

    try:
        # Insert a document
        record = await client.insert("users", {
            "name": "John Doe",
            "age": 30,
            "email": "john@example.com",
            "active": True
        })
        print(f"Inserted: {record['id']}")

        # Find by ID
        user = await client.find_by_id("users", record["id"])
        print(f"Found: {user}")

        # Find with query
        results = await client.find("users", limit=10)
        print(f"Found {len(results)} users")

        # Update
        updated = await client.update("users", record["id"], {
            "age": 31
        })
        print(f"Updated: {updated}")

        # Delete
        await client.delete("users", record["id"])
        print("Deleted")

    except RateLimitError as e:
        print(f"Rate limited! Retry after {e.retry_after_secs} seconds")

asyncio.run(main())

Usage Examples

Query Builder

from ekodb_client import Client, QueryBuilder

async def main():
    client = Client.new("http://localhost:8080", "your-api-key")

    # Simple query with operators
    query = QueryBuilder() \
        .eq("status", "active") \
        .gte("age", 18) \
        .lt("age", 65) \
        .limit(10) \
        .build()

    results = await client.find("users", query)

    # Complex query with sorting and pagination
    query = QueryBuilder() \
        .in_array("status", ["active", "pending"]) \
        .contains("email", "@example.com") \
        .sort_desc("created_at") \
        .skip(20) \
        .limit(10) \
        .build()

    results = await client.find("users", query)

Search Operations

# Basic text search
search_query = {
    "query": "programming",
    "min_score": 0.1,
    "limit": 10
}

results = await client.search("articles", search_query)
for result in results["results"]:
    print(f"Score: {result['score']:.4f} - {result['record']['title']}")

# Search with field weights
search_query = {
    "query": "rust database",
    "fields": ["title", "description"],
    "weights": {"title": 2.0},
    "limit": 5
}

results = await client.search("articles", search_query)

For directional matching, store two indexed Vector fields per record: query_embedding for a need and document_embedding for a complementary offer. Search the same source vector with vector_field="document_embedding" to find offers, or vector_field="query_embedding" to find similar needs. Both fields need compatible dimensions and a shared dual-encoder model space. The orthogonal vectors in client_search.py only demonstrate field selection; its live searches assert different top records.

Schema Management

# Create a collection with schema
schema = {
    "fields": {
        "name": {
            "field_type": "String",
            "required": True,
            "regex": "^[a-zA-Z ]+$"
        },
        "email": {
            "field_type": "String",
            "required": True,
            "unique": True
        },
        "age": {
            "field_type": "Integer",
            "min": 0,
            "max": 150
        }
    }
}

await client.create_collection("users", schema)

# Get collection schema
schema = await client.get_schema("users")

Join Operations

# Single collection join
query = {
    "join": {
        "collections": ["departments"],
        "local_field": "department_id",
        "foreign_field": "id",
        "as_field": "department"
    },
    "limit": 10
}

results = await client.find("users", query)

# Multi-collection join
query = {
    "join": [
        {
            "collections": ["departments"],
            "local_field": "department_id",
            "foreign_field": "id",
            "as_field": "department"
        },
        {
            "collections": ["profiles"],
            "local_field": "id",
            "foreign_field": "id",
            "as_field": "profile"
        }
    ],
    "limit": 10
}

results = await client.find("users", query)

API Reference

Client

Client.new(base_url: str, api_key: str, should_retry: bool = True, max_retries: int = 3, timeout_secs: int = 30) -> Client

Create a new ekoDB client.

Parameters:

• base_url: The base URL of the ekoDB server
• api_key: Your API key
• should_retry: Enable automatic retries (default: True)
• max_retries: Maximum number of retry attempts (default: 3)
• timeout_secs: Request timeout in seconds (default: 30)

Returns:

• A new Client instance

RateLimitInfo

Rate limit information is automatically tracked and logged by the client. The client will automatically retry on rate limit errors using the server's Retry-After header.

Properties

• limit: int - Maximum requests allowed per window
• remaining: int - Requests remaining in current window
• reset: int - Unix timestamp when the rate limit resets

Methods

• is_near_limit() -> bool - Check if approaching rate limit (<10% remaining)
• is_exceeded() -> bool - Check if the rate limit has been exceeded
• remaining_percentage() -> float - Get the percentage of requests remaining

RateLimitError

Exception raised when rate limit is exceeded (if retries are disabled or exhausted).

Properties

• retry_after_secs: int - Number of seconds to wait before retrying

await client.insert(collection: str, record: dict) -> dict

Insert a document into a collection.

Parameters:

• collection: The collection name
• record: A dictionary representing the document

Returns:

• The inserted document with ID

await client.find_by_id(collection: str, id: str) -> dict

Find a document by ID.

Parameters:

• collection: The collection name
• id: The document ID

Returns:

• The found document

await client.find_by_id_with_projection(collection: str, id: str, select_fields: Optional[list] = None, exclude_fields: Optional[list] = None) -> dict

Find a document by ID, returning only the projected fields.

Parameters:

• collection: The collection name
• id: The document ID
• select_fields: Optional list of fields to include
• exclude_fields: Optional list of fields to exclude

Returns:

• The found document with field projection applied

await client.find(collection: str, limit: Optional[int] = None) -> List[dict]

Find documents in a collection.

Parameters:

• collection: The collection name
• limit: Optional limit on number of results

Returns:

• List of matching documents

await client.update(collection: str, id: str, updates: dict) -> dict

Update a document.

Parameters:

• collection: The collection name
• id: The document ID
• updates: Dictionary of fields to update
• bypass_cache: Optional cache bypass flag
• select_fields: Optional returned fields to include (plus the primary key)
• exclude_fields: Optional returned fields to omit

Returns:

• The updated document

await client.delete(collection: str, id: str) -> None

Delete a document.

Parameters:

• collection: The collection name
• id: The document ID

await client.list_collections() -> List[str]

List all collections.

Returns:

• List of collection names

await client.delete_collection(collection: str) -> None

Delete a collection.

Parameters:

• collection: The collection name to delete

await client.search(collection: str, query: dict) -> dict

Perform full-text search on a collection.

Parameters:

• collection: The collection name
• query: Search query dictionary with fields like query, fields, weights, min_score, limit

Returns:

• Search results with scores and matched records

await client.create_collection(collection: str, schema: dict) -> None

Create a collection with a schema.

Parameters:

• collection: The collection name
• schema: Schema definition dictionary

await client.get_schema(collection: str) -> dict

Get the schema for a collection.

Parameters:

• collection: The collection name

Returns:

• Schema definition dictionary

await client.get_collection(collection: str) -> dict

Get collection metadata including schema.

Parameters:

• collection: The collection name

Returns:

• Collection metadata dictionary

await client.collection_exists(collection: str) -> bool

Check if a collection exists.

Parameters:

• collection: The collection name

Returns:

• True if the collection exists, False otherwise

await client.count_documents(collection: str) -> int

Count documents in a collection.

Parameters:

• collection: The collection name

Returns:

• Number of documents in the collection

Chat Models

await client.get_chat_models() -> dict

Get all available chat models organized by provider.

Returns:

• Dictionary mapping provider names (openai, anthropic, perplexity, gemini) to lists of model names, plus a providers key: a dictionary of per-provider status (status one of ok, not_configured, auth_failed, permission_denied, billing, rate_limited, unavailable, unreachable, request_error; verified; http_status; message; model_count) so a rejected key is distinguishable from a missing one

await client.get_chat_model(provider: str) -> list

Get models for a specific provider.

Parameters:

• provider: The provider name (e.g., "openai", "anthropic")

Returns:

• List of model names for the provider

User Functions

await client.save_user_function(user_function: dict) -> str

Create a new user function.

Parameters:

• user_function: Dictionary containing label, name, parameters, functions, etc.

Returns:

• The ID of the created user function

await client.get_user_function(label: str) -> dict

Get a user function by its label.

Parameters:

• label: The user function label

Returns:

• User function definition dictionary

await client.list_user_functions(tags: Optional[list] = None) -> list

List all user functions, optionally filtered by tags.

Parameters:

• tags: Optional list of tags to filter by

Returns:

• List of user function dictionaries

await client.update_user_function(label: str, user_function: dict) -> None

Update an existing user function.

Parameters:

• label: The user function label
• user_function: Updated user function definition

await client.delete_user_function(label: str) -> None

Delete a user function by its label.

Parameters:

• label: The user function label

Key-Value

await client.kv_set(key: str, value: dict, ttl: Optional[str] = None) -> None

Set a key-value pair, optionally with an expiration.

Parameters:

• key: The key
• value: A dictionary value to store
• ttl: Optional expiration (e.g. "60s")

await client.kv_get(key: str) -> Optional[dict]

Get a value by key.

Parameters:

• key: The key

Returns:

• The stored value, or None if the key does not exist

await client.kv_delete(key: str) -> None

Delete a key.

Parameters:

• key: The key

await client.kv_clear() -> None

Clear the entire KV store (all keys in the namespace).

Collections

await client.list_user_collections() -> List[str]

List collections, excluding internal chat/system collections.

Returns:

• List of user collection names

Transactions

Buffered, read-your-writes transactions. Operations issued with a transaction_id kwarg (accepted on insert, find, find_by_id, update, delete) are staged and applied atomically at commit.

await client.begin_transaction(isolation_level: str | None = None) -> str

Start a transaction and return its id. Omitting isolation_level uses the server default.

await client.commit_transaction(transaction_id: str) -> None

Apply the staged writes. May raise a retryable conflict (HTTP 409).

await client.rollback_transaction(transaction_id: str) -> None

Discard the staged writes.

await client.create_savepoint(transaction_id: str, name: str) -> None

Create a savepoint within the transaction.

await client.rollback_to_savepoint(transaction_id: str, name: str) -> None

Roll the transaction back to a savepoint.

await client.release_savepoint(transaction_id: str, name: str) -> None

Release (forget) a savepoint.

Chat-session completion budgets

create_chat_session and branch_chat_session accept an optional max_tokens argument. Branching appends it after the existing llm_model argument, preserving older positional calls. Omission or None leaves the wire field absent.

branched = await client.branch_chat_session(
    "parent-chat", 0, [("products", ["name"])], "openai",
    llm_model="gpt-6-luna", max_tokens=1024,
)

This requests a completion limit through the server; it is not a total billing limit. See AI example costs.

Examples

See the examples directory for complete working examples:

• client_simple_crud.py - Basic CRUD operations
• client_query_builder.py - Complex queries with QueryBuilder
• client_search.py - Full-text search operations
• client_schema.py - Schema management
• client_joins.py - Join operations
• client_batch_operations.py - Batch operations
• client_kv_operations.py - Key-value operations
• client_chat_models.py - Chat models API
• client_user_functions.py - User functions API
• And more...

Development

Building

# Install maturin
pip install maturin

# Build and install in development mode
maturin develop

# Build release wheel
maturin build --release

Testing

# Run Python tests
pytest

# Run with coverage
pytest --cov=ekodb

Goals, Tasks, and Agents

import asyncio
from ekodb_client import Client

async def main():
    client = Client.new("http://localhost:8080", "your-api-key")

    # Goals
    goal = await client.goal_create({"title": "Migrate data", "status": "pending"})
    goals = await client.goal_list()
    await client.goal_complete("goal-id", {"summary": "Done"})

    # Tasks
    task = await client.task_create({"title": "Backup", "schedule": "0 0 * * *"})
    await client.task_start("task-id")

    # Agents
    agent = await client.agent_create({"name": "processor", "llm_model": "gpt-4.1"})

asyncio.run(main())

Schedules

# Create a schedule
sched = await client.create_schedule({
    "name": "nightly",
    "cron_expression": "0 0 2 * * *",
    "function_label": "nightly_backup",
})

# Pause a schedule
await client.pause_schedule("sched-id")

# Run immediately
await client.trigger_schedule("sched-id")

WebSocket Operations

ws = await client.websocket("ws://localhost:8080")

# Full CRUD over WebSocket (14 methods)
result = await ws.ws_insert("users", {"name": "Alice", "email": "a@b.com"})
results = await ws.ws_query("users", filter={
    "type": "Condition",
    "content": {"field": "status", "operator": "Eq", "value": "active"},
})
user = await ws.ws_find_by_id("users", "record-id")
await ws.ws_update("users", "record-id", {"name": "Updated"})
await ws.ws_delete("users", "record-id")

# Batch operations
await ws.ws_batch_insert("logs", [{"msg": "a"}, {"msg": "b"}])
await ws.ws_batch_update("logs", [("id1", {"msg": "x"}), ("id2", {"msg": "y"})])
await ws.ws_batch_delete("logs", ["id1", "id2"])

# Search + collection management
hits = await ws.ws_text_search("docs", "python async", limit=10)
collections = await ws.ws_list_collections()
await ws.ws_create_collection("new_coll")

# Atomic field actions
await ws.ws_update_with_action("counters", "views", "increment", "count", 1)

# Subscriptions + chat
await ws.ws_unsubscribe("users")
await ws.cancel_chat("chat-id")

WebSocket Chat Streaming

stream = await ws.chat_send(chat_id, "What is the capital of France?")
while True:
    event = await stream.recv()
    if event is None:
        break

    if event["type"] == "chunk":
        print(event["content"], end="")
    elif event["type"] == "end":
        print(f"\nDone (context: {event.get('context_window')} tokens)")
        break
    elif event["type"] == "tool_call":
        print(f"[Tool] {event['tool_name']}")
        await ws.send_tool_result(
            chat_id, event["call_id"], True, {"result": "done"}
        )
    elif event["type"] == "error":
        print(f"Error: {event['error']}")
        break

License

MIT

Links

• GitHub
• Examples
• PyPI Package

API-key administration

The separate admin client exposes API-key creation, listing, revocation, current-key permissions, and collection/KV grant reads and edits. See the shared admin contract for exact methods and compatibility. Version 0.80.0 moves permission routes under /api/auth/keys/{id}/permissions. A cached route probe supports both generations, including new-route prereleases and pre-cap builds that report 0.79.2. New methods target id (or me) and return {before, after} changes. PATCH omission preserves a limit, null clears it, and arrays replace it, including an empty array. get_permissions/getPermissions retain the existing admin/collection fields and add KV grants. Explicit legacy collection edits retain the old secret-targeted shape only for detected legacy servers >=0.79.2. Registration and listing expose optional ids for compatibility with older replies. Prefer ids for revocation. Never log key records.

from ekodb_client.admin import AdminClient
admin = AdminClient(base_url, api_key)
permissions = await admin.get_permissions()
# update_api_key_collection_permissions accepts a dict: omit a field to keep,
# pass None to clear, or a list (including []) to replace.

This entry point uses the compiled Rust/PyO3 admin binding and returns Python records; it is available through ekodb_client.admin.

See the matching client_admin scenario in examples. It is read-only by default; EKODB_ADMIN_LIVE=1 enables a uniquely labeled disposable key lifecycle with guarded revocation and verified cleanup.