> ## Documentation Index
> Fetch the complete documentation index at: https://amd-gaia.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# DatabaseMixin
**Source Code:** [`src/gaia/database/mixin.py`](https://github.com/amd/gaia/blob/main/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
```python theme={null}
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
```python theme={null}
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).
```python theme={null}
rows = self.query("SELECT * FROM users")
user = self.query("SELECT * FROM users WHERE id = :id", {"id": 42}, one=True)
```
### Writes
```python theme={null}
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.
```python theme={null}
user_id = self.insert("users", {"name": "Alice", "email": "alice@example.com"})
count = self.update(
"users",
{"email": "new@example.com"},
"id = :id",
{"id": user_id},
)
self.delete("sessions", "expires_at < :now", {"now": now_ts})
```
### Transactions
```python theme={null}
@contextmanager
def transaction(self): ...
```
Atomic block: commits on success, rolls back on exception, never nested.
```python theme={null}
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
```python theme={null}
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.
```python theme={null}
if not self.table_exists("items"):
self.execute('''
CREATE TABLE items (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL
)
''')
```
***
## Example agent
```python theme={null}
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.
## Related
* [EMR Agent](/docs/playbooks/emr-agent/part-1-getting-started) uses this mixin as
its persistence layer.
* [FileWatcher](/docs/sdk/utils/file-watcher) is commonly paired with
`DatabaseMixin` to record ingestion events.