Fast Python bindings for the UnQLite embedded NoSQL database.
ACID transaction support for file-based databases with manual and context manager interfaces. Transactions provide atomicity, consistency, isolation, and durability for database operations.
Explicit transaction control with begin, commit, and rollback operations.
def begin(self):
"""Begin a new transaction. Only works for file-based databases.
Returns True if successful, False for in-memory databases."""
...
def commit(self):
"""Commit current transaction. Only works for file-based databases.
Returns True if successful, False for in-memory databases."""
...
def rollback(self):
"""Rollback current transaction. Only works for file-based databases.
Returns True if successful, False for in-memory databases."""
...Usage Example:
db = unqlite.UnQLite('transactional.db')
# Manual transaction management
try:
db.begin()
# Multiple operations in transaction
db['user:1'] = '{"name": "Alice", "balance": 1000}'
db['user:2'] = '{"name": "Bob", "balance": 500}'
# Simulate transfer
alice_data = json.loads(db['user:1'])
bob_data = json.loads(db['user:2'])
alice_data['balance'] -= 100
bob_data['balance'] += 100
db['user:1'] = json.dumps(alice_data)
db['user:2'] = json.dumps(bob_data)
# Commit all changes
db.commit()
print("Transaction committed successfully")
except Exception as e:
# Rollback on error
db.rollback()
print(f"Transaction rolled back: {e}")Automatic transaction management using Python context managers.
def transaction(self):
"""Create context manager for wrapping a transaction.
Returns Transaction object."""
...
class Transaction:
def __enter__(self):
"""Begin transaction and return self."""
...
def __exit__(self, exc_type, exc_val, exc_tb):
"""Commit transaction or rollback on exception."""
...Usage Example:
db = unqlite.UnQLite('data.db')
# Automatic transaction management
try:
with db.transaction():
db['config:theme'] = 'dark'
db['config:language'] = 'en'
db['config:notifications'] = 'true'
# If any operation fails, entire transaction is rolled back
# If all operations succeed, transaction is automatically committed
except unqlite.UnQLiteError as e:
print(f"Transaction failed: {e}")
# Complex transaction with error handling
with db.transaction():
# Multiple related operations
db['order:123'] = '{"status": "processing", "total": 99.99}'
db['inventory:widget'] = '{"stock": 45}' # Decrease stock
db['customer:456'] = '{"orders": ["order:123"]}'
# All operations commit together or all roll backControl automatic transaction behavior for file-based databases.
def disable_autocommit(self):
"""Disable autocommit for file-based databases.
Returns True if successful, False for in-memory databases."""
...Usage Example:
db = unqlite.UnQLite('manual.db')
# Disable autocommit for better performance with many operations
db.disable_autocommit()
try:
db.begin()
# Many operations without intermediate commits
for i in range(1000):
db[f'key{i}'] = f'value{i}'
# Single commit for all operations
db.commit()
except Exception as e:
db.rollback()
print(f"Batch operation failed: {e}")Automatic transaction wrapping for functions.
def commit_on_success(self, fn):
"""Decorator that wraps function in transaction.
Commits on success, rolls back on exception."""
...Usage Example:
db = unqlite.UnQLite('decorated.db')
@db.commit_on_success
def transfer_funds(from_user, to_user, amount):
"""Transfer funds between users with automatic transaction."""
from_data = json.loads(db[f'user:{from_user}'])
to_data = json.loads(db[f'user:{to_user}'])
if from_data['balance'] < amount:
raise ValueError("Insufficient funds")
from_data['balance'] -= amount
to_data['balance'] += amount
db[f'user:{from_user}'] = json.dumps(from_data)
db[f'user:{to_user}'] = json.dumps(to_data)
return True
# Set up test data
db['user:alice'] = '{"balance": 1000}'
db['user:bob'] = '{"balance": 500}'
# Use decorated function - automatically transacted
try:
transfer_funds('alice', 'bob', 100)
print("Transfer successful")
except ValueError as e:
print(f"Transfer failed: {e}")File-based databases support full ACID transactions:
# File-based database with full transaction support
db = unqlite.UnQLite('persistent.db')
with db.transaction():
# All operations are atomic
db['critical_data'] = 'important_value'
db['backup_data'] = 'backup_value'
# Both operations commit togetherIn-memory databases have limited transaction support:
# In-memory database
db = unqlite.UnQLite(':mem:')
# Transaction methods return False for in-memory databases
result = db.begin() # Returns False
result = db.commit() # Returns False
result = db.rollback() # Returns False
# Context manager still works but provides no transactional guarantees
with db.transaction():
db['key'] = 'value' # Operations execute normally but aren't truly transactedTransaction operations can encounter various error conditions:
db = unqlite.UnQLite('transactional.db')
try:
with db.transaction():
db['key1'] = 'value1'
# Simulate error condition
if some_error_condition:
raise ValueError("Business logic error")
db['key2'] = 'value2'
except ValueError as e:
print(f"Business error, transaction rolled back: {e}")
except unqlite.UnQLiteError as e:
print(f"Database error: {e}")db = unqlite.UnQLite('optimized.db')
# Efficient bulk loading
db.disable_autocommit()
try:
with db.transaction():
# Process large dataset in single transaction
for record in large_dataset:
db[record['id']] = json.dumps(record)
# Single commit for entire batch
except Exception as e:
print(f"Bulk load failed: {e}")
# Automatic rollback via context managerInstall with Tessl CLI
npx tessl i tessl/pypi-unqlite