RedisDocumentIndex

docarray.index.backends.redis.RedisDocumentIndex

Bases: BaseDocIndex, Generic[TSchema]

Source code in docarray/index/backends/redis.py

class RedisDocumentIndex(BaseDocIndex, Generic[TSchema]):
    def __init__(self, db_config=None, **kwargs):
        """Initialize RedisDocumentIndex"""
        super().__init__(db_config=db_config, **kwargs)
        self._db_config = cast(RedisDocumentIndex.DBConfig, self._db_config)

        self._runtime_config: RedisDocumentIndex.RuntimeConfig = cast(
            RedisDocumentIndex.RuntimeConfig, self._runtime_config
        )
        self._prefix = self.index_name + ':'
        self._text_scorer = self._db_config.text_scorer
        # initialize Redis client
        self._client = redis.Redis(
            host=self._db_config.host,
            port=self._db_config.port,
            username=self._db_config.username,
            password=self._db_config.password,
            decode_responses=False,
        )
        self._create_index()
        self._logger.info(f'{self.__class__.__name__} has been initialized')

    def _create_index(self) -> None:
        """Create a new index in the Redis database if it doesn't already exist."""
        if not self._check_index_exists(self.index_name):
            schema = []
            for column, info in self._column_infos.items():
                if safe_issubclass(info.docarray_type, AnyDocArray):
                    continue
                elif info.db_type == VectorField:
                    space = info.config.get('space') or info.config.get('distance')
                    if not space or space.upper() not in VALID_DISTANCES:
                        raise ValueError(
                            f"Invalid distance metric '{space}' provided. "
                            f"Must be one of: {', '.join(VALID_DISTANCES)}"
                        )
                    space = space.upper()
                    attributes = {
                        'TYPE': 'FLOAT32',
                        'DIM': info.n_dim or info.config.get('dim'),
                        'DISTANCE_METRIC': space,
                        'EF_CONSTRUCTION': info.config['ef_construction'],
                        'EF_RUNTIME': info.config['ef_runtime'],
                        'M': info.config['m'],
                        'INITIAL_CAP': info.config['initial_cap'],
                    }
                    attributes = {
                        name: value for name, value in attributes.items() if value
                    }
                    algorithm = info.config['algorithm'].upper()
                    if algorithm not in VALID_ALGORITHMS:
                        raise ValueError(
                            f"Invalid algorithm '{algorithm}' provided. "
                            f"Must be one of: {', '.join(VALID_ALGORITHMS)}"
                        )
                    schema.append(
                        info.db_type(
                            '$.' + column,
                            algorithm=algorithm,
                            attributes=attributes,
                            as_name=column,
                        )
                    )
                elif column in ['id', 'parent_id']:
                    schema.append(TagField('$.' + column, as_name=column))
                else:
                    schema.append(info.db_type('$.' + column, as_name=column))

            # Create Redis Index
            self._client.ft(self.index_name).create_index(
                schema,
                definition=IndexDefinition(
                    prefix=[self._prefix], index_type=IndexType.JSON
                ),
            )

            self._logger.info(f'index {self.index_name} has been created')
        else:
            self._logger.info(f'connected to existing {self.index_name} index')

    def _check_index_exists(self, index_name: str) -> bool:
        """
        Check if an index exists in the Redis database.

        :param index_name: The name of the index.
        :return: True if the index exists, False otherwise.
        """
        try:
            self._client.ft(index_name).info()
        except:  # noqa: E722
            self._logger.info(f'Index {index_name} does not exist')
            return False
        self._logger.info(f'Index {index_name} already exists')
        return True

    @property
    def index_name(self):
        default_index_name = (
            self._schema.__name__.lower() if self._schema is not None else None
        )
        if default_index_name is None:
            err_msg = (
                'A RedisDocumentIndex must be typed with a Document type. '
                'To do so, use the syntax: RedisDocumentIndex[DocumentType]'
            )

            self._logger.error(err_msg)
            raise ValueError(err_msg)
        index_name = self._db_config.index_name or default_index_name
        self._logger.debug(f'Retrieved index name: {index_name}')
        return index_name

    @property
    def out_schema(self) -> Type[BaseDoc]:
        """Return the real schema of the index."""
        if self._is_subindex:
            return self._ori_schema
        return cast(Type[BaseDoc], self._schema)

    class QueryBuilder(BaseDocIndex.QueryBuilder):
        def __init__(self, query: Optional[List[Tuple[str, Dict]]] = None):
            super().__init__()
            # list of tuples (method name, kwargs)
            self._queries: List[Tuple[str, Dict]] = query or []

        def build(self, *args, **kwargs) -> Any:
            """Build the query object."""
            return self._queries

        find = _collect_query_args('find')
        filter = _collect_query_args('filter')
        text_search = _raise_not_composable('text_search')
        find_batched = _raise_not_composable('find_batched')
        filter_batched = _raise_not_composable('filter_batched')
        text_search_batched = _raise_not_composable('text_search_batched')

    @dataclass
    class DBConfig(BaseDocIndex.DBConfig):
        """Dataclass that contains all "static" configurations of RedisDocumentIndex.

        :param host: The host address for the Redis server. Default is 'localhost'.
        :param port: The port number for the Redis server. Default is 6379.
        :param index_name: The name of the index in the Redis database.
            If not provided, default index name will be used.
        :param username: The username for the Redis server. Default is None.
        :param password: The password for the Redis server. Default is None.
        :param text_scorer: The method for scoring text during text search.
            Default is 'BM25'.
        :param default_column_config: Default configuration for columns.
        """

        host: str = 'localhost'
        port: int = 6379
        index_name: Optional[str] = None
        username: Optional[str] = None
        password: Optional[str] = None
        text_scorer: str = field(default='BM25')
        default_column_config: Dict[Type, Dict[str, Any]] = field(
            default_factory=lambda: defaultdict(
                dict,
                {
                    VectorField: {
                        'algorithm': 'FLAT',
                        'distance': 'COSINE',
                        'ef_construction': None,
                        'm': None,
                        'ef_runtime': None,
                        'initial_cap': None,
                    },
                },
            )
        )

        def __post_init__(self):
            self.text_scorer = self.text_scorer.upper()

            if self.text_scorer not in VALID_TEXT_SCORERS:
                raise ValueError(
                    f"Invalid text scorer '{self.text_scorer}' provided. "
                    f"Must be one of: {', '.join(VALID_TEXT_SCORERS)}"
                )

    @dataclass
    class RuntimeConfig(BaseDocIndex.RuntimeConfig):
        """Dataclass that contains all "dynamic" configurations of RedisDocumentIndex.

        :param batch_size: Batch size for index/get/del.
        """

        batch_size: int = 100

    def python_type_to_db_type(self, python_type: Type) -> Any:
        """
        Map python types to corresponding Redis types.

        :param python_type: Python type.
        :return: Corresponding Redis type.
        """
        type_map = {
            int: NumericField,
            float: NumericField,
            str: TextField,
            bytes: TextField,
            np.ndarray: VectorField,
            list: VectorField,
            AbstractTensor: VectorField,
        }

        for py_type, redis_type in type_map.items():
            if safe_issubclass(python_type, py_type):
                return redis_type
        raise ValueError(f'Unsupported column type for {type(self)}: {python_type}')

    @staticmethod
    def _generate_items(
        column_to_data: Dict[str, Generator[Any, None, None]],
        batch_size: int,
    ) -> Iterator[List[Dict[str, Any]]]:
        """
        Given a dictionary of data generators, yield a list of dictionaries where each
        item consists of a column name and a single item from the corresponding generator.

        :param column_to_data: A dictionary where each key is a column name and each value
            is a generator.
        :param batch_size: Size of batch to generate each time.

        :yield: A list of dictionaries where each item consists of a column name and
            an item from the corresponding generator. Yields until all generators
            are exhausted.
        """
        column_names = list(column_to_data.keys())
        data_generators = [iter(column_to_data[name]) for name in column_names]
        batch: List[Dict[str, Any]] = []

        while True:
            data_dict = {}
            for name, generator in zip(column_names, data_generators):
                item = next(generator, None)

                if name == 'id' and not item:
                    if batch:
                        yield batch
                    return

                if isinstance(item, AbstractTensor):
                    data_dict[name] = item._docarray_to_ndarray().tolist()
                elif isinstance(item, ndarray):
                    data_dict[name] = item.astype(np.float32).tolist()
                elif item is not None:
                    data_dict[name] = item

            batch.append(data_dict)
            if len(batch) == batch_size:
                yield batch
                batch = []

    def _index(
        self, column_to_data: Dict[str, Generator[Any, None, None]]
    ) -> List[str]:
        """
        Indexes the given data into Redis.

        :param column_to_data: A dictionary where each key is a column and each value is a generator.
        :return: A list of document ids that have been indexed.
        """
        self._index_subindex(column_to_data)
        ids: List[str] = []
        for items in self._generate_items(
            column_to_data, self._runtime_config.batch_size
        ):
            doc_id_item_pairs = [
                (self._prefix + item['id'], '$', item) for item in items
            ]
            ids.extend(doc_id for doc_id, _, _ in doc_id_item_pairs)
            self._client.json().mset(doc_id_item_pairs)  # type: ignore[attr-defined]

        return ids

    def num_docs(self) -> int:
        """
        Fetch the number of documents in the index.

        :return: Number of documents in the index.
        """
        num_docs = self._client.ft(self.index_name).info()['num_docs']
        return int(num_docs)

    def _del_items(self, doc_ids: Sequence[str]) -> None:
        """
        Deletes documents from the index based on document ids.

        :param doc_ids: A sequence of document ids to be deleted.
        """
        doc_ids = [self._prefix + id for id in doc_ids if self._doc_exists(id)]
        if doc_ids:
            for batch in self._generate_batches(
                doc_ids, batch_size=self._runtime_config.batch_size
            ):
                self._client.delete(*batch)

    def _doc_exists(self, doc_id: str) -> bool:
        """
        Checks if a document exists in the index.

        :param doc_id: The id of the document.
        :return: True if the document exists, False otherwise.
        """
        return bool(self._client.exists(self._prefix + doc_id))

    @staticmethod
    def _generate_batches(data, batch_size):
        for i in range(0, len(data), batch_size):
            yield data[i : i + batch_size]

    def _get_items(
        self, doc_ids: Sequence[str]
    ) -> Union[Sequence[TSchema], Sequence[Dict[str, Any]]]:
        """
        Fetches the documents from the index based on document ids.

        :param doc_ids: A sequence of document ids.
        :return: A sequence of documents from the index.
        """
        if not doc_ids:
            return []
        docs: List[Dict[str, Any]] = []
        for batch in self._generate_batches(
            doc_ids, batch_size=self._runtime_config.batch_size
        ):
            ids = [self._prefix + id for id in batch]
            retrieved_docs = self._client.json().mget(ids, '$')
            docs.extend(doc[0] for doc in retrieved_docs if doc)

        if not docs:
            raise KeyError(f'No document with id {doc_ids} found')
        return docs

    def execute_query(self, query: Any, *args: Any, **kwargs: Any) -> Any:
        """
        Executes a hybrid query on the index.

        :param query: Query to execute on the index.
        :return: Query results.
        """
        components: Dict[str, List[Dict[str, Any]]] = {}
        for component, value in query:
            if component not in components:
                components[component] = []
            components[component].append(value)

        if (
            len(components) != 2
            or len(components.get('find', [])) != 1
            or len(components.get('filter', [])) != 1
        ):
            raise ValueError(
                'The query must contain exactly one "find" and "filter" components.'
            )

        filter_query = components['filter'][0]['filter_query']
        query = components['find'][0]['query']
        search_field = components['find'][0]['search_field']
        limit = (
            components['find'][0].get('limit')
            or components['filter'][0].get('limit')
            or 10
        )
        docs, scores = self._hybrid_search(
            query=query,
            filter_query=filter_query,
            search_field=search_field,
            limit=limit,
        )
        docs = self._dict_list_to_docarray(docs)
        return FindResult(documents=docs, scores=scores)

    def _hybrid_search(
        self, query: np.ndarray, filter_query: str, search_field: str, limit: int
    ) -> _FindResult:
        """
        Conducts a hybrid search (a combination of vector search and filter-based search) on the index.

        :param query: The query to search.
        :param filter_query: The filter condition.
        :param search_field: The vector field to search on.
        :param limit: The maximum number of results to return.
        :return: Query results.
        """
        redis_query = (
            Query(f'{filter_query}=>[KNN {limit} @{search_field} $vec AS vector_score]')
            .sort_by('vector_score')
            .paging(0, limit)
            .dialect(2)
        )
        query_params: Mapping[str, bytes] = {
            'vec': np.array(query, dtype=np.float32).tobytes()
        }
        results = (
            self._client.ft(self.index_name).search(redis_query, query_params).docs  # type: ignore[arg-type]
        )

        scores: NdArray = NdArray._docarray_from_native(
            np.array([document['vector_score'] for document in results])
        )

        docs = []
        for out_doc in results:
            doc_dict = json.loads(out_doc.json)
            docs.append(doc_dict)
        return _FindResult(documents=docs, scores=scores)

    def _find(
        self, query: np.ndarray, limit: int, search_field: str = ''
    ) -> _FindResult:
        """
        Conducts a search on the index.

        :param query: The vector query to search.
        :param limit: The maximum number of results to return.
        :param search_field: The field to search the query.
        :return: Search results.
        """
        return self._hybrid_search(
            query=query, filter_query='*', search_field=search_field, limit=limit
        )

    def _find_batched(
        self, queries: np.ndarray, limit: int, search_field: str = ''
    ) -> _FindResultBatched:
        """
        Conducts a batched search on the index.

        :param queries: The queries to search.
        :param limit: The maximum number of results to return for each query.
        :param search_field: The field to search the queries.
        :return: Search results.
        """
        docs, scores = [], []
        for query in queries:
            results = self._find(query=query, search_field=search_field, limit=limit)
            docs.append(results.documents)
            scores.append(results.scores)

        return _FindResultBatched(documents=docs, scores=scores)

    def _filter(self, filter_query: Any, limit: int) -> Union[DocList, List[Dict]]:
        """
        Filters the index based on the given filter query.

        :param filter_query: The filter condition.
        :param limit: The maximum number of results to return.
        :return: Filter results.
        """
        q = Query(filter_query)
        q.paging(0, limit)

        results = self._client.ft(index_name=self.index_name).search(q).docs
        docs = [json.loads(doc.json) for doc in results]
        return docs

    def _filter_batched(
        self, filter_queries: Any, limit: int
    ) -> Union[List[DocList], List[List[Dict]]]:
        """
        Filters the index based on the given batch of filter queries.

        :param filter_queries: The filter conditions.
        :param limit: The maximum number of results to return for each filter query.
        :return: Filter results.
        """
        results = []
        for query in filter_queries:
            results.append(self._filter(filter_query=query, limit=limit))
        return results

    def _filter_by_parent_id(self, id: str) -> Optional[List[str]]:
        """Filter the ids of the subindex documents given id of root document.

        :param id: the root document id to filter by
        :return: a list of ids of the subindex documents
        """
        docs = self._filter(filter_query=f'@parent_id:{{{id}}}', limit=self.num_docs())
        return [doc['id'] for doc in docs]

    def _text_search(
        self, query: str, limit: int, search_field: str = ''
    ) -> _FindResult:
        """
        Conducts a text-based search on the index.

        :param query: The query to search.
        :param limit: The maximum number of results to return.
        :param search_field: The field to search the query.
        :return: Search results.
        """
        query_str = '|'.join(query.split(' '))
        q = (
            Query(f'@{search_field}:{query_str}')
            .scorer(self._text_scorer)
            .with_scores()
            .paging(0, limit)
        )

        results = self._client.ft(index_name=self.index_name).search(q).docs

        scores: NdArray = NdArray._docarray_from_native(
            np.array([document['score'] for document in results])
        )

        docs = [json.loads(doc.json) for doc in results]

        return _FindResult(documents=docs, scores=scores)

    def _text_search_batched(
        self, queries: Sequence[str], limit: int, search_field: str = ''
    ) -> _FindResultBatched:
        """
        Conducts a batched text-based search on the index.

        :param queries: The queries to search.
        :param limit: The maximum number of results to return for each query.
        :param search_field: The field to search the queries.
        :return: Search results.
        """
        docs, scores = [], []
        for query in queries:
            results = self._text_search(
                query=query, search_field=search_field, limit=limit
            )
            docs.append(results.documents)
            scores.append(results.scores)

        return _FindResultBatched(documents=docs, scores=scores)

