Bases: RecordStore
Records in the database at url (any SQLAlchemy URL); missing tables are created.
The default URL is an in-memory SQLite database, private to this store instance and shared
by all threads using it. Path and str chunk origins are kept apart with an
origin_is_path column. Backend-specific stores may subclass this where specific SQL is faster.
Source code in src/triplum/store/sql/generic.py
27
28
29
30
31
32
33
34
35
36 | def __init__(self, url: str = "sqlite://") -> None:
if url == "sqlite://":
# One shared connection; otherwise each thread gets its own empty in-memory database.
self.engine = create_engine(
url, poolclass=StaticPool, connect_args={"check_same_thread": False}
)
else:
self.engine = create_engine(url)
tables = [SQLModel.metadata.tables[row.__tablename__] for row in (SourceRow, ChunkRow)] # pyright: ignore[reportArgumentType]
SQLModel.metadata.create_all(self.engine, tables=tables) # not every imported table
|
engine
instance-attribute
engine = create_engine(
url,
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
add_sources
add_sources(sources: Iterable[Source]) -> None
Source code in src/triplum/store/sql/generic.py
38
39
40
41
42
43
44
45
46
47
48
49 | def add_sources(self, sources: Iterable[Source], /) -> None:
with Session(self.engine) as session:
for s in sources:
session.merge(
SourceRow(
id=s.id,
origin=s.origin,
text=s.text,
fingerprint=s.fingerprint,
)
)
session.commit()
|
add_chunks
add_chunks(chunks: Iterable[Chunk]) -> None
Source code in src/triplum/store/sql/generic.py
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70 | def add_chunks(self, chunks: Iterable[Chunk], /) -> None:
chunks = list(chunks)
with Session(self.engine) as session:
wanted = {c.source_id for c in chunks}
present = set(session.exec(select(SourceRow.id).where(col(SourceRow.id).in_(wanted))))
for missing in wanted - present:
raise KeyError(missing)
for c in chunks:
session.merge(
ChunkRow(
id=c.id,
source_id=c.source_id,
origin=str(c.origin),
origin_is_path=isinstance(c.origin, Path),
start=c.start,
text=c.text,
fingerprint=c.fingerprint,
)
)
session.commit()
|
source
Source code in src/triplum/store/sql/generic.py
| def source(self, id: UUID, /) -> Source:
with Session(self.engine) as session:
row = session.get(SourceRow, id)
if row is None:
raise KeyError(id)
return Source(id=row.id, origin=row.origin, text=row.text)
|
sources
Source code in src/triplum/store/sql/generic.py
| def sources(self) -> Iterator[Source]:
with Session(self.engine) as session:
rows = session.exec(select(SourceRow)).all()
for row in rows:
yield Source(id=row.id, origin=row.origin, text=row.text)
|
chunks
chunks(source_id: UUID) -> list[Chunk]
Source code in src/triplum/store/sql/generic.py
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101 | def chunks(self, source_id: UUID, /) -> list[Chunk]:
with Session(self.engine) as session:
rows = session.exec(
select(ChunkRow)
.where(ChunkRow.source_id == source_id)
.order_by(col(ChunkRow.start))
).all()
return [
Chunk(
id=row.id,
source_id=row.source_id,
origin=_origin(row.origin, row.origin_is_path),
start=row.start,
text=row.text,
)
for row in rows
]
|