DictDB is an in‑memory, dictionary-based database system for Python. It provides SQL‑like CRUD operations, schema validation, logging, indexing, and a fluent interface for building complex query conditions. DictDB can be used for rapid prototyping, testing, or any scenario where a relational‑style workflow is needed in‑memory without the overhead of a full database engine.
- Features
- Installation
- Basic Usage
- Working With Conditions and Queries
- Schema Validation
- Logging
- Persistence, Backup, and Indexing
- API Reference
- Roadmap
- Multiple Tables: Manage any number of in‑memory tables within a single DictDB.
- SQL‑like CRUD: Insert, select, update, and delete records with a Pythonic API reminiscent of SQL.
- Schema Validation: Enforce data consistency by defining a field schema for each table.
- Condition/Query System: Build complex filter expressions using logical operators (&, |, ~) and Python comparison operators (==, !=, <, <=, >, >=).
- Atomic Updates: All updates are applied atomically; if a single record fails schema validation during an update, all changes are rolled back.
- Logging and Debugging: Integrated with Loguru for flexible, configurable logging.
- Persistence and Backup: Save and load database state synchronously or asynchronously, with automatic backup support.
- Indexing: Optimize query performance by creating indices on table fields.
- Easy Testing: Fully unit-tested codebase using pytest.
DictDB is structured as a standard Python package using pyproject.toml. To install locally (for development or local usage), run:
git clone https://github.com/mhbxyz/dictdb.git
cd dictdb
pip install .When published to PyPI (NOT PUBLISHED YET), you could install it with:
pip install dictdbBelow is a short, end-to-end example showcasing DictDB usage:
from dictdb import DictDB, Query, configure_logging
# Optional logging
configure_logging(level="DEBUG", console=True)
# Create DB and tables
db = DictDB()
db.create_table("employees", primary_key="emp_id")
employees = db.get_table("employees")
# Insert data
employees.insert({"emp_id": 101, "name": "Alice", "department": "IT"})
employees.insert({"emp_id": 102, "name": "Bob", "department": "HR"})
employees.insert({"name": "Charlie", "department": "IT"}) # auto-assigns emp_id=103
# Query data
it_staff = employees.select(where=Query(employees.department == "IT"))
print("IT Department Staff:", it_staff)
# Update data
employees.update({"department": "Engineering"}, where=Query(employees.name == "Alice"))
# Delete data
employees.delete(where=Query(employees.name == "Bob"))
# Print final state
print("All employees:", employees.all())from dictdb import DictDB
# Create a new in-memory DictDB instance
db = DictDB()
# Create tables (default primary key is "id")
db.create_table("users")
db.create_table("products", primary_key="product_id")
# List all tables
print(db.list_tables()) # e.g., ["users", "products"]# Get the "users" table
users = db.get_table("users")
# Insert a record with an explicit primary key
users.insert({"id": 1, "name": "Alice", "age": 30})
# Insert a record without specifying the primary key;
# DictDB will auto-generate one.
users.insert({"name": "Bob", "age": 25})
# Insert into the "products" table
products = db.get_table("products")
products.insert({"product_id": 1001, "name": "Laptop", "price": 999.99})# Select all records (equivalent to SQL's SELECT *)
all_users = users.select()
print(all_users)
# Select only specific columns
name_only = users.select(columns=["name"])
print(name_only)from dictdb import Query
# Update a single record with a condition
rows_updated = users.update({"age": 26}, where=Query(users.name == "Bob"))
print(rows_updated)
# Update all records
rows_updated = users.update({"age": 99})
print(rows_updated)# Delete a record that meets a condition
deleted_count = users.delete(where=Query(users.name == "Alice"))
print(deleted_count)
# Delete all records in the table
deleted_count = users.delete()
print(deleted_count)DictDB allows you to build filter conditions using overloaded Python operators.
from dictdb import Query
# Suppose we have a "users" table with fields: id, name, age
users = db.get_table("users")
# Build a condition for "age == 30"
condition = Query(users.age == 30)
# Apply the condition in a select
matching_users = users.select(where=condition)# (name == "Alice") AND (age > 25)
condition = Query((users.name == "Alice") & (users.age > 25))
result = users.select(where=condition)
# (name == "Alice") OR (age > 25)
condition_or = Query((users.name == "Alice") | (users.age > 25))
# NOT (name == "Alice")
condition_not = Query(~(users.name == "Alice"))from dictdb import Table, SchemaValidationError
# Define a schema for users
schema = {
"id": int,
"name": str,
"age": int
}
# Create the table with a schema
table_with_schema = Table("schema_users", primary_key="id", schema=schema)
# Insert a valid record
table_with_schema.insert({"id": 1, "name": "Alice", "age": 30})If a record violates the schema, DictDB raises a SchemaValidationError:
try:
# Missing 'age' field
table_with_schema.insert({"id": 2, "name": "Bob"})
except SchemaValidationError as e:
print(f"Schema error: {e}")
try:
# Extra field 'nickname' not in schema
table_with_schema.insert({"id": 3, "name": "Charlie", "age": 28, "nickname": "Chaz"})
except SchemaValidationError as e:
print(f"Schema error: {e}")
try:
# Wrong type for 'age'
table_with_schema.insert({"id": 4, "name": "Diana", "age": "30"})
except SchemaValidationError as e:
print(f"Schema error: {e}")DictDB integrates with Loguru for detailed logging.
from dictdb import configure_logging, DictDB
# Configure logging to both console and a file
configure_logging(level="DEBUG", console=True, logfile="dictdb.log")
db = DictDB() # Logs "Initialized an empty DictDB instance."- Console sink: Set
console=Trueto log to stdout. - File sink: Specify
logfile="yourfile.log"to write logs to a file. - Log Levels: Options include "DEBUG", "INFO", "WARNING", etc.
DictDB supports saving and loading database states, as well as automatic backups and indexing to enhance performance.
You can persist the database state in JSON or pickle formats. DictDB offers both synchronous and asynchronous methods:
# Synchronous save and load
db.save("backup.json", "json")
loaded_db = DictDB.load("backup.json", "json")
# Asynchronous save and load (requires an async context)
await db.async_save("backup_async.json", "json")
loaded_db = await DictDB.async_load("backup_async.json", "json")The BackupManager provides automatic periodic backups or immediate backups after significant changes.
from dictdb import BackupManager
# Create a backup manager with a backup interval of 300 seconds
backup_manager = BackupManager(db, backup_dir="backups", backup_interval=300, file_format="json")
backup_manager.start()
# Trigger an immediate backup when needed
backup_manager.notify_change()
# Stop the backup manager gracefully
backup_manager.stop()To optimize query performance, you can create indices on table fields. DictDB supports both "hash" and "sorted" index types:
# Create an index on the "age" field using a hash index
users.create_index("age", index_type="hash")
# Alternatively, create a sorted index:
users.create_index("age", index_type="sorted")class DictDB:
def __init__(self) -> None:
"""Initializes an empty DictDB instance."""
def create_table(self, table_name: str, primary_key: str = 'id') -> None:
"""Creates a new table with the specified name and optional primary key."""
def drop_table(self, table_name: str) -> None:
"""Removes a table from the database by name."""
def get_table(self, table_name: str) -> Table:
"""Retrieves a Table object by its name."""
def list_tables(self) -> List[str]:
"""Returns a list of all table names in this DictDB."""
def save(self, filename: str, file_format: str) -> None:
"""Saves the current state of the database in the specified format."""
def load(self, filename: str, file_format: str) -> DictDB:
"""Loads a database state from the specified file."""
async def async_save(self, filename: str, file_format: str) -> None:
"""Asynchronously saves the current state of the database."""
@classmethod
async def async_load(cls, filename: str, file_format: str) -> DictDB:
"""Asynchronously loads the database state from the specified file."""Provides CRUD operations and schema validation.
class Table:
def __init__(self, name: str, primary_key: str = 'id', schema: Optional[Dict[str, type]] = None) -> None:
"""Creates a new table with optional schema validation."""
def insert(self, record: Dict[str, Any]) -> None:
"""Inserts a new record, with auto-assignment of the primary key if necessary."""
def select(self, columns: Optional[List[str]] = None, where: Optional[Query] = None) -> List[Dict[str, Any]]:
"""Selects records matching a condition, with optional column projection."""
def update(self, changes: Dict[str, Any], where: Optional[Query] = None) -> int:
"""Updates records matching a condition. Returns the number of records updated."""
def delete(self, where: Optional[Query] = None) -> int:
"""Deletes records matching a condition. Returns the number of records deleted."""
def all(self) -> List[Dict[str, Any]]:
"""Returns all records in the table."""
def create_index(self, field_name: str, index_type: str = "hash") -> None:
"""Creates an index on the specified field. Supported types: 'hash', 'sorted'."""class Condition:
"""Represents a condition to be applied to records. Supports logical operators (&, |, ~)."""
class Query:
"""A wrapper for Condition objects to prevent implicit boolean conversion."""DictDBError: Base exception for DictDB errors.DuplicateKeyError: Raised when inserting a record with a duplicate primary key.RecordNotFoundError: Raised when no records match a query.SchemaValidationError: Raised when a record violates the defined schema.
class BackupManager:
def __init__(self, db: DictDB, backup_dir: Union[str, Path], backup_interval: int = 300, file_format: str = "json") -> None:
"""Initializes the BackupManager."""
def start(self) -> None:
"""Starts automatic periodic backups."""
def stop(self) -> None:
"""Stops the backup manager."""
def backup_now(self) -> None:
"""Performs an immediate backup."""
def notify_change(self) -> None:
"""Triggers an immediate backup after a significant change."""See roadmap.md for planned features and enhancements.
DictDB is an evolving project. Contributions, suggestions, and bug reports are welcome!