Storage Backends

Pluggable storage backends for training, sessions, and data persistence

Storage Backends

PraisonAI provides pluggable storage backends for training data, session state, and general persistence. Switch between file-based and database storage without changing your application code.

Supported Backends

Backend	Type	Dependencies	Best For
FileBackend	JSON files	None (built-in)	Development, simple deployments
SQLiteBackend	SQLite DB	None (built-in)	Production, concurrent access
PostgreSQL	Via adapters	`psycopg2`	Large-scale production
MySQL	Via adapters	`mysql-connector`	Enterprise deployments
Redis	Via adapters	`redis`	High-speed caching

Quick Start

File Backend (Default)

from praisonaiagents.storage import FileBackend, BaseJSONStore

# File-based storage (default, zero config)
backend = FileBackend(storage_dir="~/.praisonai/data")
store = BaseJSONStore("session.json", backend=backend)

store.save({"messages": ["Hello", "World"]})
data = store.load()

SQLite Backend

from praisonaiagents.storage import SQLiteBackend, BaseJSONStore

# SQLite for better concurrency and performance
backend = SQLiteBackend(db_path="~/.praisonai/data.db")
store = BaseJSONStore("session.json", backend=backend)

store.save({"messages": ["Hello", "World"]})
data = store.load()

Factory Function

from praisonaiagents.storage import get_backend

# Use factory for easy switching
backend = get_backend("sqlite", db_path="data.db")
# or
backend = get_backend("file", storage_dir="./data")

Any Component with Any Backend

All storage components support pluggable backends. Create a backend once and use it with any component:

python

from praisonaiagents.storage import SQLiteBackend, RedisBackend

# SQLite for production
backend = SQLiteBackend(db_path="~/.praisonai/data.db")

# Redis for distributed systems
# backend = RedisBackend(url="redis://localhost:6379", prefix="praison:", ttl=3600)

# Use with any component
from praisonai.recipe.history import RunHistory
from praisonai.cli.state.sessions import SessionManager
from praisonai.mcp_server.tool_index import MCPToolIndex
from praisonai.train.agents.storage import TrainingStorage

history = RunHistory(backend=backend)
manager = SessionManager(backend=backend)
index = MCPToolIndex(backend=backend)

# TrainingStorage is a context manager — closes its backend on exit
with TrainingStorage(session_id="train-123", backend=backend) as storage:
    storage.save_iteration(...)

FileBackend

JSON file-based storage. Each key becomes a separate .json file.

python

from praisonaiagents.storage import FileBackend

backend = FileBackend(
    storage_dir="~/.praisonai/storage",  # Directory for files
    suffix=".json",                     # File extension
    pretty=True                         # Pretty-print JSON
)

# CRUD operations
backend.save("session_123", {"user": "alice", "messages": []})
data = backend.load("session_123")
backend.delete("session_123")
keys = backend.list_keys(prefix="session_")
exists = backend.exists("session_123")

Features:

Zero dependencies (uses built-in json)
Thread-safe with atomic writes — a failed save() (serialization error, encoding error, or os.replace failure) leaves the previous store intact and removes the staging .tmp file. See Thread Safety → BaseJSONStore atomic save.
Human-readable files
Easy debugging and inspection

Failure Modes

FileBackend.save is all-or-nothing: either the destination file is replaced with the new content, or it stays exactly as it was before the call.

Failure	What propagates	Destination file	Staging file
Circular reference in data	`ValueError`	Unchanged	Removed
Non-JSON-serialisable key (e.g. tuple)	`TypeError`	Unchanged	Removed
Disk I/O error during write	`OSError`	Unchanged	Removed
Replacement step fails	Original exception	Unchanged	Removed

No partial files are left behind on any failure path, so re-running the save after fixing the data is safe — the storage directory never fills up with orphaned staging files.

Error Handling

load() never raises on a corrupted file — it logs a warning and returns None. Your agent keeps running and other keys keep loading.

python

from praisonaiagents.storage import FileBackend

backend = FileBackend(storage_dir="~/.praisonai/data")

# A corrupted file (invalid UTF-8, truncated JSON, bad permissions)…
value = backend.load("session_42")    # → None, warning logged
                                      # → file on disk is untouched

# …never breaks neighbouring keys.
other = backend.load("session_43")    # → still works

