ic-python-db 0.9.0

A lightweight key-value database written in Python, intended for use on the Internet Computer (IC)

IC Python DB

A lightweight key-value database with entity relationships and audit logging capabilities, intended for small to medium-sized applications running on the Internet Computer. Forked from kybra-simple-db.

Features

• Persistent Storage: Works with StableBTreeMap stable structure for persistent storage on your canister's stable memory so your data persists automatically across canister upgrades.
• Entity-Relational Database: Create, read and write entities with OneToOne, OneToMany, ManyToOne, and ManyToMany relationships.
• Schema Versioning & Upgrade Safety: Automatic schema introspection, compatibility checking, and auto-migration for safe changes. Breaking changes without a migrate() method are rejected, preventing data corruption on canister upgrades.
• Entity Hooks: Intercept and control entity lifecycle events (create, modify, delete) with on_event hooks.
• Access Control: Thread-safe context management for user identity tracking and ownership-based permissions.
• Namespaces: Organize entities into namespaces to avoid type conflicts when you have multiple entities with the same class name.
• Audit Logging: Track all changes to your data with created/updated timestamps and who created and updated each entity.

Installation

pip install ic-python-db

Quick Start

The database storage must be initialized before using IC Python DB. Here's an example of how to do it:

from basilisk import StableBTreeMap
from ic_python_db import Database

# Initialize storage and database
storage = StableBTreeMap[str, str](memory_id=1, max_key_size=100, max_value_size=1000)  # Use a unique memory ID for each storage instance
Database.init(db_storage=storage)

Read Basilisk's documentation for more information regarding StableBTreeMap and memory IDs.

Next, define your entities:

from ic_python_db import (
    Database, Entity, String, Integer,
    OneToOne, OneToMany, ManyToOne, ManyToMany, TimestampedMixin
)

class Person(Entity, TimestampedMixin):
    __alias__ = "name"  # Use `name` as the alias field for lookup
    name = String(min_length=2, max_length=50)
    age = Integer(min_value=0, max_value=120)
    friends = ManyToMany("Person", "friends")
    mother = ManyToOne("Person", "children")
    children = OneToMany("Person", "mother")
    spouse = OneToOne("Person", "spouse")

Entity Lookup

Entities can be retrieved using Entity[key] syntax with three different lookup modes:

# Create an entity
john = Person(name="John", age=30)

# Lookup by ID
Person[1]                  # Returns john (by _id)
Person["1"]                # Also works with string ID

# Lookup by alias (defined via __alias__)
Person["John"]             # Tries ID first, then alias field "name"

# Lookup by specific field (tuple syntax)
Person["name", "John"]     # Lookup by field "name" only

Syntax	Behavior
Person[1]	Lookup by _id
Person["John"]	Try _id first, then __alias__ field
Person["name", "John"]	Lookup by specific field name only

Then use the defined entities to store objects:

    # Create and save an object
    john = Person(name="John", age=30)

    # Update an object's property
    john.age = 33  # Type checking and validation happens automatically

    # use the `_id` property to load an entity with the [] operator
    Person(name="Peter")
    peter = Person["Peter"]

    # Delete an object
    peter.delete()

    # Create relationships
    alice = Person(name="Alice")
    eva = Person(name="Eva")
    john.mother = alice
    assert john in alice.children
    eva.friends = [alice]
    assert alice in eva.friends
    assert eva in alice.friends

    print(alice.serialize())  # Prints the dictionary representation of an object
    # Prints: {'timestamp_created': '2025-09-12 22:15:35.882', 'timestamp_updated': '2025-09-12 22:15:35.883', 'creator': 'system', 'updater': 'system', 'owner': 'system', '_type': 'Person', '_id': '3', 'name': 'Alice', 'age': None, 'children': '1', 'friends': '4'}

    assert Person.count() == 3
    assert Person.max_id() == 4
    assert Person.instances() == [john, alice, eva]

    # Cursor-based pagination
    assert Person.load_some(0, 2) == [john, alice]
    assert Person.load_some(2, 2) == [eva]

    # Retrieve database contents in JSON format
    print(Database.get_instance().dump_json(pretty=True))

    # Audit log
    audit_records = Database.get_instance().get_audit(id_from=0, id_to=5)
    pprint(audit_records['0'])
    ''' Prints:

    ['save',
    1744138342934,
    'Person@1',
    {'_id': '1',
    '_type': 'Person',
    'age': 30,
    'creator': 'system',
    'name': 'John',
    'owner': 'system',
    'timestamp_created': '2025-04-08 20:52:22.934',
    'timestamp_updated': '2025-04-08 20:52:22.934',
    'updater': 'system'}]

    '''

For more usage examples, see the tests.

Namespaces

Organize entities with the __namespace__ attribute to avoid type conflicts when you have the same class name in different modules:

# In app/models.py
class User(Entity):
    __namespace__ = "app"
    name = String()
    role = String()

# In admin/models.py  
class User(Entity):
    __namespace__ = "admin"
    name = String()
    permissions = String()

from app.models import User as AppUser
from admin.models import User as AdminUser

app_user = AppUser(name="Alice", role="developer")      # Stored as "app::User"
admin_user = AdminUser(name="Bob", permissions="all")     # Stored as "admin::User"

# Each namespace has isolated ID sequences and storage
assert app_user._id == "1"
assert admin_user._id == "1"

