Connect

from oxidb import OxiDbClient

# Connect to localhost:4444 (default)
client = OxiDbClient()

# Custom host/port/timeout
client = OxiDbClient("192.168.1.10", 4444, timeout=10.0)

# As a context manager (auto-closes)
with OxiDbClient() as db:
    db.ping()  # "pong"

Insert Documents

# Insert a single document
result = client.insert("users", {
    "name": "Alice",
    "email": "[email protected]",
    "age": 30,
})
# result: {"id": "..."}

# Insert many documents
result = client.insert_many("users", [
    {"name": "Bob", "age": 25},
    {"name": "Charlie", "age": 35},
    {"name": "Diana", "age": 28},
])

Find Documents

# Find all users over 25
users = client.find("users", {"age": {"$gt": 25}})

# Find with sort, skip, limit
users = client.find("users", {},
    sort={"age": -1},
    skip=0,
    limit=10,
)

# Find one document
user = client.find_one("users", {"name": "Alice"})

# Count documents
count = client.count("users", {"age": {"$gte": 18}})

Query Operators

# Comparison: $eq, $ne, $gt, $gte, $lt, $lte
query = {"age": {"$gte": 18, "$lt": 65}}

# $in — match any value in list
query = {"status": {"$in": ["active", "pending"]}}

# $exists — check field presence
query = {"email": {"$exists": True}}

# $regex — pattern matching
query = {"name": {"$regex": "^A"}}

# $and / $or — logical operators
query = {"$or": [
    {"age": {"$lt": 18}},
    {"age": {"$gte": 65}},
]}

# Nested fields with dot notation
query = {"address.city": "Berlin"}

Update Documents

# Update all matching documents
result = client.update("users",
    {"name": "Alice"},
    {"$set": {"age": 31}},
)

# Update one document
result = client.update_one("users",
    {"name": "Bob"},
    {
        "$inc": {"login_count": 1},
        "$set": {"last_login": "2026-03-05T12:00:00Z"},
    },
)

# Update operators: $set, $unset, $inc, $mul, $min, $max,
# $rename, $currentDate, $push, $pull, $addToSet, $pop

Delete Documents

# Delete all matching documents
result = client.delete("users", {"age": {"$lt": 18}})

# Delete one document
result = client.delete_one("users", {"name": "Charlie"})

Indexes

# Create a regular index
client.create_index("users", "age")

# Create a unique index
client.create_unique_index("users", "email")

# Create a composite index
client.create_composite_index("orders", ["customer_id", "date"])

# List indexes on a collection
indexes = client.list_indexes("users")

# Drop an index
client.drop_index("users", "idx_age")

Aggregation Pipeline

# Group users by city, count and average age
results = client.aggregate("users", [
    {"$match": {"age": {"$gte": 18}}},
    {"$group": {
        "_id": "$city",
        "count": {"$sum": 1},
        "avg_age": {"$avg": "$age"},
    }},
    {"$sort": {"count": -1}},
    {"$limit": 10},
])

# Pipeline stages: $match, $group, $sort, $project,
# $limit, $skip, $unwind, $addFields, $lookup, $count

Transactions

# Context manager (auto-commit / auto-rollback)
with client.transaction():
    client.insert("accounts", {"owner": "Alice", "balance": 1000})
    client.update_one("accounts",
        {"owner": "Bob"},
        {"$inc": {"balance": -500}},
    )
# auto-committed here

# Manual transaction control
client.begin_tx()
# ... operations ...
client.commit_tx()   # or client.rollback_tx()

SQL Queries

# Execute SQL queries
result = client.sql("SELECT * FROM users WHERE age > 25 ORDER BY name")

client.sql("INSERT INTO users (name, age) VALUES ('Eve', 22)")

client.sql("UPDATE users SET age = 31 WHERE name = 'Alice'")

client.sql("DELETE FROM users WHERE age < 18")

client.sql("CREATE INDEX idx_age ON users (age)")

# JOINs and GROUP BY
result = client.sql("""
    SELECT u.name, COUNT(o._id) as order_count
    FROM users u
    JOIN orders o ON u._id = o.user_id
    GROUP BY u.name
    ORDER BY order_count DESC
""")