Failure	What `load()` does
Key missing	returns `None`
Malformed JSON	returns `None`, logs a warning, marks path unreadable
Invalid UTF-8 bytes	returns `None`, logs a warning, marks path unreadable
File unreadable (`IOError`)	returns `None`, logs a warning, marks path unreadable

Save-side guard: an unreadable key cannot be overwritten by ordinary saves

Once backend.load(key) observes a decode failure on a key (invalid UTF-8, malformed JSON, or IOError), the same backend instance refuses an ordinary backend.save(key, data) while the unreadable file is still on disk. Neighbouring, readable keys stay fully usable — the guard is per-path.

python

from pathlib import Path
from praisonaiagents.storage import FileBackend

backend = FileBackend(storage_dir="~/.praisonai/data")

# Simulate corruption out-of-band…
Path("~/.praisonai/data/session_42.json").expanduser().write_bytes(b"\xff")

backend.load("session_42")               # → None, warning logged, guard set
backend.save("session_42", {"ok": 1})    # → raises; the bad file is NOT overwritten
backend.load("session_43")               # → still works — guard is per-key

The guard clears when any of these happens:

Trigger	Who runs it
`backend.delete(key)` succeeds	your code
The source file is removed from disk	out-of-band repair
A later `backend.load(key)` succeeds (bytes replaced with valid JSON)	your code / out-of-band repair

Writes to other keys are never affected.

SQLiteBackend

SQLite database storage. All data in a single .db file.

python

from praisonaiagents.storage import SQLiteBackend

backend = SQLiteBackend(
    db_path="~/.praisonai/storage.db",  # Database file
    table_name="praison_storage",      # Table name
    auto_create=True                   # Create table if missing
)

# Same API as FileBackend
backend.save("session_123", {"user": "alice", "messages": []})
data = backend.load("session_123")
backend.delete("session_123")
keys = backend.list_keys(prefix="session_")
exists = backend.exists("session_123")

# Close when done (optional, auto-closes on exit)
backend.close()

Features:

Zero dependencies (uses built-in sqlite3)
ACID transactions
Better concurrent access
Faster for large datasets
Single file deployment

Table name rules

table_name accepts letters, digits, and underscores only. The default is praison_storage.

Diagram

SQL keywords and digit-leading names are safe. The backend quotes the identifier internally ("select", "9table"), so names like select, order, 9table, or 123 work with nothing extra required.

python

from praisonaiagents.storage import SQLiteBackend

# Default table name
backend = SQLiteBackend(db_path="data.db")  # table "praison_storage"

# SQL keyword — safe, identifier is quoted internally
backend = SQLiteBackend(db_path="data.db", table_name="select")

# Starts with a digit — safe
backend = SQLiteBackend(db_path="data.db", table_name="9table")

# Invalid — raises ValueError at construction
SQLiteBackend(db_path="data.db", table_name="bad name")   # space
SQLiteBackend(db_path="data.db", table_name='drop";')     # quote

Invalid characters are rejected at construction. Names with spaces, quotes, hyphens, or SQL-injection payloads raise ValueError before any SQL runs. SQLite itself also reserves object names starting with sqlite_, which it refuses regardless of this backend.

Different table_name values map to different, isolated tables in the same .db file — useful for sharing one database across stores without separate files.

python

# Two isolated tables in the same database file
from praisonaiagents.storage import SQLiteBackend

sessions = SQLiteBackend(db_path="app.db", table_name="sessions")
runs     = SQLiteBackend(db_path="app.db", table_name="run_history")

sessions.save("s1", {"user": "alice"})
runs.save("s1",     {"result": "ok"})

sessions.load("s1")  # -> {"user": "alice"}
runs.load("s1")      # -> {"result": "ok"}

Construction failure modes

When SQLiteBackend(...) raises during construction, the opened SQLite connection is closed before the error propagates, and the database file is left exactly as it was before the call.

Failure	When it raises	What propagates	Opened connection	Existing database file
`table_name` with invalid characters (space, quote, hyphen, SQL-injection payload)	Before any SQL runs	`ValueError`	Never opened	Unchanged
`table_name` with SQLite-reserved prefix (e.g. `sqlite_private`)	During `CREATE TABLE`	`sqlite3.OperationalError`	Closed	Unchanged
Pre-existing table of the same name missing the `key` column	During `CREATE INDEX`	`sqlite3.OperationalError`	Closed	Unchanged — existing table and its data survive
Parent directory cannot be created (permissions, read-only FS)	Before any connection opens	`OSError` / `PermissionError`	Never opened	Unchanged

