Vention Storage

A framework for storing and managing component and application data with persistence, validation, and audit trails for machine applications.

Table of Contents

• ✨ Features
• 🧠 Concepts & Overview
• ⚙️ Installation & Setup
• 🚀 Quickstart Tutorial
• 🛠 How-to Guides
• 📖 API Reference
• 🔍 Troubleshooting & FAQ

✨ Features

• Persistent storage with SQLite
• Automatic audit trails (who, when, what changed)
• Strong typing & validation via SQLModel
• Lifecycle hooks before/after insert, update, delete
• Commit listeners: react once a transaction's changes are committed, filtered by table, operation or value
• Soft delete with deleted_at fields
• ConnectRPC bundle generation with Create, Read, Update, Delete + audit
• Health & monitoring actions (audit log, schema diagram)
• Batch operations for insert/delete
• Session management with smart reuse & transactions
• Bootstrap system for one-command setup
• Opt-in SQLite auto-migrations for safe additive schema changes
• CSV export/import for backups and migration
• Database backup/restore with integrity checking

🧠 Concepts & Overview

Vention Storage is a component-based persistence layer for machine apps:

• Database → SQLite database with managed sessions and transactions
• ModelAccessor → Strongly-typed Create, Read, Update, Delete interface for your SQLModel classes
• Hooks → Functions that run before/after Create, Read, Update, Delete operations
• Commit listeners → Functions that run after a transaction that changed matching rows is committed
• AuditLog → Automatically records all data mutations
• RpcBundle → Auto-generated ConnectRPC bundle with Create, Read, Update, Delete + database management actions

⚙️ Installation & Setup

pip install vention-storage

Optional dependencies:

• sqlalchemy-schemadisplay and Graphviz → enable database schema visualization

MacOS:

brew install graphviz
pip install sqlalchemy-schemadisplay

Linux (Debian/Ubuntu)

sudo apt-get install graphviz
pip install sqlalchemy-schemadisplay

🚀 Quickstart Tutorial

Define a model, bootstrap storage, and get full Create, Read, Update, Delete RPC actions in minutes:

from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
from communication.app import VentionApp
from storage.bootstrap import bootstrap
from storage.accessor import ModelAccessor
from storage.vention_communication import build_storage_rpc_bundle

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str
    deleted_at: Optional[datetime] = Field(default=None, index=True)

# Initialize database
bootstrap(
    database_url="sqlite:///./my_app.db",
    auto_migrate=True,
)

# Create accessor
user_accessor = ModelAccessor(User, "users")

# Build RPC bundle and add to app
app = VentionApp(name="my-app")
storage_bundle = build_storage_rpc_bundle(
    accessors=[user_accessor],
    max_records_per_model=100,
    enable_db_actions=True
)
app.register_rpc_plugin(storage_bundle)
app.finalize()

➡️ You now have Create, Read, Update, Delete, audit, backup, CSV, and opt-in startup migrations available via ConnectRPC.

auto_migrate=True is intentionally conservative. It supports creating missing tables, adding safe columns, backfilling scalar defaults, and creating missing indexes. Destructive or ambiguous changes still require a manual migration.

🛠 How-to Guides

Bootstrap Multiple Models

# user_accessor was created earlier in the Quickstart example
# Reuse it here to bootstrap multiple models at once