out_schema: Type[BaseDoc] property

Return the real schema of the index.

DBConfig dataclass

Bases: DBConfig

Dataclass that contains all "static" configurations of RedisDocumentIndex.

Parameters:

Name	Type	Description	Default
host	str	The host address for the Redis server. Default is 'localhost'.	'localhost'
port	int	The port number for the Redis server. Default is 6379.	6379
index_name	Optional[str]	The name of the index in the Redis database. If not provided, default index name will be used.	None
username	Optional[str]	The username for the Redis server. Default is None.	None
password	Optional[str]	The password for the Redis server. Default is None.	None
text_scorer	str	The method for scoring text during text search. Default is 'BM25'.	field(default='BM25')
default_column_config	Dict[Type, Dict[str, Any]]	Default configuration for columns.	field(default_factory=lambda : defaultdict(dict, {VectorField: {'algorithm': 'FLAT', 'distance': 'COSINE', 'ef_construction': None, 'm': None, 'ef_runtime': None, 'initial_cap': None}}))

Source code in docarray/index/backends/redis.py

@dataclass
class DBConfig(BaseDocIndex.DBConfig):
    """Dataclass that contains all "static" configurations of RedisDocumentIndex.

    :param host: The host address for the Redis server. Default is 'localhost'.
    :param port: The port number for the Redis server. Default is 6379.
    :param index_name: The name of the index in the Redis database.
        If not provided, default index name will be used.
    :param username: The username for the Redis server. Default is None.
    :param password: The password for the Redis server. Default is None.
    :param text_scorer: The method for scoring text during text search.
        Default is 'BM25'.
    :param default_column_config: Default configuration for columns.
    """

    host: str = 'localhost'
    port: int = 6379
    index_name: Optional[str] = None
    username: Optional[str] = None
    password: Optional[str] = None
    text_scorer: str = field(default='BM25')
    default_column_config: Dict[Type, Dict[str, Any]] = field(
        default_factory=lambda: defaultdict(
            dict,
            {
                VectorField: {
                    'algorithm': 'FLAT',
                    'distance': 'COSINE',
                    'ef_construction': None,
                    'm': None,
                    'ef_runtime': None,
                    'initial_cap': None,
                },
            },
        )
    )

    def __post_init__(self):
        self.text_scorer = self.text_scorer.upper()

        if self.text_scorer not in VALID_TEXT_SCORERS:
            raise ValueError(
                f"Invalid text scorer '{self.text_scorer}' provided. "
                f"Must be one of: {', '.join(VALID_TEXT_SCORERS)}"
            )