This makes construction safe to retry — inspect or replace the .db file, fix the offending argument, and reconstruct.

python

from praisonaiagents.storage import SQLiteBackend
import sqlite3

# Pre-existing table with incompatible schema → construction fails cleanly
try:
    backend = SQLiteBackend(db_path="data.db")
except sqlite3.OperationalError:
    # Connection was closed on our behalf; the .db file is untouched
    # and immediately removable / reopenable (safe on Windows).
    ...

Diagram

Using with Training Storage

TrainingStorage is a context manager — use with to close the backend deterministically.

python

from praisonai.train.agents.storage import TrainingStorage
from praisonaiagents.storage import SQLiteBackend

# Default: file-based storage
with TrainingStorage(session_id="train-123") as storage:
    storage.save_iteration(iteration)

# With SQLite backend for better performance
with TrainingStorage(
    session_id="train-123",
    backend=SQLiteBackend(db_path="training.db"),
) as storage:
    storage.save_iteration(iteration)
    iterations = storage.load_iterations()
# Connection closed automatically — safe on Windows, no ResourceWarning

When you use a SQLiteBackend (or RedisBackend) directly — for example via list_sessions_from_backend(SQLiteBackend(db_path=db)) — close the backend yourself, or wrap the call in try / finally. TrainingStorage‘s context manager only closes backends it owns.

python

from praisonaiagents.storage import SQLiteBackend

backend = SQLiteBackend(db_path="training.db")
try:
    sessions = list_sessions_from_backend(backend)
finally:
    backend.close()

Using with Learn Stores

python

from praisonaiagents.memory.learn.stores import PersonaStore, InsightStore
from praisonaiagents.storage import SQLiteBackend

# Default: file-based
persona_store = PersonaStore(user_id="alice")

# With SQLite backend
backend = SQLiteBackend(db_path="learn.db")
persona_store = PersonaStore(user_id="alice", backend=backend)
insight_store = InsightStore(user_id="alice", backend=backend)

# Add entries
persona_store.add_preference("Prefers concise responses")
insight_store.add_insight("User is interested in AI")

Async Support

For async applications, use AsyncBaseJSONStore:

python

from praisonaiagents.storage import AsyncBaseJSONStore, SQLiteBackend

backend = SQLiteBackend(db_path="data.db")
store = AsyncBaseJSONStore("session.json", backend=backend)

# Async operations
data = await store.load_async()
await store.save_async({"key": "value"})
exists = await store.exists_async()
await store.delete_async()

Custom Backends

Implement StorageBackendProtocol for custom backends:

python

from praisonaiagents.storage import StorageBackendProtocol
from typing import Dict, Any, List, Optional

class RedisBackend:
    """Custom Redis backend example."""

    def __init__(self, url: str = "redis://localhost:6379"):
        import redis
        self.client = redis.from_url(url)

    def save(self, key: str, data: Dict[str, Any]) -> None:
        import json
        self.client.set(key, json.dumps(data))

    def load(self, key: str) -> Optional[Dict[str, Any]]:
        import json
        value = self.client.get(key)
        return json.loads(value) if value else None

    def delete(self, key: str) -> bool:
        return self.client.delete(key) > 0

    def list_keys(self, prefix: str = "") -> List[str]:
        pattern = f"{prefix}*" if prefix else "*"
        return [k.decode() for k in self.client.keys(pattern)]

    def exists(self, key: str) -> bool:
        return self.client.exists(key) > 0

# Use custom backend
backend = RedisBackend("redis://localhost:6379")
store = BaseJSONStore("session.json", backend=backend)

Protocol Reference

python

from typing import Protocol, Dict, Any, List

class StorageBackendProtocol(Protocol):
    """Protocol for pluggable storage backends."""

    def save(self, key: str, data: Dict[str, Any]) -> None:
        """Save data with the given key."""
        ...

    def load(self, key: str) -> Any:
        """Load data by key. Returns None if not found."""
        ...

    def delete(self, key: str) -> bool:
        """Delete data by key. Returns True if deleted."""
        ...

    def list_keys(self, prefix: str = "") -> List[str]:
        """List all keys, optionally filtered by prefix."""
        ...

    def exists(self, key: str) -> bool:
        """Check if a key exists."""
        ...

RedisBackend