Full-Text Search

# Create a text index on fields
client.create_text_index("articles", ["title", "body"])

# Search within a collection's text index
results = client.text_search("articles", "rust database", limit=20)

# Search across blobs (PDF, DOCX, etc.)
results = client.search("quarterly report", limit=10)

Vector Search

# Create a vector index
client.create_vector_index("products", "embedding", dimension=384, metric="cosine")

# Insert documents with vector embeddings
client.insert("products", {
    "name": "Wireless Mouse",
    "embedding": query_vector,  # list of 384 floats
})

# Find 5 nearest neighbors
results = client.vector_search("products", "embedding", query_vector, limit=5)
# Each result has _similarity and _distance fields

Blob Storage

# Create a bucket
client.create_bucket("documents")

# Upload a file
with open("report.pdf", "rb") as f:
    client.put_object("documents", "report.pdf", f.read(),
        content_type="application/pdf",
        metadata={"author": "Alice"})

# Download a file
data, metadata = client.get_object("documents", "report.pdf")

# Extract text from PDF/DOCX/HTML
text = client.extract_text("documents", "report.pdf")

# List objects with prefix filter
objects = client.list_objects("documents", prefix="reports/")

# Delete
client.delete_object("documents", "report.pdf")

Scheduled Tasks

# Create a schedule
client.create_schedule("cleanup", "delete_old_logs",
    cron="0 3 * * *",  # every day at 3 AM
    params={"days": 30},
)

# List, enable, disable, delete
schedules = client.list_schedules()
client.disable_schedule("cleanup")
client.enable_schedule("cleanup")
client.delete_schedule("cleanup")

Collections & Databases

# Collection management
client.create_collection("users")
collections = client.list_collections()
client.drop_collection("temp")

# Multi-database support
client.create_database("analytics")
client.use_database("analytics")
databases = client.list_databases()
client.drop_database("analytics")

# Compaction (reclaim disk space)
stats = client.compact("users")
# stats: {"old_size": ..., "new_size": ..., "docs_kept": ...}

Authentication & User Roles

# Per-database role grants (requires Admin)
client.grant_db_role("alice", "analytics", "Admin")
client.revoke_db_role("alice", "analytics")

Error Handling

from oxidb import OxiDbClient, OxiDbError, TransactionConflictError

try:
    client.insert("users", {"_id": "duplicate"})
except OxiDbError as e:
    print(f"Error: {e}")

# Transaction conflict (OCC)
try:
    with client.transaction():
        client.update("accounts", {}, {"$inc": {"balance": 1}})
except TransactionConflictError:
    print("Conflict — retry the transaction")

Installation

# TCP client — connects to oxidb-server
pip install oxidb

# Embedded mode — runs in-process, no server needed
pip install oxidb-embedded

# Or just copy the single TCP client file
curl -O https://oxidb.baltavista.com/oxidb.py

Embedded Mode (No Server)

from oxidb_embedded import OxiDbEmbedded

# Run the database engine directly in your Python process
# No server required — zero network overhead
with OxiDbEmbedded("./mydata") as db:
    db.insert("users", {"name": "Alice", "age": 30})
    users = db.find("users", {"age": {"$gt": 25}})
    print(users)

# Encryption at rest
db = OxiDbEmbedded("./mydata", encryption_key_path="./key.bin")

# Same API as TCP client — insert, find, update, delete,
# indexes, aggregation, transactions, search, blobs
db.create_index("users", "email")
db.aggregate("users", [
    {"$group": {"_id": "$city", "count": {"$sum": 1}}},
])

with db.transaction():
    db.insert("accounts", {"owner": "Bob", "balance": 500})
    db.update_one("accounts", {"owner": "Alice"}, {"$inc": {"balance": -500}})

db.close()

# Requires the native library (liboxidb_embedded_ffi.dylib/.so/.dll)
# Set OXIDB_LIB_PATH or place it next to the installed package

Full e-commerce analytics example on GitHub →