QueryBuilder

Bases: QueryBuilder

Source code in docarray/index/backends/redis.py

class QueryBuilder(BaseDocIndex.QueryBuilder):
    def __init__(self, query: Optional[List[Tuple[str, Dict]]] = None):
        super().__init__()
        # list of tuples (method name, kwargs)
        self._queries: List[Tuple[str, Dict]] = query or []

    def build(self, *args, **kwargs) -> Any:
        """Build the query object."""
        return self._queries

    find = _collect_query_args('find')
    filter = _collect_query_args('filter')
    text_search = _raise_not_composable('text_search')
    find_batched = _raise_not_composable('find_batched')
    filter_batched = _raise_not_composable('filter_batched')
    text_search_batched = _raise_not_composable('text_search_batched')

build(*args, **kwargs)

Build the query object.

Source code in docarray/index/backends/redis.py

def build(self, *args, **kwargs) -> Any:
    """Build the query object."""
    return self._queries

RuntimeConfig dataclass

Bases: RuntimeConfig

Dataclass that contains all "dynamic" configurations of RedisDocumentIndex.

Parameters:

Name	Type	Description	Default
batch_size	int	Batch size for index/get/del.	100

Source code in docarray/index/backends/redis.py

@dataclass
class RuntimeConfig(BaseDocIndex.RuntimeConfig):
    """Dataclass that contains all "dynamic" configurations of RedisDocumentIndex.

    :param batch_size: Batch size for index/get/del.
    """

    batch_size: int = 100

