Skip to content

sqlite

sqlite

SQLite/FTS5 memory backend — zero-dependency default.

Classes

SQLiteMemory 

SQLiteMemory(db_path: str | Path = '')

Bases: MemoryBackend

Full-text search memory backend using SQLite FTS5.

Uses the built-in sqlite3 module — no extra dependencies.

Source code in src/openjarvis/tools/storage/sqlite.py 
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, db_path: str | Path = "") -> None:
    if not db_path:
        from openjarvis.core.config import DEFAULT_CONFIG_DIR
        db_path = str(DEFAULT_CONFIG_DIR / "memory.db")

    self._db_path = str(db_path)

    from openjarvis._rust_bridge import get_rust_module
    _rust = get_rust_module()
    self._rust_impl = _rust.SQLiteMemory(self._db_path)
    self._conn = None  # type: ignore[assignment]
Functions
store 
store(content: str, *, source: str = '', metadata: Optional[Dict[str, Any]] = None) -> str

Persist content and return a unique document id.

Source code in src/openjarvis/tools/storage/sqlite.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def store(
    self,
    content: str,
    *,
    source: str = "",
    metadata: Optional[Dict[str, Any]] = None,
) -> str:
    """Persist *content* and return a unique document id."""
    meta_json = json.dumps(metadata) if metadata else None
    doc_id = self._rust_impl.store(content, source, meta_json)
    bus = get_event_bus()
    bus.publish(EventType.MEMORY_STORE, {
        "backend": self.backend_id,
        "doc_id": doc_id,
        "source": source,
    })
    return doc_id
retrieve 
retrieve(query: str, *, top_k: int = 5, **kwargs: Any) -> List[RetrievalResult]

Search via FTS5 MATCH with BM25 ranking — always via Rust backend.

Source code in src/openjarvis/tools/storage/sqlite.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def retrieve(
    self,
    query: str,
    *,
    top_k: int = 5,
    **kwargs: Any,
) -> List[RetrievalResult]:
    """Search via FTS5 MATCH with BM25 ranking — always via Rust backend."""
    if not query.strip():
        return []

    from openjarvis._rust_bridge import retrieval_results_from_json
    results = retrieval_results_from_json(
        self._rust_impl.retrieve(query, top_k),
    )
    bus = get_event_bus()
    bus.publish(EventType.MEMORY_RETRIEVE, {
        "backend": self.backend_id,
        "query": query,
        "num_results": len(results),
    })
    return results
delete 
delete(doc_id: str) -> bool

Delete a document by id — always via Rust backend.

Source code in src/openjarvis/tools/storage/sqlite.py 
129
130
131
def delete(self, doc_id: str) -> bool:
    """Delete a document by id — always via Rust backend."""
    return self._rust_impl.delete(doc_id)
clear 
clear() -> None

Remove all stored documents — always via Rust backend.

Source code in src/openjarvis/tools/storage/sqlite.py 
133
134
135
def clear(self) -> None:
    """Remove all stored documents — always via Rust backend."""
    self._rust_impl.clear()
count 
count() -> int

Return the number of stored documents — always via Rust backend.

Source code in src/openjarvis/tools/storage/sqlite.py 
137
138
139
def count(self) -> int:
    """Return the number of stored documents — always via Rust backend."""
    return self._rust_impl.count()
close 
close() -> None

Close the database connection.

Source code in src/openjarvis/tools/storage/sqlite.py 
141
142
143
def close(self) -> None:
    """Close the database connection."""
    pass

Functions