RustyBase Python SDK

The official Python client for RustyBase, a high-performance, embedded-friendly database system with a MongoDB-like API.

Async SupportType HintedAuto-Auth

Installation

pip

1pip install rustybase

Quick Start

Connect & Query

Get up and running in seconds.

1from rustybase import RustyBaseClient
2
3# 1. Connect
4client = RustyBaseClient(connection_string="rustybase://admin:password@localhost:3000/mydb")
5
6# 2. Authenticate
7client.login()
8
9# 3. Get Collection
10users = client.db("mydb").collection("users")
11
12# 4. Insert Data
13users.insert_one({"name": "Alice", "role": "engineer", "age": 28})
14
15# 5. Find Data
16alice = users.find_one({"name": "Alice"})
17print(f"Found user: {alice}")

Client Configuration

You can initialize the client using a connection string or individual parameters.

Connection String (Recommended)

1client = RustyBaseClient(
2    connection_string="rustybase://username:password@host:port/database?authSource=admin"
3)

Individual Parameters

1client = RustyBaseClient(
2    host="127.0.0.1",
3    port=3000,
4    username="admin",
5    password="password",
6    database="default",
7    timeout=30.0,
8    use_signing=False  # Enable HMAC request signing for extra security
9)

Authentication

RustyBase uses JWT authentication. You must login before performing operations. The SDK handles automatic token refreshing for you.

1# Authenticate and obtain tokens
2client.login()
3
4# ... perform operations ...
5
6# Tokens are auto-refreshed on 401 errors.

CRUD Operations

Insert

1# Insert One
2doc = {"name": "Bob", "tags": ["developer", "python"]}
3result = users.insert_one(doc)
4# result: {"inserted_count": 1, "inserted_ids": [...]}
5
6# Insert Many
7docs = [
8    {"name": "Charlie", "age": 25},
9    {"name": "Dana", "age": 30}
10]
11users.insert_many(docs)

Find & Query

The find method is powerful and supports filtering, projection, sorting, and pagination.

1results = users.find(
2    filter={"age": {"$gt": 25}},   # Query criteria
3    projection={"name": 1},        # Fields to include (1) or exclude (0)
4    sort={"age": -1},              # 1 for Ascending, -1 for Descending
5    limit=10,                      # Max documents to return
6    skip=0,                        # Documents to skip
7    auto_indexing=True             # Allow ad-hoc indexing for performance
8)
9
10# Find One
11user = users.find_one({"_id": "some_id"})

Update

1users.update_one(
2    filter={"name": "Alice"},
3    update={"$set": {"active": True}, "$inc": {"age": 1}},
4    upsert=True  # Create if it doesn't exist
5)

Delete

1users.delete_one({"name": "Charlie"})

Aggregation Framework

Perform complex data analysis using aggregation pipelines.

1pipeline = [
2    # Stage 1: Filter documents
3    {"$match": {"status": "active"}},
4
5    # Stage 2: Group by a field and calculate metrics
6    {"$group": {
7        "_id": "$department",
8        "total_salary": {"$sum": "$salary"},
9        "avg_age": {"$avg": "$age"}
10    }},
11
12    # Stage 3: Sort the results
13    {"$sort": {"total_salary": -1}}
14]
15
16results = users.aggregate(pipeline)

Async Support

The SDK includes a full Asynchronous Client built on httpx.AsyncClient. It mirrors the synchronous API perfectly, but all network methods are awaitable.

1import asyncio
2from rustybase import AsyncRustyBaseClient
3
4async def main():
5    client = AsyncRustyBaseClient(connection_string="...")
6    await client.login()
7
8    db = client.db("test")
9    users = db.collection("users")
10
11    # Awaitable methods
12    await users.insert_one({"name": "Async User"})
13    docs = await users.find({"name": "Async User"})
14    print(docs)
15
16    await client.close()
17
18asyncio.run(main())

Admin Operations

1# Create a new database
2client.create_database("new_db")
3
4# List all databases
5dbs = client.list_databases()
6
7# Create a new user
8client.create_user("new_user", "secure_password", "new_db", roles=["readWrite"])
9
10# Wipe all data (Danger!)
11# client.wipe_all_data()

Error Handling

The SDK raises specific exceptions for different failure scenarios. All exceptions inherit from RustyBaseError.

1from rustybase import AuthenticationError, ConnectionError, RequestError
2
3try:
4    client.login()
5except AuthenticationError:
6    print("Check your username/password!")
7except ConnectionError:
8    print("Is the server running?")
9except RequestError as e:
10    print(f"API Error: {e}")

Licensed under MIT

Introduction

TS/JS SDK