__contains__(item)

Checks if a given document exists in the index.

Parameters:

Name	Type	Description	Default
item	BaseDoc	The document to check. It must be an instance of BaseDoc or its subclass.	required

Returns:

Type	Description
bool	True if the document exists in the index, False otherwise.

Source code in docarray/index/abstract.py

def __contains__(self, item: BaseDoc) -> bool:
    """
    Checks if a given document exists in the index.

    :param item: The document to check.
        It must be an instance of BaseDoc or its subclass.
    :return: True if the document exists in the index, False otherwise.
    """
    if safe_issubclass(type(item), BaseDoc):
        return self._doc_exists(str(item.id))
    else:
        raise TypeError(
            f"item must be an instance of BaseDoc or its subclass, not '{type(item).__name__}'"
        )

__delitem__(key)

Delete one or multiple Documents from the index, by id. If no document is found, a KeyError is raised.

Parameters:

Name	Type	Description	Default
key	Union[str, Sequence[str]]	id or ids to delete from the Document index	required

Source code in docarray/index/abstract.py

def __delitem__(self, key: Union[str, Sequence[str]]):
    """Delete one or multiple Documents from the index, by `id`.
    If no document is found, a KeyError is raised.

    :param key: id or ids to delete from the Document index
    """
    self._logger.info(f'Deleting documents with id(s) {key} from the index')
    if isinstance(key, str):
        key = [key]

    # delete nested data
    for field_name, type_, _ in self._flatten_schema(
        cast(Type[BaseDoc], self._schema)
    ):
        if safe_issubclass(type_, AnyDocArray):
            for doc_id in key:
                nested_docs_id = self._subindices[field_name]._filter_by_parent_id(
                    doc_id
                )
                if nested_docs_id:
                    del self._subindices[field_name][nested_docs_id]
    # delete data
    self._del_items(key)

