DatabaseMixin

​

Overview

• No extra dependencies. Uses only sqlite3 from the standard library.
• Dict-oriented API. Every query returns dicts (via sqlite3.Row under the hood) so tools can serialize results without transformation.
• LLM-friendly SQL. Explicit parameterised SQL with :name placeholders and a consistent (table, data, where, params) signature for CRUD operations.
• Safe concurrency for agents. Single connection with check_same_thread=False; use transaction() for atomic writes.

​

API Reference

​

Connection lifecycle

def init_db(self, path: str = ":memory:") -> None
def close_db(self) -> None
@property
def db_ready(self) -> bool

• init_db(path) — open (or re-open) a SQLite connection. Accepts :memory: or a filesystem path; parent directories are auto-created. Enables foreign keys (PRAGMA foreign_keys = ON) and uses sqlite3.Row for dict-style access.
• close_db() — close the connection. Safe to call multiple times; safe to call before init_db.
• db_ready — True if a connection is open.

​

Reads

def query(
    self,
    sql: str,
    params: Optional[Dict[str, Any]] = None,
    one: bool = False,
) -> Union[List[Dict[str, Any]], Optional[Dict[str, Any]]]

rows = self.query("SELECT * FROM users")
user = self.query("SELECT * FROM users WHERE id = :id", {"id": 42}, one=True)

def query_readonly(
    self,
    sql: str,
    params: Optional[Dict[str, Any]] = None,
    one: bool = False,
) -> Union[List[Dict[str, Any]], Optional[Dict[str, Any]]]

​

Writes

def insert(self, table: str, data: Dict[str, Any]) -> int
def update(self, table: str, data: Dict[str, Any], where: str,
           params: Dict[str, Any]) -> int
def delete(self, table: str, where: str, params: Dict[str, Any]) -> int

• insert(table, data) — builds INSERT INTO table (...) VALUES (...) from the data dict and returns the new row’s lastrowid.
• update(table, data, where, params) — the implementation prefixes the data keys with __set_ internally to keep them distinct from the params used in the WHERE clause; a params key that collides with that prefix raises ValueError rather than silently overriding the column value. Returns the affected row count.
• delete(table, where, params) — returns the deleted row count.

user_id = self.insert("users", {"name": "Alice", "email": "[email protected]"})

count = self.update(
    "users",
    {"email": "[email protected]"},
    "id = :id",
    {"id": user_id},
)

self.delete("sessions", "expires_at < :now", {"now": now_ts})

​

Transactions

@contextmanager
def transaction(self): ...

with self.transaction():
    uid = self.insert("users", {"name": "Alice"})
    self.insert("profiles", {"user_id": uid, "bio": "Hello"})

​

Raw SQL / schema

def execute(self, sql: str) -> None
def table_exists(self, name: str) -> bool

• execute(sql) — runs arbitrary SQL via executescript(). Use for CREATE TABLE/CREATE INDEX bootstrap. Auto-commits any pending work.
• table_exists(name) — returns True if a table of that name exists.

if not self.table_exists("items"):
    self.execute('''
        CREATE TABLE items (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL
        )
    ''')

​

Example agent

from gaia.agents.base.agent import Agent
from gaia.agents.base.tools import tool
from gaia.database.mixin import DatabaseMixin


class TodoAgent(Agent, DatabaseMixin):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.init_db("data/todo.db")

        if not self.table_exists("todos"):
            self.execute(
                """
                CREATE TABLE todos (
                    id INTEGER PRIMARY KEY,
                    title TEXT NOT NULL,
                    done INTEGER NOT NULL DEFAULT 0
                )
                """
            )

    def _register_tools(self):
        super()._register_tools()

        @tool
        def add_todo(title: str) -> dict:
            """Add a new todo item."""
            todo_id = self.insert("todos", {"title": title})
            return {"id": todo_id, "title": title}

        @tool
        def list_todos() -> dict:
            """List all todos."""
            return {"todos": self.query("SELECT * FROM todos ORDER BY id")}

        @tool
        def complete_todo(todo_id: int) -> dict:
            """Mark a todo as done."""
            updated = self.update(
                "todos", {"done": 1}, "id = :id", {"id": todo_id}
            )
            return {"updated": updated}

​

Caveats

• SQLite only — there is no PostgreSQL/MySQL support. If you need a different backend, you’ll need a different mixin.
• Single connection per agent instance; the mixin assumes single-process ownership. For multi-process scenarios, either use a shared DB file with WAL-mode via execute("PRAGMA journal_mode=WAL") or route writes through a single coordinator agent.
• There is no ORM layer. You work directly with SQL and dict results, which is intentional — it keeps the surface small and LLM-friendly.
• Identifiers must be bare — see the validation contract under Writes. Table or column names that need quoting are not supported.
• query_readonly()’s authorizer is connection-scoped, not call-scoped. Concurrent and nested read-only calls are reference-counted, so one finishing never disarms another still in flight. But the window applies to the whole connection: because init_db() opens it with check_same_thread=False, a writer on another thread sharing that connection is denied for as long as any read-only call is open. Don’t share one mixin instance between a read-only caller and a background writer.

​

Related

• EMR Agent uses this mixin as its persistence layer.
• FileWatcher is commonly paired with DatabaseMixin to record ingestion events.