Examples

This section provides concise examples of how to use the pyrex library, a Python wrapper for RocksDB. The examples focus on demonstrating core functionalities directly.

Basic Put and Get

Open a database, store a key-value pair, and retrieve it.

basic_put_get.py

import pyrex
import os
import shutil

db_path = "/tmp/pyrex_example_basic"
if os.path.exists(db_path):
    shutil.rmtree(db_path)

db = pyrex.PyRocksDB(db_path)

# Put a key-value pair
db.put(b"my_key", b"my_value")

# Get the value
value = db.get(b"my_key")
print(f"Retrieved: {value.decode()}") # Expected: my_value

# Get a non-existent key
none_value = db.get(b"non_existent")
print(f"Retrieved non-existent: {none_value}") # Expected: None

del db # Close the DB
shutil.rmtree(db_path) # Clean up

Configuring Options

Customize database behavior using PyOptions.

config_options.py

import pyrex
import os
import shutil

db_path = "/tmp/pyrex_example_config"
if os.path.exists(db_path):
    shutil.rmtree(db_path)

options = pyrex.PyOptions()
options.create_if_missing = True
options.max_open_files = 1000
options.compression = pyrex.CompressionType.kLZ4Compression
options.optimize_for_small_db() # Apply an optimization preset

db = pyrex.PyRocksDB(db_path, options)

# Verify some configured options (values might be adjusted by RocksDB internally)
retrieved_options = db.get_options()
print(f"Configured max_open_files: {retrieved_options.max_open_files}") # Expected: 5000 (adjusted by optimize_for_small_db)
print(f"Configured compression: {retrieved_options.compression}") # Expected: CompressionType.kLZ4Compression

del db
shutil.rmtree(db_path)

Atomic Write Batch

Perform multiple put and delete operations in a single, atomic transaction.

write_batch.py

import pyrex
import os
import shutil

db_path = "/tmp/pyrex_example_batch"
if os.path.exists(db_path):
    shutil.rmtree(db_path)

db = pyrex.PyRocksDB(db_path)

# Prepare some initial data
db.put(b"key1", b"value1")
db.put(b"key2", b"value2")

# Create a write batch
batch = pyrex.PyWriteBatch()
batch.put(b"key3", b"value3")       # Add new key
batch.delete(b"key1")               # Delete existing key
batch.put(b"key2", b"updated_value2") # Update existing key

db.write(batch) # Apply the batch atomically

# Verify results
print(f"Key1: {db.get(b'key1')}")       # Expected: None
print(f"Key2: {db.get(b'key2').decode()}") # Expected: updated_value2
print(f"Key3: {db.get(b'key3').decode()}") # Expected: value3

del db
shutil.rmtree(db_path)

Iterating Data

Traverse key-value pairs using PyRocksDBIterator for forward and backward scans.

iterator_usage.py

import pyrex
import os
import shutil

db_path = "/tmp/pyrex_example_iterator"
if os.path.exists(db_path):
    shutil.rmtree(db_path)

db = pyrex.PyRocksDB(db_path)

# Insert sample data
db.put(b"alpha", b"A")
db.put(b"beta", b"B")
db.put(b"gamma", b"G")

it = db.new_iterator()

print("Forward iteration:")
it.seek_to_first()
while it.valid():
    print(f"  Key: {it.key().decode()}, Value: {it.value().decode()}")
    it.next()
it.check_status() # Check for errors

print("\nSeek and backward iteration:")
it.seek(b"gamma") # Seek to 'gamma'
while it.valid():
    print(f"  Key: {it.key().decode()}, Value: {it.value().decode()}")
    it.prev() # Move backward

del db
shutil.rmtree(db_path)

Error Handling

Catch RocksDB-specific exceptions for robust applications.

error_handling.py

import pyrex
import os
import shutil

db_path = "/tmp/pyrex_example_error"
if os.path.exists(db_path):
    shutil.rmtree(db_path)

# Create an options object to trigger an error
options = pyrex.PyOptions()
options.create_if_missing = True
options.error_if_exists = True

try:
    # First open creates the DB
    db1 = pyrex.PyRocksDB(db_path, options)
    db1.put(b"foo", b"bar")
    del db1 # Close it so we can try to open again

    # This second open should fail because error_if_exists is True
    print("Attempting to open existing DB with error_if_exists=True...")
    db2 = pyrex.PyRocksDB(db_path, options) # This line should raise an exception
    print("This line should not be reached.") # If it is, the error wasn't caught
except pyrex.RocksDBException as e:
    print(f"Caught expected RocksDBException: {e}")
    # Assert that the error message contains the expected string
    assert "exists (error_if_exists is true)" in str(e)
finally:
    # Clean up the database directory regardless of success or failure
    if os.path.exists(db_path):
        shutil.rmtree(db_path)
        print(f"Cleaned up database at {db_path}")