Storage Providers

from typeagent.knowpro.interfaces import IStorageProvider

async def get_message_collection(self) -> IMessageCollection[TMessage]

async def get_semantic_ref_collection(self) -> ISemanticRefCollection

async def get_semantic_ref_index(self) -> ITermToSemanticRefIndex

async def get_property_index(self) -> IPropertyToSemanticRefIndex

async def get_timestamp_index(self) -> ITimestampToTextRangeIndex

async def get_message_text_index(self) -> IMessageTextIndex[TMessage]

async def get_related_terms_index(self) -> ITermToRelatedTermsIndex

async def get_conversation_threads(self) -> IConversationThreads

async def get_conversation_metadata(self) -> ConversationMetadata

async def set_conversation_metadata(
    self,
    **kwds: str | list[str] | None
) -> None

async def update_conversation_timestamps(
    self,
    created_at: datetime | None = None,
    updated_at: datetime | None = None,
) -> None

async def is_source_ingested(self, source_id: str) -> bool

async def get_source_status(self, source_id: str) -> str | None

async def mark_source_ingested(
    self,
    source_id: str,
    status: str = STATUS_INGESTED
) -> None

async with storage_provider:
    # All operations here are atomic
    await storage_provider.mark_source_ingested("source:123")
    await collection.extend(messages)
    # Commits on success, rolls back on exception

async def __aenter__(self) -> Self

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any,
) -> None

async def close(self) -> None

from typeagent.storage.memory import MemoryStorageProvider

def __init__(
    self,
    message_text_settings: MessageTextIndexSettings,
    related_terms_settings: RelatedTermIndexSettings,
    metadata: ConversationMetadata | None = None,
) -> None

from typeagent.storage.memory import MemoryStorageProvider
from typeagent.knowpro.convsettings import (
    MessageTextIndexSettings,
    RelatedTermIndexSettings
)
from typeagent.aitools.vectorbase import TextEmbeddingIndexSettings
from typeagent.aitools.model_adapters import create_embedding_model

# Create embedding settings
model = create_embedding_model("openai:text-embedding-3-small")
embedding_settings = TextEmbeddingIndexSettings(model)

# Create provider
provider = MemoryStorageProvider(
    message_text_settings=MessageTextIndexSettings(embedding_settings),
    related_terms_settings=RelatedTermIndexSettings(embedding_settings)
)

# Use with conversation
conv = await ConversationBase.create(
    settings=ConversationSettings(storage_provider=provider)
)

from typeagent.storage.sqlite import SqliteStorageProvider

def __init__(
    self,
    db_path: str = ":memory:",
    message_type: type[TMessage] = None,
    semantic_ref_type: type[SemanticRef] = None,
    message_text_index_settings: MessageTextIndexSettings | None = None,
    related_term_index_settings: RelatedTermIndexSettings | None = None,
    metadata: ConversationMetadata | None = None,
)

PRAGMA foreign_keys = ON
PRAGMA synchronous = NORMAL       # Faster than FULL, still safe
PRAGMA journal_mode = WAL         # Write-Ahead Logging
PRAGMA cache_size = -64000        # 64MB cache
PRAGMA temp_store = MEMORY        # Temp tables in memory
PRAGMA mmap_size = 268435456      # 256MB memory-mapped I/O

ValueError: Conversation metadata embedding_model (text-embedding-ada-002) 
does not match provided embedding model (text-embedding-3-small).

from typeagent.storage.sqlite import SqliteStorageProvider

provider = SqliteStorageProvider(
    db_path="chat.db",
    message_type=ConversationMessage
)

try:
    async with provider:
        # All operations in this block are atomic
        await provider.mark_source_ingested("source:123")
        
        messages_coll = await provider.get_message_collection()
        await messages_coll.extend(messages)
        
        semrefs_coll = await provider.get_semantic_ref_collection()
        await semrefs_coll.extend(semantic_refs)
        
        # Commits here if no exceptions
except Exception as e:
    # All changes rolled back automatically
    print(f"Transaction failed: {e}")
finally:
    await provider.close()

from typeagent.storage.sqlite import SqliteStorageProvider
from typeagent.knowpro.universal_message import ConversationMessage

# Create persistent storage
provider = SqliteStorageProvider(
    db_path="conversations/team_chat.db",
    message_type=ConversationMessage
)

# Get collections
messages = await provider.get_message_collection()
semrefs = await provider.get_semantic_ref_collection()

print(f"Database contains {await messages.size()} messages")
print(f"Extracted {await semrefs.size()} semantic references")

# Check metadata
metadata = await provider.get_conversation_metadata()
print(f"Conversation: {metadata.name_tag}")
print(f"Created: {metadata.created_at}")
print(f"Embedding model: {metadata.embedding_model}")

await provider.close()

from typeagent.knowpro.interfaces import ConversationMetadata