Database

Overview

The Database class provides a persistent key-value store with in-memory operations and disk persistence using pickle serialization.

Security Notice: Never load database files from untrusted sources. Pickle serialization is unsafe with untrusted data.

Constructor

`Database(path)`

Creates a new database instance bound to a file path.

path

str | Path

required

Location of the database file. The .pkl extension is automatically added if not present.

from binarydb.database import Database

# Create a new database
db = Database("mydata")
# Creates mydata.pkl

# Using Path object
from pathlib import Path
db = Database(Path("/var/data/store"))

Core Methods

`set(key, value)`

Insert or overwrite a record in the database.

key

str

required

Record identifier. Must be a non-empty string.

value

Any

required

Pickle-serializable object to store.

return

None

Raises:

KeyValidationError: If key is not a string or is empty
DatabaseError: If database is closed

db = Database("users")

# Store simple values
db.set("username", "alice")
db.set("age", 30)

# Store complex objects
db.set("user:1", {
    "name": "Alice",
    "email": "[email protected]",
    "roles": ["admin", "user"]
})

# Overwrite existing value
db.set("username", "bob")

`get(key, default=None)`

Retrieve a record by key.

key

str

required

Record identifier to retrieve.

default

Any

default:"None"

Value to return if key does not exist.

return

Any

The stored value, or default if key not found.

Raises:

KeyValidationError: If key is not a string or is empty
DatabaseError: If database is closed

db = Database("config")
db.set("debug", True)

# Get existing key
value = db.get("debug")  # Returns: True

# Get non-existent key
value = db.get("missing")  # Returns: None

# Use custom default
value = db.get("timeout", 30)  # Returns: 30

`delete(key)`

Delete a record if it exists. Does nothing if key doesn’t exist.

key

str

required

Record identifier to delete.

return

None

Raises:

KeyValidationError: If key is not a string or is empty
DatabaseError: If database is closed

db = Database("cache")
db.set("temp", "value")

# Delete existing key
db.delete("temp")
print(db.exists("temp"))  # False

# Safe to delete non-existent key
db.delete("nonexistent")  # No error

`update(key, changes)`

Update fields of a dictionary record. The record must exist and be a dictionary.

key

str

required

Record identifier to update.

changes

dict[str, Any]

required

Dictionary of fields to update or add.

return

None

Raises:

KeyError: If key does not exist
RecordTypeError: If stored value is not a dict
KeyValidationError: If key is not a string or is empty
DatabaseError: If database is closed

db = Database("users")

# Create initial record
db.set("user:1", {"name": "Alice", "age": 25})

# Update existing fields and add new ones
db.update("user:1", {
    "age": 26,
    "city": "New York"
})

print(db.get("user:1"))
# Output: {"name": "Alice", "age": 26, "city": "New York"}

# Error: cannot update non-dict
db.set("count", 42)
db.update("count", {"value": 100})  # Raises RecordTypeError

`exists(key)`

Check if a key exists in the database.

key

str

required

Record identifier to check.

return

bool

True if key exists, False otherwise.

Raises:

KeyValidationError: If key is not a string or is empty
DatabaseError: If database is closed

db = Database("inventory")
db.set("item:1", {"name": "Widget"})

if db.exists("item:1"):
    print("Item found")

if not db.exists("item:999"):
    print("Item not found")

`all()`

Return a shallow copy of all records in the database.

return

dict[str, Any]

Dictionary containing all key-value pairs.

Raises:

DatabaseError: If database is closed

db = Database("settings")
db.set("theme", "dark")
db.set("lang", "en")
db.set("notifications", True)

# Get all records
all_data = db.all()
print(all_data)
# Output: {"theme": "dark", "lang": "en", "notifications": True}

# Modifying the copy doesn't affect the database
all_data["theme"] = "light"
print(db.get("theme"))  # Still "dark"

Persistence Methods

`commit()`

Persist the database to disk using atomic file replacement. Writes to a temporary file first, then atomically replaces the database file to prevent corruption if the write fails.

return

None

Raises:

DatabaseIOError: If disk write fails
DatabaseError: If database is closed

db = Database("products")
db.set("product:1", {"name": "Laptop", "price": 999})
db.set("product:2", {"name": "Mouse", "price": 25})

# Save to disk
db.commit()

# Changes are now persisted to products.pkl

`load()`

Load database contents from disk, replacing all current in-memory data. If the database file doesn’t exist, this method does nothing (useful for initialization).

return

None

Raises:

DatabaseCorruptedError: If file cannot be loaded or has invalid format
DatabaseError: If database is closed

db = Database("products")

# Load existing data from disk
db.load()

print(db.get("product:1"))
# Output: {"name": "Laptop", "price": 999}

# Safe to call on non-existent file
new_db = Database("newfile")
new_db.load()  # Does nothing, no error

Transaction Methods

Transactions allow you to group multiple operations together and rollback if needed.

`begin()`

Begin a new transaction. All changes are kept in memory until end() or rollback(). During a transaction, changes are not automatically persisted and the dirty flag is not set.

return

None

Raises:

TransactionError: If a transaction is already active
DatabaseError: If database is closed

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

# Start transaction
db.begin()
db.set("balance", 500)
db.set("last_withdrawal", 500)

# Changes are in memory but not committed yet

