Skip to main content
Source Code: src/gaia/database/mixin.py
Component: DatabaseMixin Module: gaia.database.mixin Import: from gaia.database.mixin import DatabaseMixin

​
Overview

DatabaseMixin is a zero-dependency SQLite mixin for agents that need persistent, queryable state (user intake records, chat sessions, job queues, etc.). It wraps Python’s stdlib sqlite3 with a small ergonomic API and an optional transaction context manager. Design goals:
  • 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

All methods are instance methods on any class that mixes in DatabaseMixin.

​
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.
All query/mutation methods raise RuntimeError("Database not initialized. Call init_db() first.") if the connection isn’t ready.

​
Reads

def query(
    self,
    sql: str,
    params: Optional[Dict[str, Any]] = None,
    one: bool = False,
) -> Union[List[Dict[str, Any]], Optional[Dict[str, Any]]]
Execute a SELECT. Uses :name placeholders. When one=True, returns either a single row dict or None; otherwise returns a list of row dicts (possibly empty). 
rows = self.query("SELECT * FROM users")
user = self.query("SELECT * FROM users WHERE id = :id", {"id": 42}, one=True)

​
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 so they can’t collide with the params used in the WHERE clause. Returns the affected row count.
  • delete(table, where, params) — returns the deleted row count.
Every mutation auto-commits unless the call is inside a transaction() block. 
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): ...
Atomic block: commits on success, rolls back on exception, never nested. 
with self.transaction():
    uid = self.insert("users", {"name": "Alice"})
    self.insert("profiles", {"user_id": uid, "bio": "Hello"})
execute() (raw DDL below) cannot be called inside a transaction — it will raise RuntimeError.

​
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.
  • EMR Agent uses this mixin as its persistence layer.
  • FileWatcher is commonly paired with DatabaseMixin to record ingestion events.