Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions pyobvector/client/seekdb_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,8 +84,10 @@ def description(self) -> list[tuple[str]] | None:
return self._description

def close(self) -> None:
self._rows = None
self._description = None
# Rows are already buffered in memory by execute(); nothing to release here.
# Clearing _rows/_description would break callers that fetch results after
# SQLAlchemy closes the cursor on exiting the connection context block.
pass


class _SeekdbConnection:
Expand Down
31 changes: 23 additions & 8 deletions tests/test_seekdb_embedded.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,23 +39,38 @@ def _skip_if_no_embedded():
"pyseekdb not installed; run: pip install pyobvector[pyseekdb]",
)
class TestSeekdbEmbeddedConnection(unittest.TestCase):
"""Test ObClient/ObVecClient with embedded SeekDB (path= or pyseekdb_client=)."""
"""Test ObClient/ObVecClient with embedded SeekDB (path= or pyseekdb_client=).

def setUp(self) -> None:
All tests in this class share a single database directory because
pylibseekdb.open() is a process-level singleton — calling it a second time
(even at a different path) raises "initialized twice" and is silently ignored.
Re-creating the directory per test would leave the native library pointing at
a deleted path, causing a segfault on the next vector operation.
"""

_tmpdir: str = ""
_db_path: str = ""

@classmethod
def setUpClass(cls) -> None:
_skip_if_no_embedded()
self.tmpdir = tempfile.mkdtemp(prefix="pyobvector_seekdb_")
self.db_path = str(Path(self.tmpdir) / "seekdb_data")
Path(self.db_path).mkdir(parents=True, exist_ok=True)
cls._tmpdir = tempfile.mkdtemp(prefix="pyobvector_seekdb_")
cls._db_path = str(Path(cls._tmpdir) / "seekdb_data")
Path(cls._db_path).mkdir(parents=True, exist_ok=True)

def tearDown(self) -> None:
@classmethod
def tearDownClass(cls) -> None:
import shutil

if hasattr(self, "tmpdir") and Path(self.tmpdir).exists():
if cls._tmpdir and Path(cls._tmpdir).exists():
try:
shutil.rmtree(self.tmpdir, ignore_errors=True)
shutil.rmtree(cls._tmpdir, ignore_errors=True)
except Exception:
pass

def setUp(self) -> None:
self.db_path = self.__class__._db_path

def test_seekdb_remote_client_path_returns_ob_vec_client(self):
from pyobvector import SeekdbRemoteClient, ObVecClient
from pyobvector.client.ob_client import ObClient
Expand Down
Loading