Lite Sqlite

Lite SQLite - Lightweight Local Database

Ultra-lightweight SQLite database management optimized for OpenClaw agents with minimal RAM (~2-5MB) and storage overhead.

Why SQLite?

✅ Zero setup - No server, no configuration, file-based ✅ Minimal RAM - 2-5MB typical usage ✅ Fast - Millions of queries/second ✅ Portable - Single .db file ✅ Reliable - ACID compliant, crash-proof ✅ Cross-platform - Works everywhere Python works

Core Features

Quick Start

Basic Database Operations

from sqlite_connector import SQLiteDB
Create database (auto-wal mode enabled)
db = SQLiteDB("agent_data.db")
Create table
db.create_table("memos", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "title": "TEXT NOT NULL",
    "content": "TEXT",
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP",
    "tags": "TEXT"
})
Insert data
db.insert("memos", [title="First memo", content="Hello world", tags="test"])
Query data
results = db.query("SELECT * FROM memos WHERE tags = ?", ("test",))
Update data
db.update("memos", "id = ?", [content="Updated content"], (1,))
Delete data
db.delete("memos", "id = ?", (1,))
Close connection
db.close()

In-Memory Database (Fastest)

# Fastest mode - RAM only, no disk I/O
db = SQLiteDB(":memory:")
Perfect for temporary operations
db.create_table("temp", {...})
Data persists only during session
Use for caching, computations, temporary storage

Performance Optimization

Essential Settings

import sqlite3
WAL mode (Write-Ahead Logging) - 3-4x faster
conn = sqlite3.connect("agent_data.db")
conn.execute("PRAGMA journal_mode=WAL")
Sync OFF (faster writes, crash-safe with proper shutdown)
conn.execute("PRAGMA synchronous=NORMAL")
Memory optimization
conn.execute("PRAGMA cache_size=-64000")  # 64MB cache
conn.execute("PRAGMA page_size=4096")
Temp store in RAM
conn.execute("PRAGMA temp_store=MEMORY")

Query Optimization

# Use indexes for frequent queries
db.create_index("memos", "tags")
db.create_index("memos", "created_at")
Use prepared statements (automatic in our wrapper)
db.query("SELECT * FROM memos WHERE id = ?", (id,))
Batch inserts for large datasets
db.batch_insert("memos", rows_data)

Predefined Schemas

Agent Memo Schema (Memory Store)

db.create_table("agent_memos", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "agent_id": "TEXT NOT NULL",           # Which agent created it
    "key": "TEXT NOT NULL",               # Lookup key
    "value": "TEXT",                      # Stored value
    "priority": "INTEGER DEFAULT 0",       # For retrieval ordering
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP",
    "expires_at": "TEXT"                  # Optional TTL
})
Create indexes
db.create_index("agent_memos", "agent_id")
db.create_index("agent_memos", "key")
db.create_index("agent_memos", "expires_at")

Session Log Schema

db.create_table("session_logs", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "session_id": "TEXT NOT NULL",
    "agent": "TEXT NOT NULL",
    "message": "TEXT",
    "metadata": "TEXT",                   # JSON
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP"
})
db.create_index("session_logs", "session_id")
db.create_index("session_logs", "created_at")

Cache Schema (TTL-based)

db.create_table("cache", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "key": "TEXT UNIQUE NOT NULL",
    "value": "BLOB",                      # Supports binary data
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP",
    "expires_at": "TEXT NOT NULL"
})
Auto-cleanup expired entries
db.query("DELETE FROM cache WHERE expires_at < ?", (datetime.now().isoformat(),))
db.create_index("cache", "key")
db.create_index("cache", "expires_at")

Advanced Features

Connection Pooling

from sqlite_connector import ConnectionPool
Pool of connections for concurrent access
pool = ConnectionPool("agent_data.db", max_connections=5)
Get connection
conn = pool.get_connection()
Use conn...
pool.release_connection(conn)

Automatic Backup

# Backup database
db.backup("agent_data_backup.db")
Automatic daily backup
db.auto_backup("backups/", "daily")

Schema Migration

# Add column if not exists
db.add_column("memos", "updated_at", "TEXT DEFAULT CURRENT_TIMESTAMP")
Migrate data
db.migrate("memos", {
    "old_column": "new_column"
})

Performance Benchmarks

Typical Performance

| Operation | Rows | Time (In-Memory) | Time (Disk) | |-----------|------|------------------|-------------| | Insert | 10,000 | 0.05s | 0.3s | | Select (indexed) | 10,000 | 0.001s | 0.01s | | Select (full scan) | 10,000 | 0.05s | 0.5s | | Update | 1,000 | 0.01s | 0.1s | | Delete | 1,000 | 0.01s | 0.1s |

Memory Usage

Best Practices for OpenClaw Agents

1. Choose the Right Mode

# Use :memory: for temporary operations
temp_db = SQLiteDB(":memory:")
Use file DB for persistent storage
persist_db = SQLiteDB("agent_storage.db")

2. Use Proper Indexes

# Always index columns used in WHERE clauses
db.create_index("table", "column_name")
Index multiple columns for composite queries
db.create_index("table", "col1, col2")

3. Batch Operations

# Instead of individual inserts:
for row in rows:
    db.insert("table", row)  # Slow!
Use batch insert:
db.batch_insert("table", rows)  # Fast!

4. Use TTL for Expiring Data

# Auto-cleanup old data
db.cleanup_expired("cache", "expires_at")
db.cleanup_old("logs", "created_at", days=7)

5. Compact Database Periodically