__getitem__(key)

Get one or multiple Documents into the index, by id. If no document is found, a KeyError is raised.

Parameters:

Name	Type	Description	Default
key	Union[str, Sequence[str]]	id or ids to get from the Document index	required

Source code in docarray/index/abstract.py

def __getitem__(
    self, key: Union[str, Sequence[str]]
) -> Union[TSchema, DocList[TSchema]]:
    """Get one or multiple Documents into the index, by `id`.
    If no document is found, a KeyError is raised.

    :param key: id or ids to get from the Document index
    """
    # normalize input
    if isinstance(key, str):
        return_singleton = True
        key = [key]
    else:
        return_singleton = False

    # retrieve data
    doc_sequence = self._get_items(key)

    # check data
    if len(doc_sequence) == 0:
        raise KeyError(f'No document with id {key} found')

    # retrieve nested data
    for field_name, type_, _ in self._flatten_schema(
        cast(Type[BaseDoc], self._schema)
    ):
        if safe_issubclass(type_, AnyDocArray) and isinstance(
            doc_sequence[0], Dict
        ):
            for doc in doc_sequence:
                self._get_subindex_doclist(doc, field_name)  # type: ignore

    # cast output
    if isinstance(doc_sequence, DocList):
        out_docs: DocList[TSchema] = doc_sequence
    elif isinstance(doc_sequence[0], Dict):
        out_docs = self._dict_list_to_docarray(doc_sequence)  # type: ignore
    else:
        docs_cls = DocList.__class_getitem__(cast(Type[BaseDoc], self._schema))
        out_docs = docs_cls(doc_sequence)

    return out_docs[0] if return_singleton else out_docs

__init__(db_config=None, **kwargs)

Initialize RedisDocumentIndex

Source code in docarray/index/backends/redis.py

def __init__(self, db_config=None, **kwargs):
    """Initialize RedisDocumentIndex"""
    super().__init__(db_config=db_config, **kwargs)
    self._db_config = cast(RedisDocumentIndex.DBConfig, self._db_config)

    self._runtime_config: RedisDocumentIndex.RuntimeConfig = cast(
        RedisDocumentIndex.RuntimeConfig, self._runtime_config
    )
    self._prefix = self.index_name + ':'
    self._text_scorer = self._db_config.text_scorer
    # initialize Redis client
    self._client = redis.Redis(
        host=self._db_config.host,
        port=self._db_config.port,
        username=self._db_config.username,
        password=self._db_config.password,
        decode_responses=False,
    )
    self._create_index()
    self._logger.info(f'{self.__class__.__name__} has been initialized')

build_query()

Build a query for this DocumentIndex.

Returns:

Type	Description
QueryBuilder	a new QueryBuilder object for this DocumentIndex

Source code in docarray/index/abstract.py

def build_query(self) -> QueryBuilder:
    """
    Build a query for this DocumentIndex.

    :return: a new `QueryBuilder` object for this DocumentIndex
    """
    return self.QueryBuilder()  # type: ignore

configure(runtime_config=None, **kwargs)

Configure the DocumentIndex. You can either pass a config object to config or pass individual config parameters as keyword arguments. If a configuration object is passed, it will replace the current configuration. If keyword arguments are passed, they will update the current configuration.

Parameters:

Name	Type	Description	Default
runtime_config		the configuration to apply	None
kwargs		individual configuration parameters	{}

Source code in docarray/index/abstract.py

def configure(self, runtime_config=None, **kwargs):
    """
    Configure the DocumentIndex.
    You can either pass a config object to `config` or pass individual config
    parameters as keyword arguments.
    If a configuration object is passed, it will replace the current configuration.
    If keyword arguments are passed, they will update the current configuration.

    :param runtime_config: the configuration to apply
    :param kwargs: individual configuration parameters
    """
    if runtime_config is None:
        self._runtime_config = replace(self._runtime_config, **kwargs)
    else:
        if not isinstance(runtime_config, self.RuntimeConfig):
            raise ValueError(f'runtime_config must be of type {self.RuntimeConfig}')
        self._runtime_config = runtime_config

execute_query(query, *args, **kwargs)

Executes a hybrid query on the index.

Parameters:

Name	Type	Description	Default
query	Any	Query to execute on the index.	required

Returns:

Type	Description
Any	Query results.

Source code in docarray/index/backends/redis.py

