CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-unqlite

Fast Python bindings for the UnQLite embedded NoSQL database.

Overview
Eval results
Files

transactions.mddocs/

Transactions

ACID transaction support for file-based databases with manual and context manager interfaces. Transactions provide atomicity, consistency, isolation, and durability for database operations.

Capabilities

Manual Transaction Management

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}")

Transaction Context Manager

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 back

Autocommit Control

Control 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}")

Decorator-Based Transactions

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}")

Transaction Behavior

File-Based Databases

File-based databases support full ACID transactions:

  • Atomicity: All operations in a transaction succeed or all fail
  • Consistency: Database remains in valid state after transaction
  • Isolation: Concurrent transactions don't interfere
  • Durability: Committed changes persist after system failure
# 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 together

In-Memory Databases

In-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 transacted

Error Handling

Transaction 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}")

Best Practices

  1. Use context managers for automatic cleanup and error handling
  2. Keep transactions short to minimize lock contention
  3. Batch related operations in single transactions for consistency
  4. Handle exceptions appropriately with rollback logic
  5. Consider disabling autocommit for bulk operations
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 manager

Install with Tessl CLI

npx tessl i tessl/pypi-unqlite

docs

cursors.md

database-management.md

index.md

json-collections.md

jx9-scripting.md

key-value.md

transactions.md

utilities.md

tile.json