# Reclaim space after many deletes
db.vacuum()  # Should be run during downtime

DuckDB Alternative (Analytics)

For analytical queries (aggregations, joins on large datasets), consider DuckDB:

import duckdb
conn = duckdb.connect(":memory:")
Faster than SQLite for complex analytics
conn.execute("""
    SELECT COUNT(*) as rows,
           AVG(value) as avg_value
    FROM large_table
""").fetchall()

When to use DuckDB:

When to use SQLite:

Common Patterns

1. Memo Storage

def save_memo(db, agent_id, key, value, ttl_hours=24):
    expires_at = (datetime.now() + timedelta(hours=ttl_hours)).isoformat()
    db.insert("agent_memos", {
        "agent_id": agent_id,
        "key": key,
        "value": json.dumps(value),
        "expires_at": expires_at
    })

2. Session Persistence

def save_session(db, session_id, agent, message, metadata=None):
    db.insert("session_logs", {
        "session_id": session_id,
        "agent": agent,
        "message": message,
        "metadata": json.dumps(metadata) if metadata else None
    })

3. Caching Layer

def cache_get(db, key):
    if expired_key := db.query_one(
        "SELECT value FROM cache WHERE key = ? AND expires_at > ?",
        (key, datetime.now().isoformat())
    ):
        return json.loads(expired_key)
    return None
def cache_set(db, key, value, ttl_seconds=3600):
    expires_at = (datetime.now() + timedelta(seconds=ttl_seconds)).isoformat()
    db.insert_or_replace("cache", {
        "key": key,
        "value": json.dumps(value),
        "expires_at": expires_at
    })

Error Handling

try:
    db.insert("metrics", {...})
except sqlite3.IntegrityError:
    # Duplicate key violation
    pass
except sqlite3.OperationalError:
    # Table doesn't exist or database locked
    pass

Size Optimization Tips

Reduce Storage

1. Use appropriate data types: - INTEGER instead of TEXT for numbers - REAL instead of TEXT for floats - Use CHECK constraints for validation

2. Normalize data: - Store JSON as TEXT - Use TEXT for variable-length strings - Avoid storing redundant data

3. Vacuum regularly:

   db.vacuum()  # Reclaims space after deletes
   

4. Use WAL instead of journal:

   conn.execute("PRAGMA journal_mode=WAL")
   

Migration from Other Stores

From JSON Files

# Load JSON into SQLite
import json
with open("data.json") as f:
    data = json.load(f)
db.create_table("json_data", {key: "TEXT" for key in data[0].keys()})
db.batch_insert("json_data", data)

From CSV Files

import pandas as pd
df = pd.read_csv("data.csv")
df.to_sql("csv_data", conn, if_exists="replace", index=False)

Troubleshooting

Database Locked Error

# Use WAL mode for concurrent access
conn.execute("PRAGMA journal_mode=WAL")
Or use connection pool
pool = ConnectionPool("db.db", timeout=5.0)

Slow Queries

# Check query plan
plan = conn.execute("EXPLAIN QUERY PLAN SELECT * FROM ...").fetchall()
Add indexes
db.create_index("table", "column")
Use ANALYZE
conn.execute("ANALYZE")

Large Database Size

# Check size info
size_info = conn.execute("PRAGMA page_count, page_size").fetchone()
print(f"Size: {(page_count * page_size) / (1024*1024):.2f} MB")
Vacuum to reclaim space
db.vacuum()

CLI Tool

The bundled sqlite_cli.py provides command-line access:

# Create database
python scripts/sqlite_cli.py create agent_data.db
Add table
python scripts/sqlite_cli.py create-table agent_memos -c id:INTEGER:P -c title:TEXT -c content:TEXT
Insert data
python scripts/sqlite_cli.py insert agent_memos '{"title": "Test", "content": "Hello"}'
Query data
python scripts/sqlite_cli.py query "SELECT * FROM agent_memos"
Optimize
python scripts/sqlite_cli.py optimize agent_data.db

Resources

Basic Database Operations

from sqlite_connector import SQLiteDB
Create database (auto-wal mode enabled)
db = SQLiteDB("agent_data.db")
Create table
db.create_table("memos", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "title": "TEXT NOT NULL",
    "content": "TEXT",
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP",
    "tags": "TEXT"
})
Insert data
db.insert("memos", [title="First memo", content="Hello world", tags="test"])
Query data
results = db.query("SELECT * FROM memos WHERE tags = ?", ("test",))
Update data
db.update("memos", "id = ?", [content="Updated content"], (1,))
Delete data
db.delete("memos", "id = ?", (1,))
Close connection
db.close()

In-Memory Database (Fastest)

# Fastest mode - RAM only, no disk I/O
db = SQLiteDB(":memory:")
Perfect for temporary operations
db.create_table("temp", {...})
Data persists only during session
Use for caching, computations, temporary storage

Database Locked Error

# Use WAL mode for concurrent access
conn.execute("PRAGMA journal_mode=WAL")
Or use connection pool
pool = ConnectionPool("db.db", timeout=5.0)

Slow Queries

# Check query plan
plan = conn.execute("EXPLAIN QUERY PLAN SELECT * FROM ...").fetchall()
Add indexes
db.create_index("table", "column")
Use ANALYZE
conn.execute("ANALYZE")

Large Database Size

# Check size info
size_info = conn.execute("PRAGMA page_count, page_size").fetchone()
print(f"Size: {(page_count * page_size) / (1024*1024):.2f} MB")
Vacuum to reclaim space
db.vacuum()