class Product(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    sku: str

product_accessor = ModelAccessor(Product, "products")

# Build bundle with multiple accessors
storage_bundle = build_storage_rpc_bundle(
    accessors=[user_accessor, product_accessor],
    max_records_per_model=100,
    enable_db_actions=True
)
app.register_rpc_plugin(storage_bundle)

Export to CSV

# Using ConnectRPC client
from communication.client import ConnectClient

client = ConnectClient("http://localhost:8000")
response = await client.call("Database_ExportZip", {})
with open("backup.zip", "wb") as f:
    f.write(response.data)

Backup & Restore

# Backup
backup_response = await client.call("Database_BackupSqlite", {})
with open(backup_response.filename, "wb") as f:
    f.write(backup_response.data)

# Restore
with open("backup.sqlite", "rb") as f:
    restore_response = await client.call(
        "Database_RestoreSqlite",
        {
            "bytes": f.read(),
            "filename": "backup.sqlite",
            "integrity_check": True,
            "dry_run": False
        }
    )

Use Lifecycle Hooks

@user_accessor.before_insert()
def validate_email(session, instance):
    if "@" not in instance.email:
        raise ValueError("Invalid email")

@user_accessor.after_insert()
def log_creation(session, instance):
    print(f"User created: {instance.name}")

React to Committed Changes

Lifecycle hooks run inside the transaction, before it commits. To react only once a change is really in the database (refresh a cache, reload a configuration, notify another thread), use after_commit:

from storage.changes import Change

@user_accessor.after_commit()
def reload_users(changes: list[Change]) -> None:
    user_cache_is_stale.set()

The function is called once per committed transaction that wrote to the accessor's table, with that table's changes in one list. It is not called if the transaction only read, or if it rolled back. A savepoint (session.begin_nested()) is only reported once the outer transaction commits, and not at all if it rolled back. Saving a row with identical values is not a change.

To watch several tables, decorate one function with each accessor. It is still called once per transaction, with the changes to all of them:

@device_accessor.after_commit()
@rule_accessor.after_commit()
def reload_rules(changes: list[Change]) -> None:
    rules_are_stale.set()

This covers writes made through RPCs too: the generated <Model>_CreateRecord, _UpdateRecord, _DeleteRecord and _RestoreRecord actions write through the accessors, and so do your own actions if they use an accessor or a Session.

Each Change has table, operation ("create", "update" or "delete", the same names as the audit log), values (the row after the write, or the deleted row) and before (the row before an update, else None).

Restoring a backup (Database_RestoreSqlite / DatabaseService.restore_sqlite) replaces the whole database file, so any row may have changed. It has its own listener, called with no arguments:

from storage.changes import on_database_restored

@on_database_restored
def reload_everything() -> None:
    user_cache_is_stale.set()

Good to know:

• Changes are recorded from SQLAlchemy itself, so every ORM write to a watched table counts: accessors (audited or not), rows removed by a relationship cascade, and plain Session writes. Raw SQL (session.execute(text(...))), database-level ON DELETE CASCADE and other processes are not seen, including an RPC action of your own that runs raw SQL.
• A soft delete is reported as an "update" that sets deleted_at, and restoring a soft-deleted row as an "update" that clears it. Only a backup restore calls the on_database_restored listeners.
• If you replace the database file some other way, call changes.notify_database_restored() yourself.
• Listeners run on the thread that committed, right after the commit. Inside database.transaction() or database.use_session() that is after the with block, so a listener can read and write through an accessor. Changes committed before the block raised are still reported. Keep listeners short (set a flag, queue work). An exception is logged and swallowed, because the data is already committed.
• A listener that writes to a watched table is called again from inside its own call, before the other listeners hear about the first transaction. Never write to a listener's own table from it: it recurses until Python's recursion limit.
• Like the other hooks, a listener can't be removed once registered.
• values and before are copies of the column values, safe to keep. When the old value isn't loaded (a session with expire_on_commit=True, or a JSON column changed in place), before is read from the database, one query per row.
• Only tables with an after_commit listener are recorded, and an app with none pays nothing.

Query Audit Logs

from storage.auditor import AuditLog
from storage import database
from sqlmodel import select

with database.transaction() as session:
    logs = session.exec(select(AuditLog).where(AuditLog.component == "users")).all()

Using the model accessors

# Create
user = user_accessor.insert(User(name="Alice", email="alice@example.com"), actor="admin")

# Read
user = user_accessor.get(user.id)

# Update some fields (returns None if there's no such user)
user_accessor.update(user.id, actor="admin", name="Alice Smith")

# Save a whole object (inserts it if it doesn't exist)
user.name = "Alice Smith"
user_accessor.save(user, actor="admin")

# Delete
user_accessor.delete(user.id, actor="admin")

# Restore (for soft-deleted models)
user_accessor.restore(user.id, actor="admin")

# Find users by exact match
users = user_accessor.find(user_accessor.where.email == "alice@example.com")

# Multiple conditions (AND logic)
users = user_accessor.find(
    user_accessor.where.name == "Alice",
    user_accessor.where.email == "alice@example.com"
)

# Comparison operators
recent_users = user_accessor.find(user_accessor.where.id > 10)
first_users = user_accessor.find(user_accessor.where.id <= 5)

# String operations
smiths = user_accessor.find(user_accessor.where.name.contains("Smith"))
gmail_users = user_accessor.find(user_accessor.where.email.endswith("@gmail.com"))
search = user_accessor.find(user_accessor.where.name.ilike("%alice%"))  # case-insensitive

# Collection check
selected_users = user_accessor.find(
    user_accessor.where.email.in_(["alice@example.com", "bob@example.com"])
)

# Null checks
active_users = user_accessor.find(user_accessor.where.deleted_at.is_(None))
deleted_users = user_accessor.find(user_accessor.where.deleted_at.isnot(None))

# With pagination and sorting
page = user_accessor.find(
    user_accessor.where.email.contains("@example.com"),
    limit=10,
    offset=20,
    order_by="id",
    order_desc=True
)

# Include soft-deleted records
all_users = user_accessor.find(
    user_accessor.where.email.contains("@example.com"),
    include_deleted=True
)

Using ConnectRPC Client

Once the bundle is added to your VentionApp, each ModelAccessor automatically exposes full CRUD actions via ConnectRPC.

Example: interacting with the Users RPC actions.

import { createPromiseClient } from "@connectrpc/connect";
import { createConnectTransport } from "@connectrpc/connect-web";

const transport = createConnectTransport({
  baseUrl: "http://localhost:8000",
});

const client = createPromiseClient(YourServiceClient, transport);

// Create
export async function createUser(name: string, email: string) {
  const res = await client.usersCreateRecord({
    record: { name, email },
    actor: "operator"
  });
  return res.record;
}

// Read
export async function getUser(id: number) {
  const res = await client.usersGetRecord({
    recordId: id,
    includeDeleted: false
  });
  return res.record;
}

// Update
export async function updateUser(id: number, name: string) {
  const res = await client.usersUpdateRecord({
    recordId: id,
    record: { name },
    actor: "operator"
  });
  return res.record;
}

// Delete (soft delete if model supports deleted_at)
export async function deleteUser(id: number) {
  await client.usersDeleteRecord({
    recordId: id,
    actor: "operator"
  });
}

// Restore
export async function restoreUser(id: number) {
  const res = await client.usersRestoreRecord({
    recordId: id,
    actor: "operator"
  });
  return res.record;
}

// List
export async function listUsers() {
  const res = await client.usersListRecords({
    includeDeleted: false
  });
  return res.records;
}

// Find by exact match
export async function findUserByEmail(email: string) {
  const res = await client.usersFindRecords({
    filters: [
      { field: "email", operation: "eq", value: email }
    ]
  });
  return res.records;
}

// Find with multiple conditions (AND logic)
export async function findNamedUser(name: string, email: string) {
  const res = await client.usersFindRecords({
    filters: [
      { field: "name", operation: "eq", value: name },
      { field: "email", operation: "eq", value: email }
    ]
  });
  return res.records;
}

// Find with null checks
export async function findDeletedUsers() {
  const res = await client.usersFindRecords({
    filters: [
      { field: "deleted_at", operation: "is_not_null" }
    ]
  });
  return res.records;
}

// Complex query example
export async function findRecentUsers(minId: string) {
  const res = await client.usersFindRecords({
    filters: [
      { field: "email", operation: "contains", value: "@example.com" },
      { field: "id", operation: "gte", value: minId },
      { field: "deleted_at", operation: "is_null" }
    ],
    limit: 50,
    orderBy: "id",
    orderDesc: true
  });
  return res.records;
}

Filter Operations Reference

📖 API Reference

bootstrap

def bootstrap(
    *,
    database_url: Optional[str] = None,
    auto_migrate: bool = False,
) -> None

Initialize the database engine and ensure the schema exists.

When auto_migrate=False, bootstrap() creates all tables from SQLModel.metadata.

When auto_migrate=True, bootstrap() plans and applies supported SQLite schema migrations using SQLModel.metadata. All models must be imported before calling bootstrap() so they are present in SQLModel.metadata.

build_storage_rpc_bundle

def build_storage_rpc_bundle(
    *,
    accessors: Sequence[ModelAccessor[Any]],
    max_records_per_model: Optional[int] = 5,
    enable_db_actions: bool = True,
) -> RpcBundle

Build a ConnectRPC RpcBundle exposing CRUD and database utilities. Returns an RpcBundle that can be registered on a VentionApp using app.register_rpc_plugin().

ModelAccessor

ModelAccessor(
    model: Type[ModelType],
    component_name: str,
    *,
    enable_auditing: bool = True,
)

Read

• get(id, include_deleted=False) -> Optional[ModelType]
• all(include_deleted=False) -> List[ModelType]

Write

• insert(obj, actor="internal") -> ModelType
• save(obj, actor="internal") -> ModelType
• update(id, actor="internal", **fields) -> Optional[ModelType]
• delete(id, actor="internal") -> bool
• restore(id, actor="internal") -> bool

Batch

• insert_many(objs, actor="internal") -> List[ModelType]
• delete_many(ids, actor="internal") -> int

Hooks

• @accessor.before_insert()
• @accessor.after_insert()
• @accessor.before_update()
• @accessor.after_update()
• @accessor.before_delete()
• @accessor.after_delete()
• @accessor.after_commit(): called with list[Change] after a commit that wrote to this table

Parameters

• enable_auditing: If False, disables audit logging for this accessor. Useful for models that shouldn't be audited (e.g., audit logs themselves). Defaults to True.

Database Helpers

• database.set_database_url(url: str) -> None
• database.get_engine() -> Engine
• database.transaction() -> Iterator[Session]
• database.use_session(session: Optional[Session] = None) -> Iterator[Session]

Commit Listeners

• @accessor.after_commit(): see Hooks above
• changes.on_database_restored(listener): call listener() after a backup restore; usable as a decorator
• changes.notify_database_restored() -> None: call those listeners yourself after replacing the database file
• changes.Change: frozen dataclass with table, operation, values, before

AuditLog model

class AuditLog(SQLModel, table=True):
    id: int
    timestamp: datetime
    component: str
    record_id: int
    operation: str
    actor: str
    before: Optional[Dict[str, Any]]
    after: Optional[Dict[str, Any]]

🔍 Troubleshooting & FAQ

• Diagram endpoint fails → Ensure Graphviz + sqlalchemy-schemadisplay are installed.
• No audit actor shown → Provide an X-User header in API requests.
• Soft delete not working → Your model must have a deleted_at field.
• Restore fails → Ensure integrity_check=True passes when restoring backups.
• after_commit listener never called → Check the write went to that accessor's table and was committed through the ORM, not raw SQL. A backup restore calls the on_database_restored listeners instead.
• Auto-migrate rejects my schema change → Only additive SQLite changes are supported automatically. Type changes, destructive edits, FK/PK changes, unique-constraint additions, and ambiguous renames still need a manual migration.