Skip to main content

Overview

BinaryDB defines a hierarchy of exception classes for different error conditions. All exceptions inherit from DatabaseError, allowing you to catch all database-related errors with a single except clause.

Exception Hierarchy

DatabaseError
├── DatabaseIOError
├── DatabaseCorruptedError
├── KeyValidationError
├── RecordTypeError
├── TransactionError
└── ConcurrencyError

DatabaseError

Base class: Exception Base exception for all database-related errors. Use this to catch any error from BinaryDB operations.

When Raised

  • When the database is closed and operations are attempted
  • As a base class for all other database exceptions

Example

from binarydb.database import Database, DatabaseError

db = Database("data")

try:
    db.set("key", "value")
    db.commit()
except DatabaseError as e:
    print(f"Database error: {e}")
Catching all database errors:
from binarydb.database import Database, DatabaseError

db = Database("app")
db.load()

try:
    # Any database operation
    db.set("user:1", {"name": "Alice"})
    db.update("user:1", {"email": "[email protected]"})
    db.commit()
except DatabaseError as e:
    # Catches all BinaryDB exceptions
    print(f"Operation failed: {e}")

DatabaseIOError

Base class: DatabaseError Raised when disk I/O operations fail during commit or load operations.

When Raised

  • When commit() fails to write to disk (permission denied, disk full, etc.)
  • When the temporary file cannot be created during atomic writes
  • When file replacement fails

Example

from binarydb.database import Database, DatabaseIOError

db = Database("/readonly/path/data")
db.set("key", "value")

try:
    db.commit()
except DatabaseIOError as e:
    print(f"Failed to write to disk: {e}")
    # Handle I/O error: retry, use different path, etc.
Handling I/O errors with retry:
import time
from binarydb.database import Database, DatabaseIOError

db = Database("data")
db.set("important", "value")

max_retries = 3
for attempt in range(max_retries):
    try:
        db.commit()
        print("Saved successfully")
        break
    except DatabaseIOError as e:
        if attempt < max_retries - 1:
            print(f"Retry {attempt + 1}/{max_retries}...")
            time.sleep(1)
        else:
            print(f"Failed after {max_retries} attempts: {e}")
            raise

DatabaseCorruptedError

Base class: DatabaseError Raised when the database file format is invalid or corrupted.

When Raised

  • When load() fails to unpickle the database file
  • When the loaded data is not a dictionary
  • When the file is corrupted or incompatible

Example

from binarydb.database import Database, DatabaseCorruptedError

db = Database("data")

try:
    db.load()
except DatabaseCorruptedError as e:
    print(f"Database file is corrupted: {e}")
    # Restore from backup or reinitialize
    print("Starting with empty database")
    # db is already empty after failed load
Handling corrupted database with backup:
import shutil
from pathlib import Path
from binarydb.database import Database, DatabaseCorruptedError

db_path = "data"
backup_path = "data.backup"

db = Database(db_path)

try:
    db.load()
    print("Database loaded successfully")
except DatabaseCorruptedError as e:
    print(f"Corruption detected: {e}")
    
    # Try to restore from backup
    if Path(f"{backup_path}.pkl").exists():
        print("Restoring from backup...")
        shutil.copy(f"{backup_path}.pkl", f"{db_path}.pkl")
        db.load()
        print("Restored from backup")
    else:
        print("No backup available, starting fresh")

KeyValidationError

Base class: DatabaseError Raised when a database key is invalid.

When Raised

  • When a key is not a string
  • When a key is an empty string
  • Raised by: set(), get(), delete(), update(), exists()

Example

from binarydb.database import Database, KeyValidationError

db = Database("data")

try:
    # Invalid: key is not a string
    db.set(123, "value")
except KeyValidationError as e:
    print(f"Invalid key: {e}")

try:
    # Invalid: empty key
    db.set("", "value")
except KeyValidationError as e:
    print(f"Invalid key: {e}")
Validating keys before use:
from binarydb.database import Database, KeyValidationError

def safe_set(db, key, value):
    """Safely set a value with key validation."""
    try:
        db.set(key, value)
        return True
    except KeyValidationError as e:
        print(f"Invalid key '{key}': {e}")
        return False

db = Database("data")
if safe_set(db, "user:1", {"name": "Alice"}):
    print("Value stored")

RecordTypeError

Base class: DatabaseError Raised when a stored record has an unexpected type.

When Raised

  • When update() is called on a non-dictionary record
  • Only applies to the update() method, which requires dict records

Example

