---
title: "Storage"
slug: "storage"
updated: 2025-10-29T00:14:00Z
published: 2025-10-29T00:14:00Z
---

> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vention.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Storage

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

## Table of Contents

- [✨ Features](/docs/storage#-features)
- [🧠 Concepts & Overview](/docs/storage#-concepts--overview)
- [⚙️ Installation & Setup](/docs/storage#%EF%B8%8F-installation--setup)
- [🚀 Quickstart Tutorial](/docs/storage#-quickstart-tutorial)
- [🛠 How-to Guides](/docs/storage#-how-to-guides)
- [📖 API Reference](/docs/storage#-api-reference)
- [🔍 Troubleshooting & FAQ](/docs/storage#-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
- 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.