def execute_query(self, query: Any, *args: Any, **kwargs: Any) -> Any:
    """
    Executes a hybrid query on the index.

    :param query: Query to execute on the index.
    :return: Query results.
    """
    components: Dict[str, List[Dict[str, Any]]] = {}
    for component, value in query:
        if component not in components:
            components[component] = []
        components[component].append(value)

    if (
        len(components) != 2
        or len(components.get('find', [])) != 1
        or len(components.get('filter', [])) != 1
    ):
        raise ValueError(
            'The query must contain exactly one "find" and "filter" components.'
        )

    filter_query = components['filter'][0]['filter_query']
    query = components['find'][0]['query']
    search_field = components['find'][0]['search_field']
    limit = (
        components['find'][0].get('limit')
        or components['filter'][0].get('limit')
        or 10
    )
    docs, scores = self._hybrid_search(
        query=query,
        filter_query=filter_query,
        search_field=search_field,
        limit=limit,
    )
    docs = self._dict_list_to_docarray(docs)
    return FindResult(documents=docs, scores=scores)

filter(filter_query, limit=10, **kwargs)

Find documents in the index based on a filter query

Parameters:

Name	Type	Description	Default
filter_query	Any	the DB specific filter query to execute	required
limit	int	maximum number of documents to return	10

Returns:

Type	Description
DocList	a DocList containing the documents that match the filter query

Source code in docarray/index/abstract.py

def filter(
    self,
    filter_query: Any,
    limit: int = 10,
    **kwargs,
) -> DocList:
    """Find documents in the index based on a filter query

    :param filter_query: the DB specific filter query to execute
    :param limit: maximum number of documents to return
    :return: a DocList containing the documents that match the filter query
    """
    self._logger.debug(f'Executing `filter` for the query {filter_query}')
    docs = self._filter(filter_query, limit=limit, **kwargs)

    if isinstance(docs, List) and not isinstance(docs, DocList):
        docs = self._dict_list_to_docarray(docs)

    return docs

filter_batched(filter_queries, limit=10, **kwargs)

Find documents in the index based on multiple filter queries.

Parameters:

Name	Type	Description	Default
filter_queries	Any	the DB specific filter query to execute	required
limit	int	maximum number of documents to return	10

Returns:

Type	Description
List[DocList]	a DocList containing the documents that match the filter query

Source code in docarray/index/abstract.py

def filter_batched(
    self,
    filter_queries: Any,
    limit: int = 10,
    **kwargs,
) -> List[DocList]:
    """Find documents in the index based on multiple filter queries.

    :param filter_queries: the DB specific filter query to execute
    :param limit: maximum number of documents to return
    :return: a DocList containing the documents that match the filter query
    """
    self._logger.debug(
        f'Executing `filter_batched` for the queries {filter_queries}'
    )
    da_list = self._filter_batched(filter_queries, limit=limit, **kwargs)

    if len(da_list) > 0 and isinstance(da_list[0], List):
        da_list = [self._dict_list_to_docarray(docs) for docs in da_list]

    return da_list  # type: ignore

filter_subindex(filter_query, subindex, limit=10, **kwargs)

Find documents in subindex level based on a filter query

Parameters:

Name	Type	Description	Default
filter_query	Any	the DB specific filter query to execute	required
subindex	str	name of the subindex to search on	required
limit	int	maximum number of documents to return	10

Returns:

Type	Description
DocList	a DocList containing the subindex level documents that match the filter query

Source code in docarray/index/abstract.py

def filter_subindex(
    self,
    filter_query: Any,
    subindex: str,
    limit: int = 10,
    **kwargs,
) -> DocList:
    """Find documents in subindex level based on a filter query

    :param filter_query: the DB specific filter query to execute
    :param subindex: name of the subindex to search on
    :param limit: maximum number of documents to return
    :return: a DocList containing the subindex level documents that match the filter query
    """
    self._logger.debug(
        f'Executing `filter` for the query {filter_query} in subindex {subindex}'
    )
    if '__' in subindex:
        fields = subindex.split('__')
        return self._subindices[fields[0]].filter_subindex(
            filter_query, '__'.join(fields[1:]), limit=limit, **kwargs
        )
    else:
        return self._subindices[subindex].filter(
            filter_query, limit=limit, **kwargs
        )

find(query, search_field='', limit=10, **kwargs)

Find documents in the index using nearest neighbor search.

Parameters:

Name	Type	Description	Default
query	Union[AnyTensor, BaseDoc]	query vector for KNN/ANN search. Can be either a tensor-like (np.array, torch.Tensor, etc.) with a single axis, or a Document	required
search_field	str	name of the field to search on. Documents in the index are retrieved based on this similarity of this field to the query.	''
limit	int	maximum number of documents to return	10

Returns:

Type	Description
FindResult	a named tuple containing documents and scores

Source code in docarray/index/abstract.py

def find(
    self,
    query: Union[AnyTensor, BaseDoc],
    search_field: str = '',
    limit: int = 10,
    **kwargs,
) -> FindResult:
    """Find documents in the index using nearest neighbor search.

    :param query: query vector for KNN/ANN search.
        Can be either a tensor-like (np.array, torch.Tensor, etc.)
        with a single axis, or a Document
    :param search_field: name of the field to search on.
        Documents in the index are retrieved based on this similarity
        of this field to the query.
    :param limit: maximum number of documents to return
    :return: a named tuple containing `documents` and `scores`
    """
    self._logger.debug(f'Executing `find` for search field {search_field}')

    self._validate_search_field(search_field)
    if isinstance(query, BaseDoc):
        query_vec = self._get_values_by_column([query], search_field)[0]
    else:
        query_vec = query
    query_vec_np = self._to_numpy(query_vec)
    docs, scores = self._find(
        query_vec_np, search_field=search_field, limit=limit, **kwargs
    )

    if isinstance(docs, List) and not isinstance(docs, DocList):
        docs = self._dict_list_to_docarray(docs)

    return FindResult(documents=docs, scores=scores)

