Skip to content

memory

triplum.store.memory

MemoryStore 

MemoryStore()

Bases: RecordStore

Records in dictionaries; gone when the process ends.

Source code in src/triplum/store/memory.py 
11
12
13
def __init__(self) -> None:
    self._sources: dict[UUID, Source] = {}
    self._chunks: dict[UUID, Chunk] = {}

add_sources 

add_sources(sources: Iterable[Source]) -> None
Source code in src/triplum/store/memory.py 
15
16
def add_sources(self, sources: Iterable[Source], /) -> None:
    self._sources.update((source.id, source) for source in sources)

add_chunks 

add_chunks(chunks: Iterable[Chunk]) -> None
Source code in src/triplum/store/memory.py 
18
19
20
21
22
23
def add_chunks(self, chunks: Iterable[Chunk], /) -> None:
    chunks = list(chunks)
    for chunk in chunks:
        if chunk.source_id not in self._sources:
            raise KeyError(chunk.source_id)
    self._chunks.update((chunk.id, chunk) for chunk in chunks)

source 

source(id: UUID) -> Source
Source code in src/triplum/store/memory.py 
25
26
def source(self, id: UUID, /) -> Source:
    return self._sources[id]

sources 

sources() -> Iterator[Source]
Source code in src/triplum/store/memory.py 
28
29
def sources(self) -> Iterator[Source]:
    return iter(list(self._sources.values()))

chunks 

chunks(source_id: UUID) -> list[Chunk]
Source code in src/triplum/store/memory.py 
31
32
33
def chunks(self, source_id: UUID, /) -> list[Chunk]:
    found = [chunk for chunk in self._chunks.values() if chunk.source_id == source_id]
    return sorted(found, key=lambda chunk: chunk.start)