symfonic.core.documents.in_memory

in_memory

In-memory document store — reference implementation for testing/CI.

Not thread-safe. Capped at 10,000 documents.

InMemoryDocumentStore

InMemoryDocumentStore()

Reference DocumentStore for testing/CI. Not thread-safe. Cap: 10_000.

Source code in src/symfonic/core/documents/in_memory.py

def __init__(self) -> None:
    self._docs: dict[str, Document] = {}

read async

read(doc_id: str) -> Document | None

Return a document by ID, or None if not found.

Source code in src/symfonic/core/documents/in_memory.py

async def read(self, doc_id: str) -> Document | None:
    """Return a document by ID, or None if not found."""
    return self._docs.get(doc_id)

search async

search(
    query: str, limit: int = 10, offset: int = 0
) -> list[Document]

Search documents by content or ID substring match.

Source code in src/symfonic/core/documents/in_memory.py

async def search(
    self, query: str, limit: int = 10, offset: int = 0
) -> list[Document]:
    """Search documents by content or ID substring match."""
    results = [
        d
        for d in self._docs.values()
        if query.lower() in d.content.lower()
        or query.lower() in d.id.lower()
    ]
    return results[offset : offset + limit]

write async

write(doc: Document) -> None

Store a document. Raises StorageError if cap exceeded.

Source code in src/symfonic/core/documents/in_memory.py

async def write(self, doc: Document) -> None:
    """Store a document. Raises StorageError if cap exceeded."""
    if len(self._docs) >= self.MAX_DOCUMENTS and doc.id not in self._docs:
        raise StorageError(
            f"InMemoryDocumentStore cap exceeded ({self.MAX_DOCUMENTS})"
        )
    self._docs[doc.id] = doc