chore: clean it

semantica/vector_store/pinecone_store.py

@@ -299,6 +299,44 @@ class PineconeSearch:
         )
 
 
+def _pinecone_listed_ids(response: Any) -> List[str]:
+    """Extract vector IDs from a list_paginated() response.
+
+    Accepts record objects, bare id strings and dicts, since what listing
+    returns has changed across pinecone SDK major versions.
+    """
+    records = getattr(response, "vectors", None)
+    if records is None and isinstance(response, dict):
+        records = response.get("vectors")
+
+    ids: List[str] = []
+    for record in records or []:
+        if isinstance(record, str):
+            ids.append(record)
+        elif isinstance(record, dict):
+            if record.get("id") is not None:
+                ids.append(record["id"])
+        else:
+            record_id = getattr(record, "id", None)
+            if record_id is not None:
+                ids.append(record_id)
+    return ids
+
+
+def _pinecone_next_token(response: Any) -> Optional[str]:
+    """Return the continuation token, or None when the listing is exhausted."""
+    pagination = getattr(response, "pagination", None)
+    if pagination is None and isinstance(response, dict):
+        pagination = response.get("pagination")
+    if pagination is None:
+        return None
+
+    token = getattr(pagination, "next", None)
+    if token is None and isinstance(pagination, dict):
+        token = pagination.get("next")
+    return token or None
+
+
 class PineconeStore:
     """
     Pinecone store for vector storage and similarity search.
@@ -735,6 +773,84 @@ class PineconeStore:
             self.logger.warning(f"Failed to filter Pinecone vectors by metadata: {e}")
             return []
 
+    def iter_all(self, batch_size: int = 500, namespace: str = ""):
+        """
+        Iterate over every stored vector by listing IDs then fetching them.
+
+        Paginates with an opaque continuation token, which is why this exists
+        instead of scan_vectors(offset, limit): the token for page N cannot be
+        constructed without walking there.
+
+        Needs two calls per page, unlike the other backends, because listing
+        returns IDs only. Both calls are namespace scoped and must agree, and
+        listing covers one namespace rather than the whole index.
+
+        Args:
+            batch_size: IDs to request per list_paginated() call
+            namespace: Namespace to enumerate (default: the default namespace)
+
+        Yields:
+            Result dicts with 'id', 'metadata', and 'vector', in listing order
+
+        Raises:
+            ProcessingError: If the index is not initialized, if the installed
+                SDK does not expose list_paginated(), or if the listing stops
+                advancing.
+        """
+        if self.index is None or not PINECONE_AVAILABLE:
+            raise ProcessingError(
+                "Index not initialized. Call create_index() or get_index() first."
+            )
+
+        # list_paginated() rather than list(): list() is an auto-paging
+        # iterator in current SDKs but reads as plain id lists in older
+        # examples. Threading the token explicitly is version-agnostic.
+        list_paginated = getattr(self.index.index, "list_paginated", None)
+        if not callable(list_paginated):
+            raise ProcessingError(
+                "This pinecone SDK version does not expose Index.list_paginated(), "
+                "which full enumeration requires."
+            )
+
+        token = None
+        while True:
+            kwargs: Dict[str, Any] = {"limit": batch_size, "namespace": namespace}
+            if token is not None:
+                kwargs["pagination_token"] = token
+
+            response = list_paginated(**kwargs)
+            vector_ids = _pinecone_listed_ids(response)
+            if not vector_ids:
+                return
+
+            fetched = self.index.fetch_vectors(vector_ids, namespace=namespace)
+            vectors = fetched.get("vectors") or {}
+
+            for vector_id in vector_ids:
+                entry = vectors.get(vector_id)
+                if entry is None:
+                    # fetch() omits ids it cannot find: deleted since listing.
+                    continue
+                values = entry.get("values")
+                yield {
+                    "id": vector_id,
+                    "metadata": entry.get("metadata") or {},
+                    "vector": np.array(values) if values is not None else None,
+                }
+
+            next_token = _pinecone_next_token(response)
+            if not next_token:
+                return
+            if next_token == token:
+                # Distinct from exhaustion above: a partial scan here would be
+                # indistinguishable from a complete one.
+                raise ProcessingError(
+                    "Pinecone returned the same pagination token twice, so the "
+                    "listing is not advancing. Refusing to return a truncated "
+                    "scan."
+                )
+            token = next_token
+
     def fetch_vectors(
         self, vector_ids: List[str], namespace: str = "", **options
     ) -> Dict[str, Any]:

tests/vector_store/test_pinecone_store.py

@@ -249,6 +249,166 @@ class TestPineconeIndex(unittest.TestCase):
         mock_index.query.assert_called_once()
 
 
+class TestPineconeIterAll(unittest.TestCase):
+    """PineconeStore.iter_all() list-then-fetch enumeration."""
+
+    def _page(self, ids, next_token):
+        """Stand-in for a list_paginated() response."""
+        response = MagicMock()
+        response.vectors = [MagicMock(id=vector_id) for vector_id in ids]
+        response.pagination = MagicMock(next=next_token)
+        return response
+
+    def _store(self, pages, fetch_results):
+        store = PineconeStore()
+        wrapper = MagicMock()
+        raw_index = MagicMock()
+        raw_index.list_paginated.side_effect = list(pages)
+        wrapper.index = raw_index
+        wrapper.fetch_vectors.side_effect = list(fetch_results)
+        store.index = wrapper
+        return store, wrapper, raw_index
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_threads_pagination_token_across_pages(self):
+        store, _, raw_index = self._store(
+            [self._page(["a", "b"], "token-1"), self._page(["c"], None)],
+            [
+                {"vectors": {"a": {"values": [0.1], "metadata": {}},
+                             "b": {"values": [0.2], "metadata": {}}}},
+                {"vectors": {"c": {"values": [0.3], "metadata": {}}}},
+            ],
+        )
+
+        result = list(store.iter_all(batch_size=2))
+
+        self.assertEqual([item["id"] for item in result], ["a", "b", "c"])
+        calls = raw_index.list_paginated.call_args_list
+        self.assertNotIn("pagination_token", calls[0][1])
+        self.assertEqual(calls[1][1]["pagination_token"], "token-1")
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_hydrates_listed_ids_with_a_fetch(self):
+        """Listing returns ids only, so each page needs a fetch()."""
+        store, wrapper, _ = self._store(
+            [self._page(["a"], None)],
+            [{"vectors": {"a": {"values": [0.1, 0.2], "metadata": {"tag": "x"}}}}],
+        )
+
+        item = list(store.iter_all())[0]
+
+        self.assertEqual(item["id"], "a")
+        self.assertEqual(item["metadata"], {"tag": "x"})
+        np.testing.assert_allclose(item["vector"], np.array([0.1, 0.2]))
+        wrapper.fetch_vectors.assert_called_once_with(["a"], namespace="")
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_list_and_fetch_use_the_same_namespace(self):
+        store, wrapper, raw_index = self._store(
+            [self._page(["a"], None)],
+            [{"vectors": {"a": {"values": [0.1], "metadata": {}}}}],
+        )
+
+        list(store.iter_all(namespace="prod"))
+
+        self.assertEqual(raw_index.list_paginated.call_args[1]["namespace"], "prod")
+        wrapper.fetch_vectors.assert_called_once_with(["a"], namespace="prod")
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_skips_ids_deleted_between_list_and_fetch(self):
+        """fetch() omits ids it cannot find rather than returning blanks."""
+        store, _, _ = self._store(
+            [self._page(["a", "gone"], None)],
+            [{"vectors": {"a": {"values": [0.1], "metadata": {}}}}],
+        )
+
+        result = list(store.iter_all())
+
+        self.assertEqual([item["id"] for item in result], ["a"])
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_raises_when_pagination_token_repeats(self):
+        """A stalled token must not loop forever, nor quietly return a partial
+        scan that reads as a complete one."""
+        store, _, raw_index = self._store(
+            [self._page(["a"], "same"), self._page(["b"], "same")],
+            [
+                {"vectors": {"a": {"values": [0.1], "metadata": {}}}},
+                {"vectors": {"b": {"values": [0.2], "metadata": {}}}},
+            ],
+        )
+
+        with self.assertRaises(ProcessingError):
+            list(store.iter_all())
+
+        self.assertEqual(raw_index.list_paginated.call_count, 2)
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_empty_listing_yields_nothing_without_fetching(self):
+        store, wrapper, _ = self._store([self._page([], None)], [])
+
+        self.assertEqual(list(store.iter_all()), [])
+        wrapper.fetch_vectors.assert_not_called()
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_accepts_plain_string_ids_from_listing(self):
+        """SDK generations differ on what listing yields."""
+        store, _, _ = self._store(
+            [self._page([], None)],
+            [{"vectors": {"a": {"values": [0.1], "metadata": {}}}}],
+        )
+        response = MagicMock()
+        response.vectors = ["a"]
+        response.pagination = MagicMock(next=None)
+        store.index.index.list_paginated.side_effect = [response]
+
+        self.assertEqual([item["id"] for item in store.iter_all()], ["a"])
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_handles_missing_values_and_metadata(self):
+        store, _, _ = self._store(
+            [self._page(["a"], None)],
+            [{"vectors": {"a": {"values": None, "metadata": None}}}],
+        )
+
+        item = list(store.iter_all())[0]
+
+        self.assertIsNone(item["vector"])
+        self.assertEqual(item["metadata"], {})
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_raises_when_list_paginated_unavailable(self):
+        store = PineconeStore()
+        wrapper = MagicMock()
+        wrapper.index = MagicMock(spec=["query", "fetch"])
+        store.index = wrapper
+
+        with self.assertRaises(ProcessingError):
+            list(store.iter_all())
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_raises_when_index_not_initialized(self):
+        """Must fail loudly: an empty scan reads the same as an empty source."""
+        with self.assertRaises(ProcessingError):
+            list(PineconeStore().iter_all())
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', False)
+    def test_raises_when_pinecone_unavailable(self):
+        store = PineconeStore()
+        store.index = MagicMock()
+
+        with self.assertRaises(ProcessingError):
+            list(store.iter_all())
+
+    @patch('semantica.vector_store.pinecone_store.PINECONE_AVAILABLE', True)
+    def test_propagates_listing_errors(self):
+        store, _, raw_index = self._store([], [])
+        raw_index.list_paginated.side_effect = RuntimeError("connection reset")
+
+        with self.assertRaises(RuntimeError):
+            list(store.iter_all())
+
+
 if __name__ == '__main__':
     print("DEBUG: Starting unittest.main()")
     unittest.main()