refactor: unit tests for pydantic and pinecone deprecations

Author: lpm0073

.vscode/settings.json

@@ -2,5 +2,8 @@
   "cornflakes.linter.executablePath": "./venv/bin/flake8",
   "[python]": {
     "editor.defaultFormatter": "ms-python.black-formatter"
-  }
+  },
+  "python.testing.unittestArgs": ["-v", "-s", "./models", "-p", "*test.py"],
+  "python.testing.pytestEnabled": false,
+  "python.testing.unittestEnabled": true
 }

models/conf.py

@@ -124,6 +124,8 @@ def empty_str_to_int_default(v: str, default: int) -> int:
 class Settings(BaseSettings):
     """Settings for Lambda functions"""
 
+    model_config = {"frozen": True}
+
     _dump: Optional[dict] = None
     _pinecone_api_key_source: str = "unset"
     _openai_api_key_source: str = "unset"
@@ -170,8 +172,8 @@ def __init__(self, **data: Any):
     )
 
     pinecone_api_key: Optional[SecretStr] = Field(SettingsDefaults.PINECONE_API_KEY)
-    pinecone_environment: Optional[str] = Field(SettingsDefaults.PINECONE_ENVIRONMENT)
-    pinecone_index_name: Optional[str] = Field(SettingsDefaults.PINECONE_INDEX_NAME)
+    pinecone_environment: str = Field(SettingsDefaults.PINECONE_ENVIRONMENT)
+    pinecone_index_name: str = Field(SettingsDefaults.PINECONE_INDEX_NAME)
     pinecone_vectorstore_text_key: Optional[str] = Field(SettingsDefaults.PINECONE_VECTORSTORE_TEXT_KEY)
     pinecone_metric: Optional[str] = Field(SettingsDefaults.PINECONE_METRIC)
     pinecone_dimensions: Optional[int] = Field(SettingsDefaults.PINECONE_DIMENSIONS, gt=0)
@@ -279,12 +281,6 @@ def recursive_sort_dict(d):
         self._dump = recursive_sort_dict(self._dump)
         return self._dump
 
-    # pylint: disable=too-few-public-methods
-    class Config:
-        """Pydantic configuration"""
-
-        frozen = True
-
     @field_validator("debug_mode")
     def parse_debug_mode(cls, v) -> bool:
         """Parse debug_mode"""

models/pinecone.py

@@ -9,16 +9,22 @@
 import json
 import logging
 import os
+from typing import Optional
 
 from langchain.text_splitter import RecursiveCharacterTextSplitter
 from langchain_community.document_loaders.pdf import PyPDFLoader
 from langchain_openai import OpenAIEmbeddings
 from langchain_pinecone import PineconeVectorStore
 
 # pinecone integration
-from pinecone import Pinecone, ServerlessSpec
+from pinecone import AwsRegion, CloudProvider, Pinecone, ServerlessSpec, VectorType
+from pinecone.core.openapi.db_data.models import (
+    IndexDescription as PineconeIndexDescription,
+)
+from pinecone.db_control.models import IndexList
+from pinecone.db_data import Index
 from pinecone.exceptions import PineconeApiException
-from pinecone.models import IndexList
+from pydantic import SecretStr
 
 # this project
 from models.conf import settings
@@ -31,20 +37,20 @@ class PineconeIndex:
     """Pinecone helper class."""
 
     _pinecone = None
-    _index: Pinecone.Index = None
-    _index_name: str = None
-    _text_splitter: RecursiveCharacterTextSplitter = None
-    _openai_embeddings: OpenAIEmbeddings = None
-    _vector_store: PineconeVectorStore = None
+    _index: Optional[Index] = None
+    _index_name: Optional[str] = None
+    _text_splitter: Optional[RecursiveCharacterTextSplitter] = None
+    _openai_embeddings: Optional[OpenAIEmbeddings] = None
+    _vector_store: Optional[PineconeVectorStore] = None
 
-    def __init__(self, index_name: str = None):
+    def __init__(self, index_name: Optional[str] = None):
         self.init()
         self.index_name = index_name or settings.pinecone_index_name
         logging.debug("PineconeIndex initialized with index_name: %s", self.index_name)
         logging.debug(self.index_stats)
 
     @property
-    def index_name(self) -> str:
+    def index_name(self) -> Optional[str]:
         """index name."""
         return self._index_name
 
@@ -57,24 +63,30 @@ def index_name(self, value: str) -> None:
             self.init_index()
 
     @property
-    def index(self) -> Pinecone.Index:
+    def index(self) -> Optional[Index]:
         """pinecone.Index lazy read-only property."""
         if self._index is None:
             self.init_index()
-            self._index = self.pinecone.Index(name=self.index_name)
+            if isinstance(self.pinecone, Pinecone) and isinstance(self.index_name, str):
+                self._index = self.pinecone.Index(name=self.index_name)
+
         return self._index
 
     @property
-    def index_stats(self) -> dict:
+    def index_stats(self) -> str:
         """index stats."""
-        retval = self.index.describe_index_stats()
-        return json.dumps(retval.to_dict(), indent=4)
+        if self.index is not None:
+            retval: PineconeIndexDescription = self.index.describe_index_stats()
+            return json.dumps(retval.to_dict(), indent=4)
+        return "Index not initialized."
 
     @property
     def initialized(self) -> bool:
         """initialized read-only property."""
-        indexes = self.pinecone.list_indexes()
-        return self.index_name in indexes.names()
+        if isinstance(self.pinecone, Pinecone) and isinstance(self.index_name, str):
+            indexes = self.pinecone.list_indexes()
+            return self.index_name in indexes.names()
+        return False
 
     @property
     def vector_store(self) -> PineconeVectorStore:
