Storage

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

✨ Features

Persistent storage with SQLite
Automatic audit trails (who, when, what changed)
Strong typing & validation via SQLModel
Lifecycle hooks before/after insert, update, delete
Soft delete with deleted_at fields
REST API generation with Create, Read, Update, Delete + audit
Health & monitoring endpoints (audit log, schema diagram)
Batch operations for insert/delete
Session management with smart reuse & transactions
Bootstrap system for one-command setup
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
AuditLog → Automatically records all data mutations
Routers → Auto-generated FastAPI endpoints for Create, Read, Update, Delete + database management

⚙️ Installation & Setup

pip install vention-storage

Optional dependencies:

sqlalchemy-schemadisplay and Graphviz → enable database schema visualization

🚀 Quickstart Tutorial

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

from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
from fastapi import FastAPI
from storage.bootstrap import bootstrap
from storage.accessor import ModelAccessor

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)

app = FastAPI()
user_accessor = ModelAccessor(User, "users")

bootstrap(
    app,
    accessors=[user_accessor],
    database_url="sqlite:///./my_app.db",
    create_tables=True
)

➡️ You now have Create, Read, Update, Delete, audit, backup, and CSV endpoints at /users and /db.

🛠 How-to Guides

Bootstrap Multiple Models

product_accessor = ModelAccessor(Product, "products")

bootstrap(
    app,
    accessors=[user_accessor, product_accessor],
    database_url="sqlite:///./my_app.db",
    create_tables=True,
    max_records_per_model=100,
    enable_db_router=True
)

Export to CSV

response = requests.get("http://localhost:8000/db/export.zip")
with open("backup.zip", "wb") as f:
    f.write(response.content)

Backup & Restore

# Backup
r = requests.get("http://localhost:8000/db/backup.sqlite")
with open("backup.sqlite", "wb") as f: f.write(r.content)

# Restore
with open("backup.sqlite", "rb") as f:
    files = {"file": ("backup.sqlite", f, "application/x-sqlite3")}
    requests.post("http://localhost:8000/db/restore", files=files)

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}")

Query Audit Logs

from storage.auditor import AuditLog
from sqlmodel import select

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

📖 API Reference

bootstrap

def bootstrap(
    app: FastAPI,
    *,
    accessors: Iterable[ModelAccessor[Any]],
    database_url: Optional[str] = None,
    create_tables: bool = True,
    max_records_per_model: Optional[int] = 5,
    enable_db_router: bool = True,
) -> None

ModelAccessor

ModelAccessor(model: Type[ModelType], component_name: str)

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
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()

Routers

build_crud_router(accessor, max_records=100) -> APIRouter
build_db_router(audit_default_limit=100, audit_max_limit=1000) -> APIRouter

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]

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 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.