find_batched(queries, search_field='', limit=10, **kwargs)

Find documents in the index using nearest neighbor search.

Parameters:

Name	Type	Description	Default
queries	Union[AnyTensor, DocList]	query vector for KNN/ANN search. Can be either a tensor-like (np.array, torch.Tensor, etc.) with a, or a DocList. If a tensor-like is passed, it should have shape (batch_size, vector_dim)	required
search_field	str	name of the field to search on. Documents in the index are retrieved based on this similarity of this field to the query.	''
limit	int	maximum number of documents to return per query	10

Returns:

Type	Description
FindResultBatched	a named tuple containing documents and scores

Source code in docarray/index/abstract.py

def find_batched(
    self,
    queries: Union[AnyTensor, DocList],
    search_field: str = '',
    limit: int = 10,
    **kwargs,
) -> FindResultBatched:
    """Find documents in the index using nearest neighbor search.

    :param queries: query vector for KNN/ANN search.
        Can be either a tensor-like (np.array, torch.Tensor, etc.) with a,
        or a DocList.
        If a tensor-like is passed, it should have shape (batch_size, vector_dim)
    :param search_field: name of the field to search on.
        Documents in the index are retrieved based on this similarity
        of this field to the query.
    :param limit: maximum number of documents to return per query
    :return: a named tuple containing `documents` and `scores`
    """
    self._logger.debug(f'Executing `find_batched` for search field {search_field}')

    if search_field:
        if '__' in search_field:
            fields = search_field.split('__')
            if safe_issubclass(self._schema._get_field_annotation(fields[0]), AnyDocArray):  # type: ignore
                return self._subindices[fields[0]].find_batched(
                    queries,
                    search_field='__'.join(fields[1:]),
                    limit=limit,
                    **kwargs,
                )

    self._validate_search_field(search_field)
    if isinstance(queries, Sequence):
        query_vec_list = self._get_values_by_column(queries, search_field)
        query_vec_np = np.stack(
            tuple(self._to_numpy(query_vec) for query_vec in query_vec_list)
        )
    else:
        query_vec_np = self._to_numpy(queries)

    da_list, scores = self._find_batched(
        query_vec_np, search_field=search_field, limit=limit, **kwargs
    )
    if (
        len(da_list) > 0
        and isinstance(da_list[0], List)
        and not isinstance(da_list[0], DocList)
    ):
        da_list = [self._dict_list_to_docarray(docs) for docs in da_list]

    return FindResultBatched(documents=da_list, scores=scores)  # type: ignore

find_subindex(query, subindex='', search_field='', limit=10, **kwargs)

Find documents in subindex level.

Parameters:

Name	Type	Description	Default
query	Union[AnyTensor, BaseDoc]	query vector for KNN/ANN search. Can be either a tensor-like (np.array, torch.Tensor, etc.) with a single axis, or a Document	required
subindex	str	name of the subindex to search on	''
search_field	str	name of the field to search on	''
limit	int	maximum number of documents to return	10

Returns:

Type	Description
SubindexFindResult	a named tuple containing root docs, subindex docs and scores

Source code in docarray/index/abstract.py

def find_subindex(
    self,
    query: Union[AnyTensor, BaseDoc],
    subindex: str = '',
    search_field: str = '',
    limit: int = 10,
    **kwargs,
) -> SubindexFindResult:
    """Find documents in subindex level.

    :param query: query vector for KNN/ANN search.
        Can be either a tensor-like (np.array, torch.Tensor, etc.)
        with a single axis, or a Document
    :param subindex: name of the subindex to search on
    :param search_field: name of the field to search on
    :param limit: maximum number of documents to return
    :return: a named tuple containing root docs, subindex docs and scores
    """
    self._logger.debug(f'Executing `find_subindex` for search field {search_field}')

    sub_docs, scores = self._find_subdocs(
        query, subindex=subindex, search_field=search_field, limit=limit, **kwargs
    )

    fields = subindex.split('__')
    root_ids = [
        self._get_root_doc_id(doc.id, fields[0], '__'.join(fields[1:]))
        for doc in sub_docs
    ]
    root_docs = DocList[self._schema]()  # type: ignore
    for id in root_ids:
        root_docs.append(self[id])

    return SubindexFindResult(
        root_documents=root_docs, sub_documents=sub_docs, scores=scores  # type: ignore
    )

index(docs, **kwargs)

index Documents into the index.

Note

Passing a sequence of Documents that is not a DocList (such as a List of Docs) comes at a performance penalty. This is because the Index needs to check compatibility between itself and the data. With a DocList as input this is a single check; for other inputs compatibility needs to be checked for every Document individually.

Parameters:

Name	Type	Description	Default
docs	Union[BaseDoc, Sequence[BaseDoc]]	Documents to index.	required

Source code in docarray/index/abstract.py

def index(self, docs: Union[BaseDoc, Sequence[BaseDoc]], **kwargs):
    """index Documents into the index.

    !!! note
        Passing a sequence of Documents that is not a DocList
        (such as a List of Docs) comes at a performance penalty.
        This is because the Index needs to check compatibility between itself and
        the data. With a DocList as input this is a single check; for other inputs
        compatibility needs to be checked for every Document individually.

    :param docs: Documents to index.
    """
    n_docs = 1 if isinstance(docs, BaseDoc) else len(docs)
    self._logger.debug(f'Indexing {n_docs} documents')
    docs_validated = self._validate_docs(docs)
    self._update_subindex_data(docs_validated)
    data_by_columns = self._get_col_value_dict(docs_validated)
    self._index(data_by_columns, **kwargs)

