Storage

class afterimage.storage.BaseStorage(*args, **kwargs)

Bases: Protocol

Protocol defining the interface for storage implementations.

abstractmethod async asave_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

abstractmethod async asave_documents(documents: List[Document]) → None

abstractmethod load_conversations(limit: int | None = None, offset: int | None = None) → List[ConversationWithContext]

abstractmethod load_documents(limit: int | None = None, offset: int | None = None) → List[Document]

abstractmethod save_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

abstractmethod save_documents(documents: List[Document]) → None

class afterimage.storage.JSONLStorage(conversations_path: str | Path | None = None, documents_path: str | Path | None = None, encoding: str = 'utf-8', lock_timeout: int = 30)

Bases: BaseStorage

Stores conversations and documents in JSONL format.

When conversations_path / documents_path are omitted, filenames default to {prefix}_{timestamp}.jsonl in the process working directory. If the environment variable AFTERIMAGE_JSONL_DIR is set, those defaults are created under that directory instead (the directory is created if missing). The pytest suite sets this so runs do not litter the repository root.

async asave_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

async asave_documents(documents: List[Document]) → None

load_conversations(limit: int | None = None, offset: int | None = None) → List[EvaluatedConversationWithContext]

Load conversations from JSONL file.

Parameters:

• limit – Maximum number of conversations to load

• offset – Number of conversations to skip

Returns:

List of conversations

load_documents(limit: int | None = None, offset: int | None = None) → List[Document]

Load documents from JSONL file.

Parameters:

• limit – Maximum number of documents to load

• offset – Number of documents to skip

Returns:

List of documents

save_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

save_documents(documents: List[Document]) → None

class afterimage.storage.SQLStorage(url: str, conversations_table_name: str = 'conversations', documents_table_name: str = 'documents', metadata_fields: List[str] | None = None, batch_size: int = 100)

Bases: BaseStorage

Stores conversations and documents using SQLAlchemy.

async asave_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

Save conversations to database asynchronously.

async asave_documents(documents: List[Document]) → None

load_conversations(limit: int | None = None, offset: int | None = None, filters: Dict[str, Any] | None = None, order_by: List[tuple] | None = None) → List[EvaluatedConversationWithContext]

Load conversations from database with filtering and sorting.

Parameters:

• limit – Maximum number of conversations to load

• offset – Number of conversations to skip

• filters – Dict of field-value pairs for filtering

• order_by – List of (field, direction) tuples for sorting

Returns:

List of conversations

load_documents(limit: int | None = None, offset: int | None = None, filters: Dict[str, Any] | None = None, order_by: List[tuple] | None = None) → List[Document]

Load documents from database with filtering and sorting.

Parameters:

• limit – Maximum number of documents to load

• offset – Number of documents to skip

• filters – Dict of field-value pairs for filtering

• order_by – List of (field, direction) tuples for sorting

Returns:

List of documents

save_conversations(conversations: List[ConversationWithContext | EvaluatedConversationWithContext]) → None

Save conversations to database.

Parameters:

conversations – List of conversations to save

save_documents(documents: List[Document]) → None