Redis-based storage for high-speed caching and ephemeral data.

python

from praisonaiagents.storage import RedisBackend

backend = RedisBackend(
    url="redis://localhost:6379",  # Redis connection URL
    prefix="praison:",              # Key prefix
    ttl=3600,                       # Optional TTL in seconds
    db=0                            # Redis database number
)

# CRUD operations
backend.save("session_123", {"user": "alice", "messages": []})
data = backend.load("session_123")
backend.delete("session_123")
keys = backend.list_keys(prefix="session_")
exists = backend.exists("session_123")

# Set TTL on specific key
backend.set_ttl("session_123", 1800)  # 30 minutes

# Clear all data with prefix
backend.clear()

# Close connection
backend.close()

Features:

Sub-millisecond latency
Built-in TTL support
Automatic key prefixing
Requires redis package: pip install redis

Using with RunHistory (Recipe History)

Store recipe run history with pluggable backends:

python

from praisonai.recipe.history import RunHistory
from praisonaiagents.storage import SQLiteBackend, RedisBackend

# Default: file-based storage in ~/.praisonai/runs/
history = RunHistory()

# With SQLite backend
backend = SQLiteBackend(db_path="~/.praisonai/runs.db")
history = RunHistory(backend=backend)

# With Redis backend (for distributed systems)
backend = RedisBackend(url="redis://localhost:6379", prefix="runs:")
history = RunHistory(backend=backend)

# Store and retrieve runs
run_id = history.store(result, input_data={"query": "test"})
run_data = history.get(run_id)
history.delete(run_id)

Using with SessionManager (Session State)

Persist CLI session state with different backends:

python

from praisonai.cli.state.sessions import SessionManager
from praisonaiagents.storage import SQLiteBackend, RedisBackend

# Default: file-based in ~/.praisonai/sessions/
manager = SessionManager()

# With SQLite backend
backend = SQLiteBackend(db_path="~/.praisonai/sessions.db")
manager = SessionManager(backend=backend)

# With Redis backend (for shared sessions)
backend = RedisBackend(url="redis://localhost:6379", prefix="session:")
manager = SessionManager(backend=backend)

# Session operations
session = manager.create(run_context)
manager.append_event(session.session_id, {"type": "message", "content": "Hello"})
events = manager.get_events(session.session_id)
sessions = manager.list(limit=10)

Using with MCPToolIndex (Tool Index)

Store MCP tool schemas with pluggable backends:

python

from praisonai.mcp_server.tool_index import MCPToolIndex
from praisonaiagents.storage import SQLiteBackend, RedisBackend

# Default: file-based in ~/.praisonai/mcp/
index = MCPToolIndex()

# With SQLite backend
backend = SQLiteBackend(db_path="~/.praisonai/mcp.db")
index = MCPToolIndex(backend=backend)

# With Redis backend (for shared tool registry)
backend = RedisBackend(url="redis://localhost:6379", prefix="mcp:")
index = MCPToolIndex(backend=backend)

# Tool operations
index.sync("brave-search", tools=[...])
tools = index.list_tools("brave-search")
status = index.get_status("brave-search")

Choosing a Backend

Use Case	Recommended Backend	Why
Development	FileBackend	Easy debugging, human-readable
Single-user app	SQLiteBackend	Better performance, single file
Multi-process	SQLiteBackend	Handles concurrent access
Production web	PostgreSQL adapter	Scalable, reliable
High-speed cache	RedisBackend	Sub-ms latency, TTL support
Serverless	SQLiteBackend or cloud DB	No server management
Session caching	RedisBackend	Fast, ephemeral data
Distributed systems	RedisBackend	Shared state across nodes
Recipe history	SQLiteBackend	Reliable, queryable
MCP tool registry	SQLiteBackend or RedisBackend	Fast lookups

Backend Comparison

Feature	FileBackend	SQLiteBackend	RedisBackend
Dependencies	None	None	`redis` package
Concurrency	File locks	ACID transactions	Native
Performance	Good	Better	Best
Atomic writes	✅ (temp + rename)	✅ (transactions)	✅ (single-op)
TTL Support	❌	❌	✅
Distributed	❌	❌	✅
Human-readable	✅	❌	❌
Single file	❌	✅	N/A

Database Overview

All supported databases

Memory Storage

Agent memory persistence

Training

Agent training with storage

Session Resume

Continue conversations

Storage Backends