num_docs()

Fetch the number of documents in the index.

Returns:

Type	Description
int	Number of documents in the index.

Source code in docarray/index/backends/redis.py

def num_docs(self) -> int:
    """
    Fetch the number of documents in the index.

    :return: Number of documents in the index.
    """
    num_docs = self._client.ft(self.index_name).info()['num_docs']
    return int(num_docs)

python_type_to_db_type(python_type)

Map python types to corresponding Redis types.

Parameters:

Name	Type	Description	Default
python_type	Type	Python type.	required

Returns:

Type	Description
Any	Corresponding Redis type.

Source code in docarray/index/backends/redis.py

def python_type_to_db_type(self, python_type: Type) -> Any:
    """
    Map python types to corresponding Redis types.

    :param python_type: Python type.
    :return: Corresponding Redis type.
    """
    type_map = {
        int: NumericField,
        float: NumericField,
        str: TextField,
        bytes: TextField,
        np.ndarray: VectorField,
        list: VectorField,
        AbstractTensor: VectorField,
    }

    for py_type, redis_type in type_map.items():
        if safe_issubclass(python_type, py_type):
            return redis_type
    raise ValueError(f'Unsupported column type for {type(self)}: {python_type}')

subindex_contains(item)

Checks if a given BaseDoc item is contained in the index or any of its subindices.

Parameters:

Name	Type	Description	Default
item	BaseDoc	the given BaseDoc	required

Returns:

Type	Description
bool	if the given BaseDoc item is contained in the index/subindices

Source code in docarray/index/abstract.py

def subindex_contains(self, item: BaseDoc) -> bool:
    """Checks if a given BaseDoc item is contained in the index or any of its subindices.

    :param item: the given BaseDoc
    :return: if the given BaseDoc item is contained in the index/subindices
    """
    if self._is_index_empty:
        return False

    if safe_issubclass(type(item), BaseDoc):
        return self.__contains__(item) or any(
            index.subindex_contains(item) for index in self._subindices.values()
        )
    else:
        raise TypeError(
            f"item must be an instance of BaseDoc or its subclass, not '{type(item).__name__}'"
        )

text_search(query, search_field='', limit=10, **kwargs)

Find documents in the index based on a text search query.

Parameters:

Name	Type	Description	Default
query	Union[str, BaseDoc]	The text to search for	required
search_field	str	name of the field to search on	''
limit	int	maximum number of documents to return	10

Returns:

Type	Description
FindResult	a named tuple containing documents and scores

Source code in docarray/index/abstract.py

def text_search(
    self,
    query: Union[str, BaseDoc],
    search_field: str = '',
    limit: int = 10,
    **kwargs,
) -> FindResult:
    """Find documents in the index based on a text search query.

    :param query: The text to search for
    :param search_field: name of the field to search on
    :param limit: maximum number of documents to return
    :return: a named tuple containing `documents` and `scores`
    """
    self._logger.debug(f'Executing `text_search` for search field {search_field}')
    self._validate_search_field(search_field)
    if isinstance(query, BaseDoc):
        query_text = self._get_values_by_column([query], search_field)[0]
    else:
        query_text = query
    docs, scores = self._text_search(
        query_text, search_field=search_field, limit=limit, **kwargs
    )

    if isinstance(docs, List) and not isinstance(docs, DocList):
        docs = self._dict_list_to_docarray(docs)

    return FindResult(documents=docs, scores=scores)

text_search_batched(queries, search_field='', limit=10, **kwargs)

Find documents in the index based on a text search query.

Parameters:

Name	Type	Description	Default
queries	Union[Sequence[str], Sequence[BaseDoc]]	The texts to search for	required
search_field	str	name of the field to search on	''
limit	int	maximum number of documents to return	10

Returns:

Type	Description
FindResultBatched	a named tuple containing documents and scores

Source code in docarray/index/abstract.py

def text_search_batched(
    self,
    queries: Union[Sequence[str], Sequence[BaseDoc]],
    search_field: str = '',
    limit: int = 10,
    **kwargs,
) -> FindResultBatched:
    """Find documents in the index based on a text search query.

    :param queries: The texts to search for
    :param search_field: name of the field to search on
    :param limit: maximum number of documents to return
    :return: a named tuple containing `documents` and `scores`
    """
    self._logger.debug(
        f'Executing `text_search_batched` for search field {search_field}'
    )
    self._validate_search_field(search_field)
    if isinstance(queries[0], BaseDoc):
        query_docs: Sequence[BaseDoc] = cast(Sequence[BaseDoc], queries)
        query_texts: Sequence[str] = self._get_values_by_column(
            query_docs, search_field
        )
    else:
        query_texts = cast(Sequence[str], queries)
    da_list, scores = self._text_search_batched(
        query_texts, search_field=search_field, limit=limit, **kwargs
    )

    if len(da_list) > 0 and isinstance(da_list[0], List):
        docs = [self._dict_list_to_docarray(docs) for docs in da_list]
        return FindResultBatched(documents=docs, scores=scores)

    da_list_ = cast(List[DocList], da_list)
    return FindResultBatched(documents=da_list_, scores=scores)