`end()`

Commit the active transaction and persist changes to disk.

return

None

Raises:

TransactionError: If no transaction is active
DatabaseIOError: If disk write fails
DatabaseError: If database is closed

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

db.begin()
db.set("balance", 500)
db.end()  # Commits transaction and saves to disk

print(db.get("balance"))  # 500

`rollback()`

Roll back the active transaction, discarding all changes made since begin().

return

None

Raises:

TransactionError: If no transaction is active
DatabaseError: If database is closed

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

db.begin()
db.set("balance", 500)
db.set("overdraft", True)

# Something went wrong, rollback
db.rollback()

print(db.get("balance"))  # Still 1000
print(db.get("overdraft"))  # None (never committed)

Complete transaction example:

db = Database("bank")
db.set("account:1", 1000)
db.set("account:2", 500)
db.commit()

try:
    db.begin()
    
    # Transfer money
    balance1 = db.get("account:1")
    balance2 = db.get("account:2")
    
    db.set("account:1", balance1 - 100)
    db.set("account:2", balance2 + 100)
    
    # Commit transaction
    db.end()
    print("Transfer successful")
    
except Exception as e:
    # Rollback on error
    db.rollback()
    print(f"Transfer failed: {e}")

Lifecycle Methods

`close()`

Close the database, committing any pending changes and preventing further operations.

return

None

db = Database("data")
db.set("key", "value")
db.close()  # Commits and closes

# Further operations will raise DatabaseError
try:
    db.set("another", "value")
except DatabaseError:
    print("Database is closed")

Using as context manager:

# Note: Context manager support requires __enter__ and __exit__
# which are not implemented in the current version.
# Manual close is required:

db = Database("data")
try:
    db.set("key", "value")
finally:
    db.close()

Magic Methods

`len()`

Return the number of records in the database.

db = Database("items")
db.set("a", 1)
db.set("b", 2)
db.set("c", 3)

print(len(db))  # 3

`contains(key)`

Check if a key exists using the in operator.

db = Database("cache")
db.set("token", "abc123")

if "token" in db:
    print("Token found")

if "missing" not in db:
    print("Key not found")

`repr()`

Return a string representation of the database.

db = Database("mydata")
db.set("key", "value")

print(repr(db))
# Output: <Database path='mydata.pkl' entries=1 state=open>

db.close()
print(repr(db))
# Output: <Database path='mydata.pkl' entries=1 state=closed>

Internal Attributes

These attributes are internal implementation details and should not be accessed directly.

_path

Path

Path to the database file with .pkl extension.

_data

dict[str, Any]

In-memory storage of all key-value pairs.

_dirty

bool

Flag indicating if there are uncommitted changes.

_in_transaction

bool

Flag indicating if a transaction is active.

_tx_snapshot

dict[str, Any] | None

Snapshot of data at transaction start, used for rollback.

_closed

bool

Flag indicating if the database has been closed.

Complete Example

from binarydb.database import Database

# Create and load database
db = Database("app_data")
db.load()  # Load existing data if file exists

# Store various types of data
db.set("config", {
    "debug": True,
    "max_connections": 100
})

db.set("users", ["alice", "bob", "charlie"])
db.set("version", "1.0.0")

# Update nested dictionary
db.update("config", {"debug": False})

# Check existence
if db.exists("users"):
    users = db.get("users")
    print(f"Found {len(users)} users")

# Use transactions for atomic operations
db.begin()
try:
    db.set("status", "processing")
    # ... do work ...
    db.set("status", "complete")
    db.end()  # Commit transaction
except Exception as e:
    db.rollback()  # Undo changes
    print(f"Error: {e}")

# Manual commit (if not using transactions)
db.commit()

# Get all data
all_data = db.all()
print(f"Database contains {len(all_data)} records")

# Close database
db.close()

Get Started

Core Concepts

API Reference

Guides

Overview

Constructor

`Database(path)`

Core Methods

`set(key, value)`

`get(key, default=None)`

`delete(key)`

`update(key, changes)`

`exists(key)`

`all()`

Persistence Methods

`commit()`

`load()`

Transaction Methods

`begin()`

`end()`

`rollback()`

Lifecycle Methods

`close()`

Magic Methods

`len()`

`contains(key)`

`repr()`

Internal Attributes

Complete Example

Build docs developers (and LLMs) love

Get Started

Core Concepts

API Reference

Guides

​Overview

​Constructor

​Database(path)

​Core Methods

​set(key, value)

​get(key, default=None)

​delete(key)

​update(key, changes)

​exists(key)

​all()

​Persistence Methods

​commit()

​load()

​Transaction Methods

​begin()

​end()

​rollback()

​Lifecycle Methods

​close()

​Magic Methods

​__len__()

​__contains__(key)

​__repr__()

​Internal Attributes

​Complete Example

Build docs developers (and LLMs) love

Overview

Constructor

`Database(path)`

Core Methods

`set(key, value)`

`get(key, default=None)`

`delete(key)`

`update(key, changes)`

`exists(key)`

`all()`

Persistence Methods

`commit()`

`load()`

Transaction Methods

`begin()`

`end()`

`rollback()`

Lifecycle Methods

`close()`

Magic Methods

`len()`

`contains(key)`

`repr()`

Internal Attributes

Complete Example