fix: Address CI failures - type annotations, lint, security

- Fix TypeAlias annotation in elasticsearch/types.py using TYPE_CHECKING
- Add 'elasticsearch' to _MissingProvider Literal type in base.py
- Remove unused variable in test_client.py
- Add usedforsecurity=False to MD5 hash in config.py for security check

Co-Authored-By: João <joao@crewai.com>
This commit is contained in:
Devin AI
2025-08-27 01:17:40 +00:00
parent e3a575920c
commit dce26e8276
4 changed files with 13 additions and 8 deletions

View File

@@ -14,7 +14,7 @@ class _MissingProvider:
Raises RuntimeError when instantiated to indicate missing dependencies. Raises RuntimeError when instantiated to indicate missing dependencies.
""" """
provider: Literal["chromadb", "qdrant", "__missing__"] = field( provider: Literal["chromadb", "qdrant", "elasticsearch", "__missing__"] = field(
default="__missing__" default="__missing__"
) )

View File

@@ -61,7 +61,7 @@ def _default_embedding_function() -> ElasticsearchEmbeddingFunctionWrapper:
import hashlib import hashlib
import struct import struct
hash_obj = hashlib.md5(text.encode()) hash_obj = hashlib.md5(text.encode(), usedforsecurity=False)
hash_bytes = hash_obj.digest() hash_bytes = hash_obj.digest()
vector = [] vector = []

View File

@@ -1,15 +1,20 @@
"""Type definitions for Elasticsearch RAG implementation.""" """Type definitions for Elasticsearch RAG implementation."""
from typing import Any, Protocol, TypedDict, Union from typing import Any, Protocol, TypedDict, Union, TYPE_CHECKING
from typing_extensions import NotRequired from typing_extensions import NotRequired
from pydantic import GetCoreSchemaHandler from pydantic import GetCoreSchemaHandler
from pydantic_core import CoreSchema, core_schema from pydantic_core import CoreSchema, core_schema
try: if TYPE_CHECKING:
from typing import TypeAlias
from elasticsearch import Elasticsearch, AsyncElasticsearch from elasticsearch import Elasticsearch, AsyncElasticsearch
ElasticsearchClientType = Union[Elasticsearch, AsyncElasticsearch] ElasticsearchClientType: TypeAlias = Union[Elasticsearch, AsyncElasticsearch]
except ImportError: else:
ElasticsearchClientType = Any try:
from elasticsearch import Elasticsearch, AsyncElasticsearch
ElasticsearchClientType = Union[Elasticsearch, AsyncElasticsearch]
except ImportError:
ElasticsearchClientType = Any
class ElasticsearchClientParams(TypedDict, total=False): class ElasticsearchClientParams(TypedDict, total=False):

View File

@@ -179,7 +179,7 @@ class TestElasticsearchClient:
"""Test that get_or_create_collection creates new index if not exists.""" """Test that get_or_create_collection creates new index if not exists."""
mock_elasticsearch_client.indices.exists.return_value = False mock_elasticsearch_client.indices.exists.return_value = False
result = client.get_or_create_collection(collection_name="test_index") client.get_or_create_collection(collection_name="test_index")
mock_elasticsearch_client.indices.exists.assert_called_once_with(index="test_index") mock_elasticsearch_client.indices.exists.assert_called_once_with(index="test_index")
mock_elasticsearch_client.indices.create.assert_called_once() mock_elasticsearch_client.indices.create.assert_called_once()