from binarydb.database import Database, RecordTypeError

db = Database("data")

# Store a non-dict value
db.set("counter", 42)

try:
    # Cannot update non-dict
    db.update("counter", {"value": 100})
except RecordTypeError as e:
    print(f"Type error: {e}")
    # Use set() instead for non-dict values
    db.set("counter", 100)
Type-safe update helper:
from binarydb.database import Database, RecordTypeError

def safe_update(db, key, changes):
    """Update a record if it's a dict, otherwise set it."""
    try:
        db.update(key, changes)
    except RecordTypeError:
        # Convert to dict if needed
        db.set(key, changes)

db = Database("data")
db.set("config", "simple_value")

# This will convert the value to a dict
safe_update(db, "config", {"debug": True})
print(db.get("config"))  # {"debug": True}

TransactionError

Base class: DatabaseError Raised on invalid transaction operations.

When Raised

  • When begin() is called while a transaction is already active
  • When end() is called without an active transaction
  • When rollback() is called without an active transaction

Example

from binarydb.database import Database, TransactionError

db = Database("data")

try:
    # Error: no transaction to end
    db.end()
except TransactionError as e:
    print(f"Transaction error: {e}")

db.begin()
try:
    # Error: transaction already active
    db.begin()
except TransactionError as e:
    print(f"Transaction error: {e}")
finally:
    db.rollback()
Safe transaction wrapper:
from binarydb.database import Database, TransactionError
from contextlib import contextmanager

@contextmanager
def transaction(db):
    """Context manager for safe transactions."""
    db.begin()
    try:
        yield db
        db.end()
    except Exception:
        db.rollback()
        raise

db = Database("accounts")
db.set("balance", 1000)
db.commit()

try:
    with transaction(db):
        balance = db.get("balance")
        db.set("balance", balance - 100)
        db.set("last_transaction", "withdrawal")
        # Auto-commits on success
except Exception as e:
    print(f"Transaction failed: {e}")
    # Auto-rollback on error

ConcurrencyError

Base class: DatabaseError Raised when concurrent access rules are violated.

When Raised

Currently defined but not actively used in the implementation. Reserved for future concurrency control features such as:
  • File locking conflicts
  • Multiple process access
  • Optimistic locking failures

Example

from binarydb.database import Database, ConcurrencyError

# Future use case:
try:
    db = Database("data")
    db.load()
    # ... operations ...
    db.commit()
except ConcurrencyError as e:
    print(f"Concurrency conflict: {e}")
    # Reload and retry
    db.load()

Error Handling Best Practices

Catch Specific Exceptions

Catch specific exceptions when you need different handling for different errors:
from binarydb.database import Database
from binarydb.errors import (
    DatabaseIOError,
    DatabaseCorruptedError,
    KeyValidationError,
    TransactionError
)

db = Database("data")

try:
    db.load()
    db.begin()
    db.set("key", "value")
    db.end()
    
except DatabaseCorruptedError as e:
    print(f"Corruption: {e}. Reinitializing...")
    # Start fresh
    
except DatabaseIOError as e:
    print(f"I/O error: {e}. Check permissions.")
    
except KeyValidationError as e:
    print(f"Invalid key: {e}")
    
except TransactionError as e:
    print(f"Transaction error: {e}")
    db.rollback()

Catch All Database Errors

For general error handling, catch DatabaseError:
from binarydb.database import Database, DatabaseError

db = Database("data")

try:
    db.load()
    db.set("key", "value")
    db.commit()
except DatabaseError as e:
    print(f"Database operation failed: {e}")
    # Log error, notify user, etc.

Clean Up Resources

Always close the database, even if errors occur:
from binarydb.database import Database, DatabaseError

db = Database("data")
try:
    db.load()
    # ... operations ...
    db.commit()
except DatabaseError as e:
    print(f"Error: {e}")
finally:
    db.close()

Validate Input

Prevent errors by validating input before operations:
from binarydb.database import Database

def add_user(db, user_id, user_data):
    """Add user with validation."""
    # Validate inputs
    if not isinstance(user_id, str) or not user_id:
        raise ValueError("user_id must be a non-empty string")
    
    if not isinstance(user_data, dict):
        raise ValueError("user_data must be a dictionary")
    
    # Safe to proceed
    db.set(f"user:{user_id}", user_data)
    db.commit()

db = Database("users")
db.load()

try:
    add_user(db, "123", {"name": "Alice"})
except ValueError as e:
    print(f"Validation error: {e}")

Build docs developers (and LLMs) love