Modules - Storage

Each module gets its own SQLite database file, isolated from other modules. Storage is accessed through ctx.storage. The API is async with a single lazy connection per module. Each operation auto-commits independently; use transaction() when you need atomicity across multiple operations. Queries are written in standard SQL.

Schema Setup

Tables should be created in an @on_setup function. Use CREATE TABLE IF NOT EXISTS so the module is safe to restart:

from owlbot.api import ModuleContext, on_setup

@on_setup
async def setup(ctx: ModuleContext) -> None:
    await ctx.storage.execute("""
        CREATE TABLE IF NOT EXISTS scores (
            user_id TEXT PRIMARY KEY,
            display_name TEXT NOT NULL,
            score INTEGER NOT NULL DEFAULT 0
        )
    """)

The database file lives at <storage_dir>/<module_name>.db (default: data/<module_name>.db). The directory is created automatically if it doesn't exist.

Query Methods

All methods are available on ctx.storage:

Method	Returns	Description
`execute(sql, parameters=())`	`aiosqlite.Cursor`	Run any SQL statement.
`execute_many(sql, parameters)`	`aiosqlite.Cursor`	Run the same statement with multiple parameter sets.
`fetch_one(sql, parameters=())`	`aiosqlite.Row \| None`	Return a single row.
`fetch_all(sql, parameters=())`	`list[aiosqlite.Row]`	Return all matching rows.
`fetch_value(sql, parameters=())`	`Any \| None`	Return the first column of the first row.

Writing

execute() runs any SQL statement and returns a cursor with lastrowid and rowcount:

cursor = await ctx.storage.execute(
    "INSERT INTO scores (user_id, display_name, score) VALUES (?, ?, ?)",
    (user.id, user.display_name, 100),
)
new_id = cursor.lastrowid

execute_many() is the batch equivalent, running the same statement across multiple parameter sets:

await ctx.storage.execute_many(
    "INSERT OR IGNORE INTO scores (user_id, display_name, score) VALUES (?, ?, ?)",
    [(u.id, u.display_name, 0) for u in users],
)

Reading

fetch_one() returns a single row, or None if there are no results:

row = await ctx.storage.fetch_one(
    "SELECT score FROM scores WHERE user_id = ?", (user.id,)
)
if row:
    score = row["score"]

fetch_all() returns all matching rows as a list:

rows = await ctx.storage.fetch_all(
    "SELECT display_name, score FROM scores ORDER BY score DESC LIMIT 10"
)
for row in rows:
    print(f"{row['display_name']}: {row['score']}")

fetch_value() returns the first column of the first row, useful for aggregates like COUNT(*). Returns None if there are no results:

count = await ctx.storage.fetch_value(
    "SELECT COUNT(*) FROM scores WHERE score > ?", (1000,)
)

Row Access

Rows are returned as aiosqlite.Row objects, which support both dict-style and index access:

row = await ctx.storage.fetch_one("SELECT user_id, score FROM scores LIMIT 1")

# Dict-style:
user_id = row["user_id"]
score = row["score"]

# Index-style:
user_id = row[0]
score = row[1]

Parameterized Queries

Use ? placeholders with tuple parameters, or :name with dict parameters:

# Positional (tuple):
await ctx.storage.execute(
    "UPDATE scores SET score = ? WHERE user_id = ?",
    (new_score, user_id),
)

# Named (dict):
await ctx.storage.execute(
    "UPDATE scores SET score = :score WHERE user_id = :uid",
    {"score": new_score, "uid": user_id},
)

Never use f-strings or string formatting for SQL parameters. This prevents SQL injection.

Transactions

Each storage operation auto-commits independently. A call to execute(), fetch_one(), or any other query method commits on success or rolls back on failure. No transaction state is held between calls.

When multiple operations must succeed or fail together, use the transaction() context manager:

async with ctx.storage.transaction():
    await ctx.storage.execute("UPDATE balances SET amount = amount - ?", (cost,))
    await ctx.storage.execute("INSERT INTO purchases (item) VALUES (?)", (item,))
    # Both committed together on success, or both rolled back if either fails

Outside of a transaction() block, each operation is its own atomic unit. A handler that calls execute() three times makes three independent commits.

Error Handling

Storage operations raise StorageError on failure:

from owlbot.api import StorageError

try:
    await ctx.storage.execute("INSERT INTO scores VALUES (?, ?, ?)", (id, name, 0))
except StorageError as e:
    ctx.logger.error(f"Failed to insert score: {e}")

StorageError wraps the underlying aiosqlite.Error so the message is readable without needing to catch the low-level exception.

Built-in Modules

Creating New Modules