Riak

import riak

# Connect to Riak cluster
client = riak.RiakClient(host='127.0.0.1', pb_port=8087)

# Create a bucket with custom properties
bucket = client.bucket('users')
bucket.set_properties({
    'n_val': 3,  # 3 replicas
    'r': 2,      # Read from 2 nodes
    'w': 2,      # Write to 2 nodes
    'pr': 1,     # Primary read quorum
    'pw': 1      # Primary write quorum
})

# Store an object
user_data = {
    'name': 'John Doe',
    'email': 'john@example.com',
    'created': '2024-01-15T10:00:00Z'
}

user_obj = bucket.new('user123', data=user_data)
user_obj.store()

# Retrieve an object
retrieved_user = bucket.get('user123')
print(retrieved_user.data['name'])  # 'John Doe'

# Update with conflict resolution
retrieved_user.data['last_login'] = '2024-01-16T14:30:00Z'
retrieved_user.store()

# Search using Secondary Indexes (2i)
bucket.new('user124', data={
    'name': 'Jane Smith',
    'email': 'jane@example.com',
    'department': 'engineering'
}).add_index('department_bin', 'engineering').store()

# Query by index
engineering_users = bucket.get_index('department_bin', 'engineering')
for key in engineering_users:
    user = bucket.get(key)
    print(f"User: {user.data['name']}")

# MapReduce example
mr = client.add('users')
mr.map('function(v) { var data = JSON.parse(v.values[0].data); return [data.department]; }')
mr.reduce('function(values) { return values.sort(); }')
departments = mr.run()
print(departments)