@@ -95,19 +107,20 @@ def openai_embeddings(self) -> OpenAIEmbeddings:
         if self._openai_embeddings is None:
             # pylint: disable=no-member
             self._openai_embeddings = OpenAIEmbeddings(
-                api_key=settings.openai_api_key.get_secret_value(),
+                api_key=settings.openai_api_key,
                 organization=settings.openai_api_organization,
             )
         return self._openai_embeddings
 
     @property
-    def pinecone(self) -> Pinecone:
+    def pinecone(self) -> Optional[Pinecone]:
         """Pinecone lazy read-only property."""
         if self._pinecone is None:
             print("Initializing Pinecone...")
-            api_key = settings.pinecone_api_key.get_secret_value()
-            print(f"API Key: {api_key[:12]}****------")
-            self._pinecone = Pinecone(api_key=api_key)
+            if isinstance(settings.pinecone_api_key, SecretStr):
+                api_key = settings.pinecone_api_key.get_secret_value()
+                print(f"API Key: {api_key[:12]}****------")
+                self._pinecone = Pinecone(api_key=api_key)
         return self._pinecone
 
     @property
@@ -119,11 +132,11 @@ def text_splitter(self) -> RecursiveCharacterTextSplitter:
 
     def init_index(self):
         """Verify that an index named self.index_name exists in Pinecone. If not, create it."""
-        indexes: IndexList = None
-        indexes = self.pinecone.list_indexes()
-        if self.index_name not in indexes.names():
-            logging.debug("Index does not exist.")
-            self.create()
+        if isinstance(self.pinecone, Pinecone):
+            indexes: IndexList = self.pinecone.list_indexes()
+            if self.index_name not in indexes.names():
+                logging.debug("Index does not exist.")
+                self.create()
 
     # pylint: disable=no-member
     def init(self):
@@ -140,24 +153,27 @@ def delete(self):
         if not self.initialized:
             logging.debug("Index does not exist. Nothing to delete.")
             return
-        print("Deleting index...")
-        self.pinecone.delete_index(self.index_name)
+        if isinstance(self.pinecone, Pinecone) and isinstance(self.index_name, str):
+            print("Deleting index...")
+            self.pinecone.delete_index(self.index_name)
 
     def create(self):
         """Create index."""
         print("Creating index. This may take a few minutes...")
         serverless_spec = ServerlessSpec(
-            cloud="aws",
-            region="us-east-1",
+            cloud=CloudProvider.AWS,
+            region=AwsRegion.US_EAST_1,
         )
         try:
-            self.pinecone.create_index(
-                name=self.index_name,
-                dimension=settings.pinecone_dimensions,
-                metric=settings.pinecone_metric,
-                spec=serverless_spec,
-            )
-            print("Index created.")
+            if isinstance(self.pinecone, Pinecone) and isinstance(self.index_name, str):
+                self.pinecone.create_index(
+                    name=self.index_name,
+                    dimension=settings.pinecone_dimensions,
+                    metric=settings.pinecone_metric,
+                    spec=serverless_spec,
+                    vector_type=VectorType.DENSE,
+                )
+                print("Index created.")
         except PineconeApiException:
             pass
 

models/tests/test_pinecone.py

@@ -6,9 +6,10 @@
 
 import os
 
-import pinecone as oem_pinecone
-import pytest  # pylint: disable=unused-import
+import pytest
 from pinecone import Pinecone
+from pinecone.db_data import Index
+from pydantic import SecretStr
 
 from models.conf import settings
 from models.pinecone import PineconeIndex
@@ -47,13 +48,15 @@ def test_04_initialize(self):
             pinecone.initialize()
         except Exception as e:
             assert False, f"Pinecone.initialize() failed with exception: {e}"
-        assert isinstance(pinecone.index, oem_pinecone.Index)
+        assert isinstance(pinecone.index, Index)
 
     def test_05_delete(self):
         """Test that the index can be deleted."""
         pinecone_index = PineconeIndex()
 
-        # pylint: disable=E1101
+        if not isinstance(settings.pinecone_api_key, SecretStr):
+            raise ValueError("Pinecone API key is not a SecretStr")
+        # pylint: disable=no-member
         api_key = settings.pinecone_api_key.get_secret_value()
         pinecone = Pinecone(api_key=api_key)
         indexes = pinecone.list_indexes().names()
@@ -68,7 +71,10 @@ def test_06_create(self):
         """Test that the index can be created."""
         pinecone_index = PineconeIndex()
 
-        # pylint: disable=E1101
+        if not isinstance(settings.pinecone_api_key, SecretStr):
+            raise ValueError("Pinecone API key is not a SecretStr")
+
+        # pylint: disable=no-member
         api_key = settings.pinecone_api_key.get_secret_value()
         pinecone = Pinecone(api_key=api_key)
 
@@ -81,7 +87,7 @@ def test_06_create(self):
             pinecone_index.create()
         except Exception as e:
             assert False, f"Pinecone.create() failed with exception: {e}"
-        assert isinstance(pinecone_index.index, oem_pinecone.Index)
+        assert isinstance(pinecone_index.index, Index)
         pinecone_index.delete()
 
     def test_07_load_pdf(self):

requirements/base.txt

@@ -13,9 +13,9 @@ langsmith==0.3.45
 openai==1.88.0
 pinecone==7.2.0
 pinecone-text==0.10.0
-pydantic==2.10.4
+pydantic==2.11.7
 pydantic-settings==2.9.1
-pydantic_core==2.27.2
+pydantic_core==2.33.2
 python-decouple==3.8
 python-dotenv==1.1.0
 pypdf==5.6.0