Daita Logo

SQLite Plugin

Async SQLite database operations backed by `aiosqlite`, with full agent tool integration and Focus DSL support.

#Installation

bash
pip install 'daita-agents[sqlite]'

#Quick Start

python
from daita import Agent
from daita.plugins import sqlite
 
db = sqlite(path="mydata.db")
 
agent = Agent(
    name="Data Agent",
    prompt="You are a data analyst. Help users query and analyze data.",
    tools=[db]
)
 
await agent.start()
result = await agent.run("Show me all orders placed this month")

#Connection Parameters

python
sqlite(
    path: str = ":memory:",
    wal_mode: bool = True,
    timeout: float = 5.0,
)

#Parameters

  • path (str): Path to the SQLite database file, or ":memory:" for an in-process database discarded on close. Default: ":memory:"
  • wal_mode (bool): Enable WAL journal mode for better concurrent read performance. Default: True
  • timeout (float): Seconds to wait when the database is locked before raising an error. Default: 5.0

#Connection Methods

python
from daita.plugins import sqlite
 
# File-based database
async with sqlite(path="analytics.db") as db:
    results = await db.query("SELECT * FROM events")
 
# In-memory database (testing / ephemeral workloads)
async with sqlite() as db:
    await db.execute("CREATE TABLE t (id INTEGER, val TEXT)")
    await db.execute("INSERT INTO t VALUES (1, 'hello')")
    rows = await db.query("SELECT * FROM t")

#Query Operations

python
from daita.plugins import sqlite
 
async with sqlite(path="app.db") as db:
    # Simple SELECT
    users = await db.query("SELECT id, name, email FROM users")
    # Returns: [{"id": 1, "name": "Alice", "email": "alice@example.com"}, ...]
 
    # Parameterized query (use ?)
    active = await db.query(
        "SELECT * FROM users WHERE status = ? AND created_at > ?",
        ["active", "2024-01-01"]
    )

Parameter syntax: SQLite uses ? placeholders.

#Data Modification

python
from daita.plugins import sqlite
 
async with sqlite(path="app.db") as db:
    # Insert
    await db.execute(
        "INSERT INTO users (name, email) VALUES (?, ?)",
        ["Jane", "jane@example.com"]
    )
 
    # Update
    await db.execute(
        "UPDATE users SET last_login = datetime('now') WHERE id = ?",
        [user_id]
    )
 
    # Delete
    await db.execute("DELETE FROM users WHERE status = ?", ["inactive"])

#Bulk Operations & Schema

python
from daita.plugins import sqlite
 
async with sqlite(path="app.db") as db:
    # Bulk insert
    rows = [
        {"name": "Alice", "email": "alice@example.com"},
        {"name": "Bob", "email": "bob@example.com"},
    ]
    count = await db.insert_many("users", rows)
 
    # Run multiple statements (DDL, migrations)
    await db.execute_script("""
        CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, total REAL);
        CREATE INDEX IF NOT EXISTS idx_total ON orders (total);
    """)
 
    # List tables
    tables = await db.tables()
 
    # Describe a table
    schema = await db.describe("users")
 
    # Read / set a PRAGMA
    page_size = await db.pragma("page_size")
    await db.pragma("journal_mode", "WAL")

#Data Quality Assertions

Use query_checked() to enforce row-level rules at query time:

python
from daita.plugins import sqlite
from daita import ItemAssertion
 
async with sqlite(path="transactions.db") as db:
    rows = await db.query_checked(
        "SELECT * FROM transactions",
        assertions=[
            ItemAssertion(lambda r: r["amount"] > 0, "All amounts must be positive"),
            ItemAssertion(lambda r: r["customer_id"] is not None, "customer_id required"),
        ],
    )

Raises DataQualityError (a permanent error — not retried) if any assertion fails.

#Using with Agents

#Tool-Based Integration (Recommended)

python
from daita import Agent
from daita.plugins import sqlite
import os
 
db = sqlite(path=os.getenv("SQLITE_PATH", "app.db"))
 
agent = Agent(
    name="Data Analyst",
    prompt="You are a data analyst. Help users explore and query the database.",
    tools=[db]
)
 
await agent.start()
result = await agent.run("What are the top 5 products by revenue?")
await agent.stop()

#Available Tools

ToolDescriptionParameters
sqlite_queryExecute a SELECT querysql (required), params (optional), focus (optional DSL)
sqlite_executeRun INSERT / UPDATE / DELETEsql (required), params (optional)
sqlite_list_tablesList all tablesNone
sqlite_get_schemaGet column info for a tabletable_name (required)
sqlite_inspectList all tables and their column schemastables (optional array to filter by name)

#Focus DSL

The sqlite_query tool supports Focus DSL pushdown — clauses are compiled into SQL before execution, so the database does the filtering:

python
agent = Agent(
    name="Analyst",
    tools=[sqlite(path="sales.db")],
    focus={"sqlite_query": "status == 'active' | ORDER BY created_at DESC | LIMIT 20"}
)

#Error Handling

python
from daita.plugins import sqlite
from daita import DataQualityError
 
try:
    async with sqlite(path="app.db") as db:
        results = await db.query("SELECT * FROM users")
except ImportError:
    print("Install aiosqlite: pip install 'daita-agents[sqlite]'")
except DataQualityError as e:
    print(f"Quality violations: {e.violations}")

#Best Practices

  • Use ":memory:" for tests and ephemeral workloads — no cleanup needed
  • Keep wal_mode=True (default) for any database accessed concurrently
  • Use execute_script() for multi-statement migrations
  • Use insert_many() for bulk inserts rather than looping execute()
  • Store credentials and paths in environment variables

#Next Steps