Entity Hooks

Intercept and control entity changes with the on_event hook:

from ic_python_db import Entity, String, ACTION_MODIFY

class User(Entity):
    name = String()
    email = String()
    
    @staticmethod
    def on_event(entity, field_name, old_value, new_value, action):
        # Validate email format
        if field_name == "email" and "@" not in new_value:
            return False, None  # Reject invalid email
        
        # Auto-capitalize names
        if field_name == "name":
            return True, new_value.upper()
        
        return True, new_value

user = User(name="alice", email="[e-mail]")
assert user.name == "ALICE"  # Auto-capitalized

See docs/HOOKS.md for more patterns.

Access Control

User context management with as_user():

from ic_python_db import Database, Entity, String, ACTION_MODIFY, ACTION_DELETE
from ic_python_db.mixins import TimestampedMixin
from ic_python_db.context import get_caller_id

class Document(Entity, TimestampedMixin):
    title = String()
    
    @staticmethod
    def on_event(entity, field_name, old_value, new_value, action):
        caller = get_caller_id()
        
        # Only owner can modify or delete
        if action in (ACTION_MODIFY, ACTION_DELETE):
            if entity._owner != caller:
                return False, None
        
        return True, new_value

db = Database.get_instance()

# Alice creates a document
with db.as_user("alice"):
    doc = Document(title="My Doc")  # Owner: alice

# Bob cannot modify Alice's document
with db.as_user("bob"):
    doc.title = "Hacked"  # Raises ValueError

See docs/ACCESS_CONTROL.md and examples/simple_access_control.py.

Type Hints

The library is fully typed (PEP 561 compliant). Type checkers and IDEs automatically infer property types:

class User(Entity):
    name = String()      # Inferred as str
    age = Integer()      # Inferred as int
    active = Boolean()   # Inferred as bool

user = User(name="Alice", age=30, active=True)
user.name   # IDE knows this is str
user.age    # IDE knows this is int

For stricter typing, you can add explicit annotations:

from typing import Optional

class User(Entity):
    name: str = String()
    age: int = Integer()
    profile: Optional["Profile"] = OneToOne("Profile", "user")

Schema Versioning & Upgrade Safety

ic-python-db includes built-in upgrade compatibility checking. The system introspects your Entity class definitions to detect schema changes and ensure safe upgrades.

Auto-migration for safe changes

Adding a field with a default value requires no migration code — the system handles it automatically:

# v1
class Product(Entity):
    __version__ = 1
    name = String()

# v2 — just add the field with a default, no migrate() needed
class Product(Entity):
    __version__ = 2
    name = String()
    price = Float(default=0.0)    # auto-injected for existing entities
    active = Boolean(default=True) # auto-injected for existing entities

Custom migration for breaking changes

For type changes, field renames, or data transformations, override migrate():

class Product(Entity):
    __version__ = 2
    name = String()
    price_dollars = Float()  # was price_cents: Integer in v1

    @classmethod
    def migrate(cls, obj, from_version, to_version):
        if from_version == 1:
            obj["price_dollars"] = obj.pop("price_cents") / 100.0
        return obj

Upgrade compatibility enforcement

Call check_upgrade_compatibility() from your canister's post_upgrade to reject incompatible upgrades before they corrupt data:

from basilisk import post_upgrade
from ic_python_db import Database

@post_upgrade
def on_post_upgrade():
    db = Database.get_instance()
    db.check_upgrade_compatibility()
    # If a breaking change lacks migrate(), this raises SchemaIncompatibleError
    # which traps post_upgrade, causing the IC to roll back the upgrade

The system classifies schema changes as:

Change	Safety	Action
Add field with default=	Safe	Auto-migrated
Remove field	Safe	Old data ignored
Add new Entity type	Safe	No migration needed
Change field type	Breaking	Requires migrate()
Add field without default	Breaking	Requires migrate()
Change relationship type	Breaking	Requires migrate()

Schema introspection

You can inspect and compare schemas programmatically:

from ic_python_db import Database, build_schema, diff_schemas

db = Database.get_instance()

# Build current schema from Entity definitions
schema = db.build_schema_from_entities()

# Compare two schemas
changes = diff_schemas(old_schema, new_schema)
for change in changes:
    print(f"{change.entity_type}.{change.field}: {change.reason}")

See docs/SCHEMA_VERSIONING.md for the full reference.

API Reference

• Core: Database, Entity
• Properties: String, Integer, Float, Boolean
• Relationships: OneToOne, OneToMany, ManyToOne, ManyToMany
• Mixins: TimestampedMixin (timestamps and ownership tracking)
• Hooks: ACTION_CREATE, ACTION_MODIFY, ACTION_DELETE
• Context: get_caller_id(), set_caller_id(), Database.as_user()
• Schema: build_schema, diff_schemas, schema_hash, SchemaIncompatibleError

Development

Setup Development Environment

# Clone the repository
git clone https://github.com/smart-social-contracts/ic-python-db.git
cd ic-python-db

# Recommended setup
pyenv install 3.10.7
pyenv local 3.10.7
python -m venv venv
source venv/bin/activate

# Install development dependencies
pip install -r requirements-dev.txt

# Running tests
./run_linters.sh && (cd tests && ./run_test.sh && ./run_test_ic.sh)

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

MIT.