Python Binding

from rocksdict import Rdict

# Open or create a database
db = Rdict("/path/to/db")

# Dict-like operations
db["key1"] = "value1"
db["key2"] = "value2"

# Read values
print(db["key1"])  # Output: value1

# Check existence
if "key1" in db:
    print("Key exists")

# Delete keys
del db["key1"]

# Iterate over keys
for key in db.keys():
    print(f"{key}: {db[key]}")

# Close the database
db.close()

from rocksdict import Rdict, Options

# Create options
opts = Options()
opts.create_if_missing(True)
opts.set_max_open_files(10000)
opts.set_use_fsync(False)
opts.set_bytes_per_sync(8388608)
opts.optimize_for_point_lookup(1024)
opts.set_table_cache_num_shard_bits(6)
opts.set_max_write_buffer_number(32)
opts.set_write_buffer_size(536870912)
opts.set_target_file_size_base(1073741824)
opts.set_min_write_buffer_number_to_merge(4)
opts.set_level_zero_stop_writes_trigger(2000)
opts.set_level_zero_slowdown_writes_trigger(0)
opts.set_compaction_style("universal")

# Open database with options
db = Rdict("/path/to/db", options=opts)

db.close()

from rocksdict import Rdict

# Open with column families
db = Rdict("/path/to/db", column_families={"users", "posts", "comments"})

# Access specific column family
users_cf = db.get_column_family("users")
posts_cf = db.get_column_family("posts")

# Write to different column families
users_cf["user:1000"] = {"name": "Alice"}
posts_cf["post:5000"] = {"title": "Hello World"}

# Read from column families
user = users_cf["user:1000"]
post = posts_cf["post:5000"]

db.close()

from rocksdict import Rdict

db = Rdict("/path/to/db")

# Write some data
db["counter"] = 100

# Create a snapshot
snapshot = db.snapshot()

# Modify data
db["counter"] = 200

# Read from snapshot (still sees old value)
print(snapshot["counter"])  # 100

# Read from current db (sees new value)
print(db["counter"])  # 200

# Release snapshot
snapshot.close()
db.close()

from rocksdict import Rdict
import time

db = Rdict("/path/to/db")

# Slow: Individual writes
start = time.time()
for i in range(10000):
    db[f"key_{i}"] = f"value_{i}"
print(f"Individual writes: {time.time() - start:.2f}s")

# Fast: Batch writes
start = time.time()
with db.write_batch() as batch:
    for i in range(10000):
        batch[f"key_{i}"] = f"value_{i}"
print(f"Batch writes: {time.time() - start:.2f}s")

db.close()

from rocksdict import Rdict

# Automatic cleanup with context manager
with Rdict("/path/to/db") as db:
    db["key"] = "value"
    print(db["key"])
    # Database is automatically closed when exiting the context

from rocksdict import Rdict, RocksDBError

try:
    db = Rdict("/path/to/db")
    
    # Attempt operations
    value = db["nonexistent_key"]
    
except KeyError as e:
    print(f"Key not found: {e}")
except RocksDBError as e:
    print(f"RocksDB error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")
finally:
    if 'db' in locals():
        db.close()

from typing import Dict, Optional
from rocksdict import Rdict
from dataclasses import dataclass

@dataclass
class User:
    name: str
    age: int
    email: str

class UserStore:
    def __init__(self, db_path: str):
        self.db = Rdict(db_path)
    
    def save_user(self, user_id: str, user: User) -> None:
        self.db[f"user:{user_id}"] = {
            "name": user.name,
            "age": user.age,
            "email": user.email
        }
    
    def get_user(self, user_id: str) -> Optional[User]:
        data = self.db.get(f"user:{user_id}")
        if data:
            return User(**data)
        return None
    
    def close(self) -> None:
        self.db.close()

# Usage
store = UserStore("/path/to/db")
user = User(name="Alice", age=30, email="[email protected]")
store.save_user("1000", user)
retrieved = store.get_user("1000")
print(retrieved.name)  # Alice
store.close()