SQLite Database Backend

This module provides the SQLite implementation of the BaseDatabase interface. It is a lightweight, file-based database backend that is suitable for local development and testing.

Overview

The SQLiteDatabase class can be used to create a database connection to a local SQLite file.

Typical usage:

import asyncio
from agentstr.database.sqlite import SQLiteDatabase

# Create a database instance for an in-memory database
db = SQLiteDatabase(conn_str="sqlite://:memory:")

async def main():
    await db.async_init()
    print("Connection to SQLite successful.")
    # ... perform database operations
    await db.close()
    print("Connection closed.")

# To run this, you would typically use:
# if __name__ == "__main__":
#     asyncio.run(main())

Reference

class agentstr.database.sqlite.SQLiteDatabase(conn_str: str | None = None, *, agent_name: str | None = None)[source]

Bases: BaseDatabase

SQLite implementation using aiosqlite.

__init__(conn_str: str | None = None, *, agent_name: str | None = None)[source]
async async_init() → Self[source]

Perform any asynchronous initialisation required for the backend.

async close() → None[source]

Close the underlying connection (if any).

async get_user(user_id: str) → User[source]

Fetch a User by user_id. Non-existent users yield a default model with a zero balance.

async get_current_thread_id(user_id: str) → str | None[source]

Return the current thread id for user_id within this agent scope.

async set_current_thread_id(user_id: str, thread_id: str | None) → None[source]

Persist thread_id as the current thread for user_id.

async upsert_user(user: User) → None[source]

Create or update user in storage atomically.

async add_message(thread_id: str, user_id: str, role: Literal['user', 'agent', 'tool'], message: str = '', content: str = '', kind: str = 'request', satoshis: int | None = None, extra_inputs: dict[str, Any] = {}, extra_outputs: dict[str, Any] = {}) → Message[source]

Append a message to a thread and return the stored model.

async get_messages(thread_id: str, user_id: str, *, limit: int | None = None, before_idx: int | None = None, after_idx: int | None = None, reverse: bool = False) → List[Message][source]

Retrieve messages for thread_id with optional pagination.

See Also