import asyncio
import json
import threading
import time
import warnings
import weakref
from dataclasses import dataclass
from math import ceil
from typing import (
TYPE_CHECKING,
Any,
AsyncGenerator,
Callable,
Generator,
Iterable,
Iterator,
Sequence,
cast,
)
import redis.exceptions
# Add missing imports
from redis import Redis
from redis.asyncio import Redis as AsyncRedis
from redis.asyncio.cluster import RedisCluster as AsyncRedisCluster
from redis.cluster import RedisCluster
from redisvl.query.hybrid import HybridQuery
from redisvl.query.query import VectorQuery
from redisvl.query.sql import SQLQuery
from redisvl.redis.utils import (
_keys_share_hash_tag,
async_cluster_create_index,
async_cluster_search,
cluster_create_index,
cluster_search,
convert_bytes,
make_dict,
)
from redisvl.types import AsyncRedisClient, SyncRedisClient, SyncRedisCluster
from redisvl.utils.utils import deprecated_argument, deprecated_function, sync_wrapper
if TYPE_CHECKING:
from redis.commands.search.aggregation import AggregateResult
from redis.commands.search.document import Document
from redis.commands.search.result import Result
from redisvl.query.query import BaseQuery
from redis.client import NEVER_DECODE
from redis.commands.search.aggregation import AggregateRequest, Cursor
from redis.commands.search.index_definition import IndexDefinition
# Need Result outside TYPE_CHECKING for cast
from redis.commands.search.result import Result
from redisvl.exceptions import (
QueryValidationError,
RedisModuleVersionError,
RedisSearchError,
RedisVLError,
SchemaValidationError,
_is_missing_index_error,
)
from redisvl.index.storage import BaseStorage, HashStorage, JsonStorage
from redisvl.query import (
AggregationQuery,
BaseQuery,
BaseVectorQuery,
CountQuery,
FilterQuery,
TextQuery,
)
from redisvl.query.aggregate import AggregateHybridQuery
from redisvl.query.filter import FilterExpression
from redisvl.redis.connection import (
RedisConnectionFactory,
_split_from_existing_kwargs,
convert_index_info_to_schema,
supports_svs,
supports_svs_async,
)
from redisvl.redis.constants import SVS_MIN_REDIS_VERSION
from redisvl.schema import IndexSchema, StorageType
from redisvl.schema.fields import (
VECTOR_NORM_MAP,
SVSVectorField,
VectorDistanceMetric,
VectorIndexAlgorithm,
)
from redisvl.utils.log import get_logger
from redisvl.utils.redis_protocol import get_protocol_version
logger = get_logger(__name__)
def _parse_batch_search_result(
search: Any, result: Any, query: Any, duration: float
) -> Result:
"""Parse a pipelined FT.SEARCH response across supported redis-py versions."""
parsed_result = search._parse_results( # type: ignore
"FT.SEARCH", result, query=query, duration=duration
)
if not isinstance(parsed_result, dict):
return parsed_result
resp3_parser = getattr(search, "_parse_search_resp3", None)
if resp3_parser is not None:
return resp3_parser(parsed_result, query=query, duration=duration)
# redis-py 6.x returns the raw RESP3 map from _parse_results. Convert it to
# the RESP2 shape expected by its private _parse_search callback.
def value(mapping: dict[Any, Any], key: str, default: Any = None) -> Any:
return mapping.get(key, mapping.get(key.encode(), default))
response: list[Any] = [value(parsed_result, "total_results", 0)]
for document in value(parsed_result, "results", []):
response.append(value(document, "id", ""))
if query._with_scores:
response.append(value(document, "score", 0))
if query._with_payloads:
response.append(value(document, "payload"))
if not query._no_content:
fields = value(document, "extra_attributes", {})
response.append([item for pair in fields.items() for item in pair])
return search._parse_search(response, query=query, duration=duration)
_HYBRID_SEARCH_ERROR_MESSAGE = "Hybrid search is not available in this version of redis-py. Please upgrade to redis-py >= 7.1.0."
REQUIRED_MODULES_FOR_INTROSPECTION = [
{"name": "search", "ver": 20810},
{"name": "searchlight", "ver": 20810},
]
SearchParams = tuple[str | BaseQuery, dict[str, str | int | float | bytes] | None] | (
str | BaseQuery
)
def _get_sql_redis_options(sql_query: SQLQuery) -> dict[str, Any]:
"""Return normalized sql-redis executor options for a SQLQuery."""
return {
"schema_cache_strategy": "lazy",
**getattr(sql_query, "sql_redis_options", {}),
}
def _sql_executor_cache_key(sql_redis_options: dict[str, Any]) -> str:
"""Build a stable cache key for sql-redis executor reuse."""
return json.dumps(sql_redis_options, sort_keys=True, default=repr)
def _close_owned_sync_client(client: SyncRedisClient) -> None:
"""weakref.finalize callback that closes an index-owned sync client.
Must stay a module-level function taking the client as an argument: a
callback that references the index (e.g. a bound method) would be kept
alive by the finalizer registry and prevent the index from ever being
garbage collected.
"""
try:
client.close()
except Exception:
# Finalizers may run during interpreter shutdown; never propagate.
pass
def _close_owned_async_client(client: AsyncRedisClient) -> None:
"""weakref.finalize callback that closes an index-owned async client.
Same constraint as :func:`_close_owned_sync_client`: must not reference
the index instance.
"""
async def _aclose() -> None:
await client.aclose()
try:
sync_wrapper(_aclose)()
except Exception:
pass
# Default number of documents processed per round-trip in bulk operations.
DEFAULT_BULK_BATCH_SIZE = 500
# Idle timeout (seconds) applied to the FT.AGGREGATE cursor used by bulk
# updates, so an abandoned cursor is reclaimed server-side rather than
# lingering. redis-py's AggregateRequest.cursor(max_idle=...) takes seconds.
BULK_CURSOR_MAX_IDLE_SECONDS = 300
# update_by_filter resolves keys before it writes, so a key can be deleted by
# another client in between. These scripts make each write conditional on the
# key still existing, atomically — so a concurrently-deleted document is
# skipped rather than resurrected as a partial document. Each returns 1 if the
# document was written, 0 if it no longer existed.
_UPDATE_HASH_IF_EXISTS_LUA = """
if redis.call('EXISTS', KEYS[1]) == 0 then return 0 end
redis.call('HSET', KEYS[1], unpack(ARGV))
return 1
"""
_UPDATE_JSON_IF_EXISTS_LUA = """
if redis.call('EXISTS', KEYS[1]) == 0 then return 0 end
redis.call('JSON.MERGE', KEYS[1], '$', ARGV[1])
return 1
"""
@dataclass(frozen=True)
class BulkResult:
"""Outcome of a bulk operation (:meth:`SearchIndex.drop_by_filter` /
:meth:`SearchIndex.update_by_filter`).
Because bulk operations are not atomic across the match set, the result
distinguishes how many documents matched from how many were actually
processed, and whether the run ran to completion. Read the fields
explicitly (``result.processed`` / ``result.matched`` / ``result.completed``).
Attributes:
matched (int): Documents matching the filter when the run started.
processed (int): Documents actually deleted/updated (for ``dry_run``,
the number that *would* be processed).
completed (bool): False if the run stopped early (e.g. a delete hit its
runaway backstop under heavy concurrent inserts); re-run to finish.
dry_run (bool): True if this was a preview and nothing was mutated.
"""
matched: int
processed: int
completed: bool = True
dry_run: bool = False
def _is_match_all_filter(filter_expression: str | FilterExpression | None) -> bool:
"""Return True if the filter would match every document in the index.
Guards the bulk ``*_by_filter`` methods against an accidental full-index
wipe/update. ``None`` is treated as match-all because it defaults to
``FilterExpression("*")`` downstream; a default/empty ``FilterExpression``
(whose ``str()`` raises) is likewise treated as match-all rather than
surfacing an opaque error.
"""
if filter_expression is None:
return True
try:
rendered = str(filter_expression).strip()
except ValueError:
# Improperly initialized FilterExpression() - treat as the match-all sentinel
return True
return rendered in ("", "*")
def _require_specific_filter(
filter_expression: str | FilterExpression | None, allow_all: bool
) -> None:
"""Raise unless the filter is specific or the caller opted into match-all."""
if not allow_all and _is_match_all_filter(filter_expression):
raise ValueError(
"Refusing to run a bulk operation that matches all documents. "
"Pass a specific filter_expression, set allow_all=True to override, "
"or use clear() to intentionally remove every document."
)
def _agg_row_to_key(row: Any) -> str:
"""Extract the document key from a ``LOAD 1 @__key`` aggregation row.
Aggregation rows are a flat ``[field, value, field, value, ...]`` list, so
the key is the element immediately after the ``__key`` field name.
"""
decoded = convert_bytes(row)
key_field_index = decoded.index("__key")
return decoded[key_field_index + 1]
def _dedupe_keys(keys: Iterable[str], seen: set[str]) -> list[str]:
"""Filter ``keys`` down to those not already in ``seen``, updating ``seen``.
Repeats *within* ``keys`` are collapsed too, so one pass over a cursor page
yields each key at most once across the whole iteration. Cursor order falls
out of the implementation but nothing depends on it. See
:meth:`SearchIndex._iter_keys_by_filter` for why a cursor repeats keys.
"""
fresh = []
for key in keys:
if key not in seen:
seen.add(key)
fresh.append(key)
return fresh
def _update_script_and_args(
is_json: bool, values: dict[str, Any]
) -> tuple[str, list[Any]]:
"""Return the (Lua script, args) for an existence-guarded partial update.
Hash values are flattened into ``HSET`` field/value pairs; JSON values are
serialized into a single ``JSON.MERGE`` payload.
"""
if is_json:
return _UPDATE_JSON_IF_EXISTS_LUA, [json.dumps(values)]
flattened: list[Any] = []
for field, value in values.items():
flattened.extend((field, value))
return _UPDATE_HASH_IF_EXISTS_LUA, flattened
[docs]
class SearchResults(list):
"""A list of result documents that also reports result completeness.
This is a drop-in ``list`` — iteration, indexing, ``len()``, and every other
list operation behave exactly as before, so existing callers need no
changes. It additionally carries:
- ``dropped_count``: the number of matched documents that were skipped
because their field payload came back missing (the Redis 8.8+
background-search TTL/expiry race); ``0`` in the normal case.
- ``complete``: ``False`` when ``dropped_count > 0``, i.e. the result set may
be missing one or more matches that the server reported but could not
materialize.
Callers that need completeness guarantees — audit/compliance queries, eval
harnesses, or agents making control-flow decisions — can inspect these to
detect and react to a partial result (e.g. retry, reconcile against a
``CountQuery``, or annotate a downstream answer as incomplete). Everyone else
can ignore them. Note that operations returning a new list (slicing,
``sorted()``, concatenation) yield a plain ``list`` without this metadata.
"""
def __init__(self, iterable: Any = (), dropped_count: int = 0):
super().__init__(iterable)
self.dropped_count = dropped_count
@property
def complete(self) -> bool:
"""True when no matched documents were dropped from this result set."""
return self.dropped_count == 0
# Sentinel returned by the per-document processor to mark a matched document
# whose field payload came back missing (the Redis 8.8+ background-WORKERS
# TTL/expiry race). Such documents are filtered out of the final result list.
_SKIP_DOC = object()
def _has_missing_field_payload(
doc_dict: dict[str, Any], query: BaseQuery, unpack_json: bool
) -> bool:
"""Detect a matched document whose field payload is missing.
On Redis 8.8+ the RediSearch worker pool defaults to a nonzero background
executor. If a document key or an indexed hash field expires (TTL/HPEXPIRE)
exactly while a background ``FT.SEARCH`` is running, the server returns the
matched document id with a ``nil`` field array instead of dropping the
expired doc. redis-py collapses that ``nil`` to an empty field set, leaving a
``Document`` whose ``__dict__`` is only ``{"id": ..., "payload": None}``.
A race-victim doc is byte-for-byte identical to a *legitimate* id-only result
(e.g. ``FilterQuery(return_fields=["id"])``) or an ``INDEXMISSING`` sparse
doc, so we cannot key off "the dict has only id". We only report a missing
payload when the query type *guarantees* a field would be present on a
healthy match:
- vector/range queries always project ``DISTANCE_ID`` (``vector_distance``)
when ``return_score=True`` (the default), so its absence is unambiguous;
- the JSON full-object unpack path always sees a ``"json"`` key on a healthy
match.
Every other query passes through untouched.
A ``NOCONTENT`` query breaks both guarantees: the server then returns "the
document ids and not the content" for *every* healthy match (``RETURN 0`` acts
the same way), so no field payload is expected and its absence says nothing.
redis-py's ``Query.no_content()`` sets only ``_no_content`` and leaves
``_return_fields`` untouched, so neither predicate above notices -- without
this short-circuit every healthy document would be reported as missing.
"""
if getattr(query, "_no_content", False):
return False
if unpack_json:
return "json" not in doc_dict
if isinstance(query, BaseVectorQuery) and query.DISTANCE_ID in getattr(
query, "_return_fields", []
):
return query.DISTANCE_ID not in doc_dict
return False
def process_results(
results: "Result", query: BaseQuery, schema: IndexSchema
) -> list[dict[str, Any]]:
"""Convert a list of search Result objects into a list of document
dictionaries.
This function processes results from Redis, handling different storage
types and query types. For JSON storage with empty return fields, it
unpacks the JSON object while retaining the document ID. The 'payload'
field is also removed from all resulting documents for consistency.
Where it can be detected, a matched document whose field payload came back
missing is silently dropped (and logged at WARNING level). On Redis 8.8+ the
RediSearch worker pool defaults to a nonzero number of background worker
threads, and a document that expires (TTL/HPEXPIRE) exactly while a
background search runs is returned as a matched id with a ``nil`` field array
instead of being dropped server-side. This is only detectable for queries
that guarantee a field on a healthy match — vector/range queries with
``return_score=True`` (missing ``vector_distance``) and JSON full-object
unpack (missing ``json``). For a plain hash ``FilterQuery``/``TextQuery`` a
race victim is indistinguishable from a legitimately sparse/id-only result,
so it is passed through as an id-only dict rather than dropped.
Because of this, the returned list is best-effort and may contain fewer
items than ``num_results`` (or a paginated ``page_size``) when such docs are
present. ``CountQuery.total`` is unaffected — it reflects the server's match
count and can therefore exceed the number of materialized documents.
Args:
results (Result): The search results from Redis.
query (BaseQuery): The query object used for the search.
schema (IndexSchema): The index schema, used to determine storage type
and field metadata (e.g. the vector distance metric).
Returns:
List[Dict[str, Any]]: A list of processed document dictionaries.
"""
# Handle count queries
if isinstance(query, CountQuery):
return results.total
# Determine if unpacking JSON is needed
unpack_json = (
(schema.index.storage_type == StorageType.JSON)
and isinstance(query, FilterQuery)
and not query._return_fields # type: ignore
)
if (isinstance(query, BaseVectorQuery)) and query._normalize_vector_distance:
dist_metric = VectorDistanceMetric(
schema.fields[query._vector_field_name].attrs.distance_metric.upper() # type: ignore
)
if dist_metric == VectorDistanceMetric.IP:
warnings.warn(
"Attempting to normalize inner product distance metric. Use cosine distance instead which is normalized inner product by definition."
)
norm_fn = VECTOR_NORM_MAP[dist_metric.value]
else:
norm_fn = None
# Collect ids of documents skipped because their field payload was missing
# (the Redis 8.8+ background-WORKERS expiry race); warned once per query.
skipped_ids: list[Any] = []
# Process records
def _process(doc: "Document") -> Any:
doc_dict = doc.__dict__
# Skip matched documents whose field payload came back missing rather
# than raising (KeyError below) or leaking a partial record downstream.
if _has_missing_field_payload(doc_dict, query, unpack_json):
skipped_ids.append(doc_dict.get("id"))
return _SKIP_DOC
# Unpack and Project JSON fields properly. The skip guard above already
# dropped the unpack_json case where "json" is absent, so here it is
# guaranteed present.
if unpack_json:
json_data = doc_dict.get("json", {})
if isinstance(json_data, str):
json_data = json.loads(json_data)
if isinstance(json_data, dict):
return {"id": doc_dict.get("id"), **json_data}
raise ValueError(f"Unable to parse json data from Redis {json_data}")
# The skip guard above guarantees the distance is present for a normal
# vector query, but not for a NOCONTENT one, where the server returns ids
# only and there is no distance to normalize.
if norm_fn and query.DISTANCE_ID in doc_dict: # type: ignore
# convert float back to string to be consistent
doc_dict[query.DISTANCE_ID] = str( # type: ignore
norm_fn(float(doc_dict[query.DISTANCE_ID])) # type: ignore
)
# Remove 'payload' if present
doc_dict.pop("payload", None)
return doc_dict
processed = [_process(doc) for doc in results.docs]
if skipped_ids:
logger.warning(
"Skipped %d matched document(s) with missing field data (likely "
"expired during a background FT.SEARCH on Redis 8.8+). "
"query_type=%s index=%s ids=%s",
len(skipped_ids),
type(query).__name__,
schema.index.name,
skipped_ids[:10],
)
return SearchResults(
(doc for doc in processed if doc is not _SKIP_DOC),
dropped_count=len(skipped_ids),
)
def process_aggregate_results(
results: "AggregateResult", query: AggregationQuery, storage_type: StorageType
) -> list[dict[str, Any]]:
"""Convert an aggregate result object into a list of document dictionaries.
This function processes results from Redis, handling different storage
types and query types. For JSON storage with empty return fields, it
unpacks the JSON object while retaining the document ID. The 'payload'
field is also removed from all resulting documents for consistency.
Args:
results (AggregateResult): The aggregate results from Redis.
query (AggregationQuery): The aggregation query object used for the aggregation.
storage_type (StorageType): The storage type of the search
index (json or hash).
Returns:
List[Dict[str, Any]]: A list of processed document dictionaries.
"""
def _process(row):
result = make_dict(convert_bytes(row))
# Determine emptiness BEFORE removing __score: a row that carried only a
# score is a legitimate (if unusual) aggregation result and must be
# kept, whereas a row that came back with no fields at all is dropped.
had_content = bool(result)
result.pop("__score", None)
return result, had_content
processed = [_process(r) for r in results.rows]
kept = [result for result, had_content in processed if had_content]
dropped = len(processed) - len(kept)
if dropped:
logger.warning(
"Dropped %d empty aggregate row(s) with no field data.",
dropped,
)
return SearchResults(kept, dropped_count=dropped)
def _convert_and_drop_empty_rows(rows: Any, source: str) -> list[dict[str, Any]]:
"""Byte-decode result rows and drop any that came back entirely empty.
Used by the hybrid (FT.HYBRID) and SQL result paths, which decode rows with
``convert_bytes`` and perform no keyed field access (so they never raise).
An entirely-empty row carries no data and is dropped for behavioral
consistency with the FT.SEARCH/FT.AGGREGATE paths under the Redis 8.8+
background-search expiry race. Non-empty rows are returned unchanged.
"""
converted = [convert_bytes(r) for r in rows]
kept = [r for r in converted if r]
dropped = len(converted) - len(kept)
if dropped:
logger.warning(
"Dropped %d empty %s row(s) with no field data.",
dropped,
source,
)
return SearchResults(kept, dropped_count=dropped)
def _page_had_matches(results: Any) -> bool:
"""Whether the server reported any match for one page of paginated results.
``paginate`` must not treat "empty page" as "result set exhausted". A page
comes back empty for two very different reasons:
1. the server reported no more matches — iteration is genuinely done;
2. the server reported matches whose field payload could not be
materialized, so ``process_results`` dropped them -- a key that expires or
is updated mid-query is returned as a matched id with a ``nil`` field
array, and is still counted in the server's total (see
``_has_missing_field_payload``).
Stopping on case 2 silently truncates iteration and discards every remaining
page. ``SearchResults.dropped_count`` tells the two apart: a page had matches
when it either yielded documents or dropped some.
Truthiness (not ``len()``) is deliberate, and ``getattr`` is defensive: every
query path returns ``SearchResults`` today, but a subclass or test double may
substitute a plain ``list``, and ``process_results`` returns a bare ``int``
for a ``CountQuery``. Neither should raise here.
"""
return bool(results) or bool(getattr(results, "dropped_count", 0))
def _fold_carried_drops(results: Any, carried: int) -> None:
"""Add drops carried over from skipped pages to this page's ``dropped_count``.
``paginate`` never yields an empty batch, so a page whose matches were *all*
dropped would otherwise carry its ``dropped_count`` out of the stream and
leave the remaining batches reporting ``complete is True`` — the same silent
incompleteness the drop accounting exists to surface. Folding the count into
the next yielded batch keeps the signal reachable from the batches a caller
actually sees, without breaking the non-empty-batch guarantee.
Note the residual gap: if the *trailing* pages of a result set are entirely
dropped there is no subsequent batch to fold into, and those drops are
reported only by the ``process_results`` warning.
"""
if carried and isinstance(results, SearchResults):
results.dropped_count += carried
class BaseSearchIndex:
"""Base search engine class"""
_STORAGE_MAP = {
StorageType.HASH: HashStorage,
StorageType.JSON: JsonStorage,
}
schema: IndexSchema
# Module-level callback (set per subclass) used to close an owned client
# when the index is garbage collected. Must never be a bound method.
_finalizer_close_client: Callable[[Any], None]
_client_finalizer: "weakref.finalize | None" = None
def __init__(*args, **kwargs):
pass
def _register_client_finalizer(self, client: Any) -> None:
"""Register a finalizer that closes ``client`` once this index is
garbage collected.
Call this at every point where the index creates or takes over a
client it owns. The finalizer callback must only capture the client,
never ``self``: ``weakref.finalize`` keeps the callback alive in a
module-level registry until it fires, so a callback holding the
instance (a bound method, or any closure over ``self``) would keep
the instance reachable forever and the finalizer would never run.
"""
self._detach_client_finalizer()
if client is not None and getattr(self, "_owns_redis_client", False):
self._client_finalizer = weakref.finalize(
self, type(self)._finalizer_close_client, client
)
def _detach_client_finalizer(self) -> None:
"""Cancel the pending client finalizer, if any.
Called when the owned client is closed or replaced so the finalizer
neither double-closes a client nor keeps a replaced client alive.
"""
finalizer = self._client_finalizer
if finalizer is not None:
finalizer.detach()
self._client_finalizer = None
@property
def _storage(self) -> BaseStorage:
"""The storage type for the index schema."""
return self._STORAGE_MAP[self.schema.index.storage_type](
index_schema=self.schema
)
def _uses_svs_vamana(self) -> bool:
"""Check if schema contains any SVS-VAMANA vector fields."""
return any(
isinstance(field, SVSVectorField) for field in self.schema.fields.values()
)
def _validate_query(self, query: BaseQuery) -> None:
"""Validate a query."""
if isinstance(query, VectorQuery):
field = self.schema.fields[query._vector_field_name]
if query.ef_runtime and field.attrs.algorithm != VectorIndexAlgorithm.HNSW: # type: ignore
raise QueryValidationError(
"Vector field using 'flat' algorithm does not support EF_RUNTIME query parameter."
)
# Warn if using query-time stopwords with index-level STOPWORDS 0
if isinstance(query, (TextQuery, AggregateHybridQuery)):
index_stopwords = self.schema.index.stopwords
query_stopwords = query.stopwords
# Check if index has STOPWORDS 0 (empty list) and query has stopwords configured
# Note: query.stopwords is a set, and when any falsy value (None, False, '', 0, [], etc.)
# is passed to TextQuery/AggregateHybridQuery, it becomes an empty set. So we check if the set is non-empty.
if (
index_stopwords is not None
and len(index_stopwords) == 0
and len(query_stopwords) > 0
):
query_type = (
"TextQuery"
if isinstance(query, TextQuery)
else "AggregateHybridQuery"
)
warnings.warn(
f"Query-time stopwords are configured but the index has STOPWORDS 0 (stopwords = []). "
"This is counterproductive: all words including common words like 'of', 'the', 'a' are indexed, "
"but your query-time stopwords will filter them from the search query. "
"This makes your search less precise than it could be. "
f"Consider setting stopwords=None (or any falsy value) in {query_type} to search for all indexed words.",
UserWarning,
stacklevel=3,
)
def _validate_hybrid_query(self, query: Any) -> None:
"""Validate that a hybrid query can be executed."""
try:
from redis.commands.search.hybrid_result import ( # noqa: F401
HybridResult as _HybridResult,
)
from redisvl.query.hybrid import HybridQuery
del _HybridResult # Only imported to check availability
except (ImportError, ModuleNotFoundError):
raise ImportError(_HYBRID_SEARCH_ERROR_MESSAGE)
if not isinstance(query, HybridQuery):
raise TypeError(f"query must be of type HybridQuery, got {type(query)}")
@property
def name(self) -> str:
"""The name of the Redis search index."""
return self.schema.index.name
@property
def prefix(self) -> str:
"""The key prefix used in forming Redis keys.
For multi-prefix indexes, returns the first prefix.
"""
prefix = self.schema.index.prefix
return prefix[0] if isinstance(prefix, list) else prefix
@property
def prefixes(self) -> list[str]:
"""All key prefixes configured for this index."""
prefix = self.schema.index.prefix
return prefix if isinstance(prefix, list) else [prefix]
@property
def key_separator(self) -> str:
"""The optional separator between a defined prefix and key value in
forming a Redis key."""
return self.schema.index.key_separator
@property
def storage_type(self) -> StorageType:
"""The underlying storage type for the search index; either
hash or json."""
return self.schema.index.storage_type
@classmethod
def from_yaml(cls, schema_path: str, **kwargs):
"""Create a SearchIndex from a YAML schema file.
Args:
schema_path (str): Path to the YAML schema file.
Returns:
SearchIndex: A RedisVL SearchIndex object.
.. code-block:: python
from redisvl.index import SearchIndex
index = SearchIndex.from_yaml("schemas/schema.yaml", redis_url="redis://localhost:6379")
"""
schema = IndexSchema.from_yaml(schema_path)
return cls(schema=schema, **kwargs)
@classmethod
def from_dict(cls, schema_dict: dict[str, Any], **kwargs):
"""Create a SearchIndex from a dictionary.
Args:
schema_dict (Dict[str, Any]): A dictionary containing the schema.
Returns:
SearchIndex: A RedisVL SearchIndex object.
.. code-block:: python
from redisvl.index import SearchIndex
index = SearchIndex.from_dict({
"index": {
"name": "my-index",
"prefix": "rvl",
"storage_type": "hash",
},
"fields": [
{"name": "doc-id", "type": "tag"}
]
}, redis_url="redis://localhost:6379")
"""
schema = IndexSchema.from_dict(schema_dict)
return cls(schema=schema, **kwargs)
def disconnect(self):
"""Disconnect from the Redis database."""
raise NotImplementedError("This method should be implemented by subclasses.")
def key(self, id: str) -> str:
"""Construct a redis key as a combination of an index key prefix (optional)
and specified id.
The id is typically either a unique identifier, or
derived from some domain-specific metadata combination (like a document
id or chunk id).
Args:
id (str): The specified unique identifier for a particular
document indexed in Redis.
Returns:
str: The full Redis key including key prefix and value as a string.
"""
return self._storage._key(
id=id,
prefix=self.prefix,
key_separator=self.schema.index.key_separator,
)
def __repr__(self) -> str:
return (
f"{type(self).__name__}(name={self.name!r}, prefixes={self.prefixes!r}, "
f"storage_type={self.storage_type.value!r})"
)
[docs]
class SearchIndex(BaseSearchIndex):
"""A search index class for interacting with Redis as a vector database.
The SearchIndex is instantiated with a reference to a Redis database and an
IndexSchema (YAML path or dictionary object) that describes the various
settings and field configurations.
.. code-block:: python
from redisvl.index import SearchIndex
# initialize the index object with schema from file
index = SearchIndex.from_yaml(
"schemas/schema.yaml",
redis_url="redis://localhost:6379",
validate_on_load=True
)
# create the index
index.create(overwrite=True, drop=False)
# data is an iterable of dictionaries
index.load(data)
# delete index and data
index.delete(drop=True)
"""
@deprecated_argument("connection_args", "Use connection_kwargs instead.")
def __init__(
self,
schema: IndexSchema,
redis_client: SyncRedisClient | None = None,
redis_url: str | None = None,
connection_kwargs: dict[str, Any] | None = None,
validate_on_load: bool = False,
**kwargs,
):
"""Initialize the RedisVL search index with a schema, Redis client
(or URL string with other connection args), connection_args, and other
kwargs.
Args:
schema (IndexSchema): Index schema object.
redis_client(Optional[Redis]): An
instantiated redis client.
redis_url (Optional[str]): The URL of the Redis server to
connect to.
connection_kwargs (Dict[str, Any], optional): Redis client connection
args.
validate_on_load (bool, optional): Whether to validate data against schema
when loading. Defaults to False.
"""
if "connection_args" in kwargs:
connection_kwargs = kwargs.pop("connection_args")
if not isinstance(schema, IndexSchema):
raise ValueError("Must provide a valid IndexSchema object")
self.schema = schema
self._validate_on_load = validate_on_load
self._lib_name: str | None = kwargs.pop("lib_name", None)
# Store connection parameters
self.__redis_client = redis_client
self._redis_url = redis_url
self._connection_kwargs = connection_kwargs or {}
self._lock = threading.Lock()
self._sql_executors: dict[str, Any] = {}
self._validated_client = kwargs.pop("_client_validated", False)
self._owns_redis_client = kwargs.pop("_owns_redis_client", redis_client is None)
self._client_finalizer = None
# Close the owned client when this index is garbage collected. When
# the client is created lazily, registration happens at creation time
# instead (see the _redis_client property).
self._register_client_finalizer(redis_client)
_finalizer_close_client = staticmethod(_close_owned_sync_client)
[docs]
def disconnect(self):
"""Disconnect from the Redis database."""
self.invalidate_sql_schema_cache()
if self._owns_redis_client is False:
logger.info("Index does not own client, not disconnecting")
return
self._detach_client_finalizer()
if self.__redis_client:
self.__redis_client.close()
self.__redis_client = None
[docs]
@classmethod
def from_existing(
cls,
name: str,
redis_client: SyncRedisClient | None = None,
redis_url: str | None = None,
**kwargs,
):
"""
Initialize from an existing search index in Redis by index name.
Args:
name (str): Name of the search index in Redis.
redis_client(Optional[Redis]): An
instantiated redis client.
redis_url (Optional[str]): The URL of the Redis server to
connect to.
Raises:
ValueError: If redis_url or redis_client is not provided.
"""
init_kwargs, connection_kwargs = _split_from_existing_kwargs(
dict(kwargs),
nested_connection_keys=("connection_kwargs", "connection_args"),
)
lib_name = cast(str | None, init_kwargs.get("lib_name"))
created_redis_client = False
if redis_client:
# Validate client type and set lib name
RedisConnectionFactory.validate_sync_redis(redis_client, lib_name)
# Mark that client was already validated to avoid duplicate calls
init_kwargs["_client_validated"] = True
elif redis_url:
factory_kwargs = {**connection_kwargs}
if lib_name is not None:
factory_kwargs["lib_name"] = lib_name
redis_client = RedisConnectionFactory.get_redis_connection(
redis_url=redis_url,
**factory_kwargs,
)
init_kwargs["_client_validated"] = True
created_redis_client = True
if not redis_client:
raise ValueError("Must provide either a redis_url or redis_client")
# Fetch index info and convert to schema
index_info = cls._info(name, redis_client)
schema_dict = convert_index_info_to_schema(index_info)
schema = IndexSchema.from_dict(schema_dict)
if created_redis_client:
init_kwargs["_owns_redis_client"] = True
return cls(
schema,
redis_client=redis_client,
redis_url=redis_url,
connection_kwargs=connection_kwargs or None,
**init_kwargs,
)
return cls(schema, redis_client=redis_client, **init_kwargs)
@property
def client(self) -> SyncRedisClient | None:
"""The underlying redis-py client object."""
return self.__redis_client
@property
def _redis_client(self) -> SyncRedisClient:
"""
Get a Redis client instance.
Lazily creates a Redis client instance if it doesn't exist.
"""
if self.__redis_client is None:
with self._lock:
if self.__redis_client is None:
# Pass lib_name to connection factory
kwargs = {**self._connection_kwargs}
if self._lib_name:
kwargs["lib_name"] = self._lib_name
self.__redis_client = RedisConnectionFactory.get_redis_connection(
redis_url=self._redis_url,
**kwargs,
)
self._register_client_finalizer(self.__redis_client)
if not self._validated_client and self._lib_name:
# Only set lib name for user-provided clients
RedisConnectionFactory.validate_sync_redis(
self.__redis_client,
self._lib_name,
)
self._validated_client = True
return self.__redis_client
[docs]
@deprecated_function("connect", "Pass connection parameters in __init__.")
def connect(self, redis_url: str | None = None, **kwargs):
"""Connect to a Redis instance using the provided `redis_url`, falling
back to the `REDIS_URL` environment variable (if available).
Note: Additional keyword arguments (`**kwargs`) can be used to provide
extra options specific to the Redis connection.
Args:
redis_url (Optional[str], optional): The URL of the Redis server to
connect to.
Raises:
redis.exceptions.ConnectionError: If the connection to the Redis
server fails.
ValueError: If the Redis URL is not provided nor accessible
through the `REDIS_URL` environment variable.
ModuleNotFoundError: If required Redis modules are not installed.
"""
self.invalidate_sql_schema_cache()
self.__redis_client = RedisConnectionFactory.get_redis_connection(
redis_url=redis_url, **kwargs
)
self._register_client_finalizer(self.__redis_client)
[docs]
@deprecated_function("set_client", "Pass connection parameters in __init__.")
def set_client(self, redis_client: SyncRedisClient, **kwargs):
"""Manually set the Redis client to use with the search index.
This method configures the search index to use a specific Redis or
Async Redis client. It is useful for cases where an external,
custom-configured client is preferred instead of creating a new one.
Args:
redis_client (Redis): A Redis or Async Redis
client instance to be used for the connection.
Raises:
TypeError: If the provided client is not valid.
"""
RedisConnectionFactory.validate_sync_redis(redis_client)
self.invalidate_sql_schema_cache()
self.__redis_client = redis_client
self._register_client_finalizer(redis_client)
return self
def _check_svs_support(self) -> None:
"""Validate SVS-VAMANA support.
Raises:
RedisModuleVersionError: If SVS-VAMANA requirements are not met.
"""
if not supports_svs(self._redis_client):
raise RedisModuleVersionError.for_svs_vamana(SVS_MIN_REDIS_VERSION)
[docs]
def create(self, overwrite: bool = False, drop: bool = False) -> None:
"""Create an index in Redis with the current schema and properties.
Args:
overwrite (bool, optional): Whether to overwrite the index if it
already exists. Defaults to False.
drop (bool, optional): Whether to drop all keys associated with the
index in the case of overwriting. Defaults to False.
Raises:
RuntimeError: If the index already exists and 'overwrite' is False.
ValueError: If no fields are defined for the index.
.. code-block:: python
# create an index in Redis; only if one does not exist with given name
index.create()
# overwrite an index in Redis without dropping associated data
index.create(overwrite=True)
# overwrite an index in Redis; drop associated data (clean slate)
index.create(overwrite=True, drop=True)
"""
# Check that fields are defined.
redis_fields = self.schema.redis_fields
if not redis_fields:
raise ValueError("No fields defined for index")
if not isinstance(overwrite, bool):
raise TypeError("overwrite must be of type bool")
# Check if schema uses SVS-VAMANA and validate Redis capabilities
if self._uses_svs_vamana():
self._check_svs_support()
if self.exists():
if not overwrite:
logger.info("Index already exists, not overwriting.")
return None
logger.info("Index already exists, overwriting.")
self.delete(drop=drop)
try:
# Prefix can be a string or list of strings
prefix = self.schema.index.prefix
prefix_list = prefix if isinstance(prefix, list) else [prefix]
definition = IndexDefinition(
prefix=prefix_list, index_type=self._storage.type
)
# Extract stopwords from schema
stopwords = self.schema.index.stopwords
if isinstance(self._redis_client, RedisCluster):
cluster_create_index(
index_name=self.name,
client=self._redis_client,
fields=redis_fields,
definition=definition,
stopwords=stopwords,
)
else:
self._redis_client.ft(self.name).create_index(
fields=redis_fields,
definition=definition,
stopwords=stopwords,
)
self.invalidate_sql_schema_cache()
except redis.exceptions.RedisError as e:
raise RedisSearchError(
f"Failed to create index '{self.name}' on Redis: {str(e)}"
) from e
except Exception as e:
logger.exception("Error while trying to create the index")
raise RedisSearchError(
f"Unexpected error creating index '{self.name}': {str(e)}"
) from e
[docs]
def delete(self, drop: bool = True):
"""Delete the search index while optionally dropping all keys associated
with the index.
``FT.DROPINDEX`` resolves index aliases, so if this schema's name happens
to be an alias, the index the alias points at is dropped instead -- along
with its documents when ``drop`` is True. :meth:`exists` reports an alias
as absent to keep :meth:`create` away from this path.
Args:
drop (bool, optional): Delete the key / documents pairs in the
index. Defaults to True.
Raises:
RedisSearchError: If the index cannot be deleted, for example when it
does not exist. The original ``redis-py`` exception is available
as ``__cause__``.
"""
try:
# For Redis Cluster with drop=True, we need to handle key deletion manually
# to avoid cross-slot errors since we control the keys opaquely
if drop and isinstance(self._redis_client, RedisCluster):
# First clear all keys manually (handles cluster compatibility)
self.clear()
# Then drop the index without the DD flag
cmd_args = ["FT.DROPINDEX", self.schema.index.name]
else:
# Standard approach for non-cluster or when not dropping keys
cmd_args = ["FT.DROPINDEX", self.schema.index.name]
if drop:
cmd_args.append("DD")
if isinstance(self._redis_client, RedisCluster):
target_nodes = [self._redis_client.get_default_node()]
self._redis_client.execute_command(*cmd_args, target_nodes=target_nodes)
else:
self._redis_client.execute_command(*cmd_args)
self.invalidate_sql_schema_cache()
except Exception as e:
raise RedisSearchError(f"Error while deleting index: {str(e)}") from e
def _delete_batch(self, batch_keys: list[str]) -> int:
"""Delete a batch of keys from Redis.
For Redis Cluster, keys are deleted individually due to potential
cross-slot limitations. For standalone Redis, keys are deleted in
a single operation for better performance.
Args:
batch_keys (List[str]): List of Redis keys to delete.
Returns:
int: Count of records deleted from Redis.
"""
client = cast(SyncRedisClient, self._redis_client)
is_cluster = isinstance(client, RedisCluster)
if is_cluster:
records_deleted_in_batch = 0
for key_to_delete in batch_keys:
try:
records_deleted_in_batch += cast(int, client.delete(key_to_delete))
except redis.exceptions.RedisError as e:
logger.warning(f"Failed to delete key {key_to_delete}: {e}")
else:
records_deleted_in_batch = cast(int, client.delete(*batch_keys))
return records_deleted_in_batch
[docs]
def clear(self) -> int:
"""Clear all keys in Redis associated with the index, leaving the index
available and in-place for future insertions or updates.
NOTE: This method requires custom behavior for Redis Cluster because
here, we can't easily give control of the keys we're clearing to the
user so they can separate them based on hash tag.
Note:
The sweep enumerates through the index, so it removes only what the
index currently returns, and it can stop early -- against an index
still being backfilled, or when a page's keys cannot be deleted. The
returned count is the only signal, and ``0`` does not distinguish an
empty index from a sweep that deleted nothing. Re-running is safe.
Returns:
int: Count of records deleted from Redis.
"""
batch_size = 500
matched = cast(int, self.query(CountQuery(FilterExpression("*"))))
# Runaway backstop sized to the matched count plus slack for concurrent
# inserts, as in drop_by_filter. Deliberately not FT.INFO's num_docs:
# that command is @search only, so reading it for a loop bound denied
# this whole method to a `+@read +@write` credential.
max_records = ceil(matched * 1.5) + batch_size
total_records_deleted: int = 0
offset = 0
query = FilterQuery(FilterExpression("*"), return_fields=["id"])
while True:
if total_records_deleted > max_records:
logger.warning(
"clear() of index %s hit its runaway backstop (%d) with "
"documents possibly still indexed; %d records were deleted. "
"Re-run to continue.",
self.schema.index.name,
max_records,
total_records_deleted,
)
break
query.paging(offset, batch_size)
batch = self._query(query)
if not batch:
break
batch_keys = [record["id"] for record in batch]
records_deleted = self._delete_batch(batch_keys)
total_records_deleted += records_deleted
if records_deleted:
# Deleted documents leave the index, so the next page of
# survivors is at offset 0 again.
offset = 0
else:
# Nothing in this page could be deleted, most plausibly a
# permission denial swallowed by _delete_batch's cluster branch.
# Page past it: the documents behind may still be deletable.
offset += batch_size
if offset > max_records:
logger.warning(
"clear() of index %s paged past its runaway backstop "
"(%d) without being able to delete; %d records were "
"deleted. Documents remain.",
self.schema.index.name,
max_records,
total_records_deleted,
)
break
self.invalidate_sql_schema_cache()
return total_records_deleted
[docs]
def invalidate_sql_schema_cache(self) -> None:
"""Clear cached sql-redis executors and schema state for this index."""
self._sql_executors.clear()
def _unlink_batch(self, batch_keys: list[str]) -> int:
"""Unlink a batch of keys from Redis.
Mirrors ``_delete_batch`` but uses non-blocking ``UNLINK``. For Redis
Cluster, keys are unlinked individually to avoid cross-slot errors;
for standalone Redis a single variadic ``UNLINK`` is used.
Args:
batch_keys (List[str]): List of Redis keys to unlink.
Returns:
int: Count of records unlinked from Redis.
"""
client = cast(SyncRedisClient, self._redis_client)
if isinstance(client, RedisCluster):
unlinked = 0
for key_to_unlink in batch_keys:
try:
unlinked += cast(int, client.unlink(key_to_unlink))
except redis.exceptions.RedisError as e:
logger.warning(f"Failed to unlink key {key_to_unlink}: {e}")
return unlinked
return cast(int, client.unlink(*batch_keys))
[docs]
def drop_keys(
self, keys: str | list[str], batch_size: int = DEFAULT_BULK_BATCH_SIZE
) -> int:
"""Remove a specific entry or entries from the index by it's key ID.
Uses ``UNLINK`` rather than ``DEL`` so memory reclamation runs on a
background thread. This avoids blocking the main thread when a large
number of keys are dropped at once (for example, scope-targeted
``SemanticCache`` invalidation). The returned count is unchanged.
Large key lists are unlinked in chunks of ``batch_size``. On Redis
Cluster, keys are unlinked individually so a chunk that spans hash
slots does not raise ``CROSSSLOT``.
Args:
keys (Union[str, List[str]]): The document ID or IDs to remove from the index.
batch_size (int): Number of keys to unlink per round-trip.
Returns:
int: Count of records deleted from Redis.
"""
if not isinstance(keys, list):
return self._redis_client.unlink(keys) # type: ignore
if not keys:
return 0
total = 0
for i in range(0, len(keys), batch_size):
total += self._unlink_batch(keys[i : i + batch_size])
return total
[docs]
def drop_documents(
self, ids: str | list[str], batch_size: int = DEFAULT_BULK_BATCH_SIZE
) -> int:
"""Remove documents from the index by their document IDs.
This method converts document IDs to Redis keys automatically by applying
the index's key prefix and separator configuration.
NOTE: Cluster users will need to incorporate hash tags into their
document IDs and only call this method with documents from a single hash
tag at a time.
Args:
ids (Union[str, List[str]]): The document ID or IDs to remove from the index.
batch_size (int): Number of documents to delete per round-trip
(standalone Redis only; cluster deletes in a single call after
the shared-hash-tag check).
Returns:
int: Count of documents deleted from Redis.
Raises:
ValueError: On Redis Cluster, if the resolved keys do not all share
a hash tag (a cross-slot ``DELETE`` is not permitted).
"""
if not isinstance(ids, list):
key = self.key(ids)
return self._redis_client.delete(key) # type: ignore
if not ids:
return 0
keys = [self.key(id) for id in ids]
# On cluster, a multi-key DELETE must stay within one hash slot, so
# require the keys to share a hash tag and delete them in a single call.
if isinstance(self._redis_client, RedisCluster):
if not _keys_share_hash_tag(keys):
raise ValueError(
"All keys must share a hash tag when using Redis Cluster."
)
return self._redis_client.delete(*keys) # type: ignore
# Standalone: delete in chunks to bound the size of any single command.
total = 0
for i in range(0, len(keys), batch_size):
batch = keys[i : i + batch_size]
total += cast(int, self._redis_client.delete(*batch))
return total
[docs]
def expire_keys(self, keys: str | list[str], ttl: int) -> int | list[int]:
"""Set the expiration time for a specific entry or entries in Redis.
Args:
keys (Union[str, List[str]]): The entry ID or IDs to set the expiration for.
ttl (int): The time-to-live in seconds.
"""
if isinstance(keys, list):
pipe = self._redis_client.pipeline() # type: ignore
for key in keys:
pipe.expire(key, ttl)
return pipe.execute()
else:
return self._redis_client.expire(keys, ttl) # type: ignore
[docs]
def drop_by_filter(
self,
filter_expression: str | FilterExpression,
*,
batch_size: int = DEFAULT_BULK_BATCH_SIZE,
dry_run: bool = False,
allow_all: bool = False,
on_progress: Callable[[int, int], None] | None = None,
) -> BulkResult:
"""Delete every document matching a filter expression.
Redis has no server-side "delete by query", so RedisVL resolves the
matching document keys and removes them with non-blocking ``UNLINK`` in
batches. Matching documents leave the result set as they are deleted, so
this re-queries from offset 0 each round (the same strategy as
:meth:`clear`) and is not subject to the ``MAXSEARCHRESULTS`` limit
(provided ``batch_size`` itself does not exceed it).
Args:
filter_expression (Union[str, FilterExpression]): Selects the
documents to delete. Prefer the escaping builders (``Tag``,
``Num``, ``Text``...) over raw filter strings; **never
string-concatenate untrusted input into a filter** — unlike a
read query, an injected predicate here deletes data.
batch_size (int): Number of documents to resolve and unlink per
round-trip. Defaults to 500.
dry_run (bool): If True, report how many documents *would* be
deleted (via a count query) without deleting anything.
allow_all (bool): Must be True to run against a match-all filter
(empty/``"*"``/``None``). Prefer :meth:`clear` to intentionally
empty the index.
on_progress (Optional[Callable[[int, int], None]]): Called after each
batch with ``(processed, matched)`` — the cumulative documents
deleted and the total matched at the start. Invoked
synchronously (do not pass a coroutine); raising from it aborts
the run (already-deleted documents stay deleted).
Returns:
BulkResult: ``matched``/``processed`` counts, plus ``completed``
(False if the runaway backstop tripped) and ``dry_run``. Read the
fields explicitly (``result.processed``, ``result.completed``).
See Also:
:meth:`update_by_filter` (bulk partial update), :meth:`drop_documents`
/ :meth:`drop_keys` (delete by id/key), :meth:`clear` (delete all).
Note:
This operation is **not atomic** across the match set. Each key is
unlinked atomically, but batches are applied incrementally with no
rollback, so a crash or connection error mid-run leaves the already
-deleted documents gone and the rest in place. Deletes are
idempotent: re-running the same call after a failure removes only
whatever still matches, converging on the intended state.
"""
_require_specific_filter(filter_expression, allow_all)
matched = cast(int, self.query(CountQuery(filter_expression)))
if dry_run:
return BulkResult(
matched=matched, processed=matched, completed=True, dry_run=True
)
# Runaway backstop sized to the matched count (plus slack for concurrent
# inserts); normal termination is an empty offset-0 re-query.
max_records = ceil(matched * 1.5) + batch_size
total_deleted = 0
completed = True
query = FilterQuery(filter_expression, return_fields=["id"])
query.paging(0, batch_size)
while True:
if total_deleted > max_records:
logger.warning(
"drop_by_filter hit its runaway backstop (%d) with documents "
"possibly still matching; returning an incomplete result. "
"Re-run to continue.",
max_records,
)
completed = False
break
batch = self._query(query)
if not batch:
break
batch_keys = [record["id"] for record in batch]
total_deleted += self._unlink_batch(batch_keys)
if on_progress is not None:
on_progress(total_deleted, matched)
self.invalidate_sql_schema_cache()
return BulkResult(matched=matched, processed=total_deleted, completed=completed)
[docs]
def update_by_filter(
self,
filter_expression: str | FilterExpression,
values: dict[str, Any],
*,
batch_size: int = DEFAULT_BULK_BATCH_SIZE,
dry_run: bool = False,
allow_all: bool = False,
on_progress: Callable[[int, int], None] | None = None,
) -> BulkResult:
"""Set ``values`` on every document matching a filter expression.
This is a partial update: fields not present in ``values`` are left
untouched. For hash indexes the fields are written with ``HSET``; for
JSON indexes they are merged at the document root (``$``) with
``JSON.MERGE`` (RFC 7396), so nested objects merge recursively, arrays
are replaced wholesale, and a ``None`` value deletes that path.
The read phase (an ``FT.AGGREGATE`` cursor) is drained in full *before*
any write, holding every matching key in memory in between. Interleaving
the two does not terminate: RediSearch has no in-place update, so each
write appends a new, higher document id ahead of the cursor, which hands
that document back to be written again. Memory use is therefore
proportional to the match count; for very large match sets, narrow the
filter and run in partitions (see "Bulk Delete and Update" in the
search-and-indexing concepts guide).
Args:
filter_expression (Union[str, FilterExpression]): Selects the
documents to update. Prefer the escaping builders (``Tag``,
``Num``...) over raw filter strings; **never string-concatenate
untrusted input into a filter** — an injected predicate here
mutates data.
values (Dict[str, Any]): Field/value pairs to set on each match.
Values are written **as-is with no schema validation** (unlike
:meth:`load`): callers must pre-encode vectors/bytes and format
numerics as the schema expects. For JSON indexes, keys must
match the document's JSON **layout**, not the schema field name
— a field indexed at a nested path (e.g. ``$.metadata.status``)
must be passed nested (``{"metadata": {"status": ...}}``);
passing the flat field name writes the wrong path and leaves the
indexed field unchanged. Only static values are supported (no
callable/expression transforms).
batch_size (int): Number of documents to update per round-trip.
dry_run (bool): If True, report how many documents *would* be
updated without writing anything.
allow_all (bool): Must be True to run against a match-all filter.
on_progress (Optional[Callable[[int, int], None]]): Called after each
write batch with ``(processed, matched)``. Invoked synchronously
(do not pass a coroutine); raising from it aborts the run.
Returns:
BulkResult: ``matched``/``processed`` counts, plus ``completed``
(always True for update — it runs to completion or raises) and
``dry_run``.
See Also:
:meth:`drop_by_filter`, :meth:`load` (validated whole-document
upsert by key).
Note:
This operation is **not atomic** across the match set. Each
document is updated atomically (one ``HSET``/``JSON.MERGE``), but
batches use a non-transactional pipeline and are applied
incrementally with no rollback, so a crash or connection error
mid-run can leave some documents updated and others not. Because
the update is a fixed field set, it is idempotent: re-running the
same call after a failure converges on the intended state.
Keys are resolved before writing, so a document may be deleted by
another client in between. Each write is conditional on the key
still existing (applied atomically), so such a document is
**skipped rather than recreated** as a partial document; it simply
isn't counted in ``processed``. ``processed`` therefore reflects the
documents actually written, which can be less than ``matched`` under
concurrent deletion.
``processed`` counts **distinct** documents — a document is written
*at most* once, even though the resolving cursor can emit it on more
than one page — but there is no *at least* once guarantee. It can
exceed ``matched`` if the match set grew after the initial count, and
fall short of it either from a concurrent deletion or because the
cursor itself returned fewer rows than matched (which Redis
documents as possible). A short ``processed`` is therefore not
evidence of any particular cause.
"""
_require_specific_filter(filter_expression, allow_all)
if not values:
raise ValueError("values must be a non-empty mapping of field to value.")
matched = cast(int, self.query(CountQuery(filter_expression)))
if dry_run:
return BulkResult(
matched=matched, processed=matched, completed=True, dry_run=True
)
# Phase 1 (read): fully resolve matching keys before mutating. The
# aggregation cursor cannot be read while the index is written, so the
# read phase must finish first. on_progress fires only in phase 2.
keys = [
key
for batch in self._iter_keys_by_filter(filter_expression, batch_size)
for key in batch
]
# Phase 2 (write): cursor closed, safe to mutate the index.
total_updated = 0
for i in range(0, len(keys), batch_size):
total_updated += self._apply_update_batch(keys[i : i + batch_size], values)
if on_progress is not None:
on_progress(total_updated, matched)
return BulkResult(matched=matched, processed=total_updated, completed=True)
def _iter_keys_by_filter(
self, filter_expression: str | FilterExpression, batch_size: int
) -> Iterator[list[str]]:
"""Yield batches of distinct document keys matching a filter.
Uses ``FT.AGGREGATE ... LOAD 1 @__key WITHCURSOR`` to page through a
match set of any size (unlike ``FT.SEARCH`` + ``LIMIT``, which is capped
by ``MAXSEARCHRESULTS`` and non-deterministic without a unique sort).
The server-side cursor is always released on exit, even if iteration is
abandoned early or errors.
Keys are de-duplicated because a cursor is a position in an ascending
walk of internal document ids, not a snapshot, so it can return the same
key on two pages — RediSearch has no in-place update, so any write
appends a new, higher id ahead of the cursor. Dropping the ``seen`` set
would make callers write and count a document twice, which churns the
index further and provokes more repeats. It does not help the opposite
direction: a cursor may also return *fewer* rows than matched, so a full
drain is no proof every match was seen.
Two consequences: a batch can be smaller than ``batch_size`` (an
all-repeat page yields nothing), and memory is ``O(match count)`` — this
is not a streaming iterator, so callers with very large match sets should
partition the filter.
"""
request = (
AggregateRequest(str(filter_expression))
.load("__key")
.cursor(count=batch_size, max_idle=BULK_CURSOR_MAX_IDLE_SECONDS)
.dialect(2)
)
name = self.schema.index.name
ft = self._redis_client.ft(name) # type: ignore[union-attr]
cid = 0
seen: set[str] = set()
try:
result = ft.aggregate(request)
while True:
# Capture the cursor id BEFORE parsing rows, so the finally
# block can release the cursor even if row parsing raises.
# A cursor id of 0 signals the server has released the cursor.
cid = result.cursor.cid if result.cursor else 0
keys = _dedupe_keys((_agg_row_to_key(row) for row in result.rows), seen)
if keys:
yield keys
if not cid:
break
result = ft.aggregate(Cursor(cid))
except redis.exceptions.RedisError as e:
raise RedisSearchError(
f"Error while iterating documents by filter: {str(e)}"
) from e
finally:
# Release the cursor if it is still open (early break/error/abandon).
if cid:
try:
self._redis_client.execute_command( # type: ignore[union-attr]
"FT.CURSOR", "DEL", name, cid
)
except redis.exceptions.RedisError:
pass
def _apply_update_batch(self, batch_keys: list[str], values: dict[str, Any]) -> int:
"""Apply a partial update to a batch of keys, skipping any that no longer
exist (guards against recreating a concurrently-deleted document).
Returns the number of documents actually written.
"""
lua, args = _update_script_and_args(
self.storage_type == StorageType.JSON, values
)
client = self._redis_client
if isinstance(client, RedisCluster):
# Per-key to stay within a single slot per command.
written = 0
for key in batch_keys:
written += int(cast(int, client.eval(lua, 1, key, *args)))
return written
pipe = client.pipeline(transaction=False) # type: ignore[union-attr]
for key in batch_keys:
pipe.eval(lua, 1, key, *args)
return sum(int(r) for r in pipe.execute())
[docs]
def load(
self,
data: Iterable[Any],
id_field: str | None = None,
keys: Iterable[str] | None = None,
ttl: int | None = None,
preprocess: Callable | None = None,
batch_size: int | None = None,
) -> list[str]:
"""Load objects to the Redis database. Returns the list of keys loaded
to Redis.
RedisVL automatically handles constructing the object keys, batching,
optional preprocessing steps, and setting optional expiration
(TTL policies) on keys.
Args:
data (Iterable[Any]): An iterable of objects to store.
id_field (Optional[str], optional): Specified field used as the id
portion of the redis key (after the prefix) for each
object. Defaults to None.
keys (Optional[Iterable[str]], optional): Optional iterable of keys.
Must match the length of objects if provided. Defaults to None.
ttl (Optional[int], optional): Time-to-live in seconds for each key.
Defaults to None.
preprocess (Optional[Callable], optional): A function to preprocess
objects before storage. Defaults to None.
batch_size (Optional[int], optional): Number of objects to write in
a single Redis pipeline execution. Defaults to class's
default batch size.
Returns:
List[str]: List of keys loaded to Redis.
Raises:
SchemaValidationError: If validation fails when validate_on_load is enabled.
RedisVLError: If there's an error loading data to Redis.
"""
try:
return self._storage.write(
self._redis_client,
objects=data,
id_field=id_field,
keys=keys,
ttl=ttl,
preprocess=preprocess,
batch_size=batch_size,
validate=self._validate_on_load,
)
except SchemaValidationError:
# Log the detailed validation error with actionable information
logger.error("Data validation failed during load operation")
raise
except Exception as exc:
# Wrap other errors as general RedisVL errors
logger.exception("Error while loading data to Redis")
raise RedisVLError(f"Failed to load data: {str(exc)}") from exc
[docs]
def fetch(self, id: str) -> dict[str, Any] | None:
"""Fetch an object from Redis by id.
The id is typically either a unique identifier,
or derived from some domain-specific metadata combination
(like a document id or chunk id).
Args:
id (str): The specified unique identifier for a particular
document indexed in Redis.
Returns:
Dict[str, Any]: The fetched object.
"""
obj = self._storage.get(self._redis_client, [self.key(id)])
if obj:
return convert_bytes(obj[0])
return None
def _aggregate(self, aggregation_query: AggregationQuery) -> list[dict[str, Any]]:
"""Execute an aggregation query and processes the results."""
results = self.aggregate(
aggregation_query,
query_params=aggregation_query.params, # type: ignore[attr-defined]
)
return process_aggregate_results(
results,
query=aggregation_query,
storage_type=self.schema.index.storage_type,
)
def _sql_query(self, sql_query: SQLQuery) -> list[dict[str, Any]]:
"""Execute a SQL query and return results.
Args:
sql_query: The SQLQuery object containing the SQL statement.
Returns:
List of dictionaries containing the query results.
Raises:
ImportError: If sql-redis package is not installed.
"""
try:
from sql_redis import create_executor
except ImportError:
raise ImportError(
"sql-redis is required for SQL query support. "
"Install it with: pip install redisvl[sql-redis]"
)
sql_redis_options = _get_sql_redis_options(sql_query)
cache_key = _sql_executor_cache_key(sql_redis_options)
with self._lock:
executor = self._sql_executors.get(cache_key)
if executor is None:
executor = create_executor(self._redis_client, **sql_redis_options)
self._sql_executors[cache_key] = executor
# Execute the query with any params
result = executor.execute(sql_query.sql, params=sql_query.params)
return _convert_and_drop_empty_rows(result.rows, "SQL")
[docs]
def aggregate(self, *args, **kwargs) -> "AggregateResult":
"""Perform an aggregation operation against the index.
Wrapper around the aggregation API that adds the index name
to the query and passes along the rest of the arguments
to the redis-py ft().aggregate() method.
Returns:
Result: Raw Redis aggregation results.
"""
try:
return self._redis_client.ft(self.schema.index.name).aggregate(
*args, **kwargs
)
except redis.exceptions.RedisError as e:
if "CROSSSLOT" in str(e):
raise RedisSearchError(
"Cross-slot error during aggregation. Ensure consistent hash tags in your keys."
)
raise RedisSearchError(f"Error while aggregating: {str(e)}") from e
except Exception as e:
raise RedisSearchError(
f"Unexpected error while aggregating: {str(e)}"
) from e
[docs]
def batch_search(
self, queries: list[SearchParams], batch_size: int = 10
) -> list["Result"]:
"""Perform a search against the index for multiple queries.
This method takes a list of queries and optionally query params and
returns a list of Result objects for each query. Results are
returned in the same order as the queries.
NOTE: Cluster users may need to incorporate hash tags into their query
to avoid cross-slot operations.
Args:
queries (List[SearchParams]): The queries to search for.
batch_size (int, optional): The number of queries to search for at a time.
Defaults to 10.
Returns:
List[Result]: The search results for each query.
"""
all_results = []
search = self._redis_client.ft(self.schema.index.name)
options = {}
if get_protocol_version(self._redis_client) not in ["3", 3]:
options[NEVER_DECODE] = True
for i in range(0, len(queries), batch_size):
batch_queries = queries[i : i + batch_size]
# redis-py doesn't support calling `search` in a pipeline,
# so we need to manually execute each command in a pipeline
# and parse the results
with self._redis_client.pipeline(transaction=False) as pipe:
batch_built_queries = []
for query in batch_queries:
if isinstance(query, tuple):
query_args, q = search._mk_query_args( # type: ignore
query[0], query_params=query[1]
)
else:
query_args, q = search._mk_query_args( # type: ignore
query, query_params=None
)
batch_built_queries.append(q)
pipe.execute_command(
"FT.SEARCH",
*query_args,
**options,
)
st = time.time()
results = pipe.execute()
# We don't know how long each query took, so we'll use the total time
# for all queries in the batch as the duration for each query
duration = (time.time() - st) * 1000.0
for j, query_results in enumerate(results):
_built_query = batch_built_queries[j]
parsed_result = _parse_batch_search_result(
search, query_results, _built_query, duration
)
# Return a parsed Result object for each query
all_results.append(parsed_result)
return all_results
[docs]
def search(self, *args, **kwargs) -> "Result":
"""Perform a search against the index.
Wrapper around the search API that adds the index name
to the query and passes along the rest of the arguments
to the redis-py ft().search() method.
Returns:
Result: Raw Redis search results.
"""
try:
if isinstance(self._redis_client, RedisCluster):
# Use special cluster search for RedisCluster
return cluster_search(
self._redis_client.ft(self.schema.index.name),
*args,
**kwargs, # type: ignore
)
else:
return self._redis_client.ft(self.schema.index.name).search(
*args, **kwargs
) # type: ignore
except redis.exceptions.RedisError as e:
if "CROSSSLOT" in str(e):
raise RedisSearchError(
"Cross-slot error during search. Ensure consistent hash tags in your keys."
)
raise RedisSearchError(f"Error while searching: {str(e)}") from e
except Exception as e:
raise RedisSearchError(f"Unexpected error while searching: {str(e)}") from e
def _hybrid_search(self, query: HybridQuery, **kwargs) -> list[dict[str, Any]]:
"""Perform a hybrid search against the index, combining text and vector search.
Args:
query (HybridQuery): The text+vector search query to be performed, with configurable fusion methods and
post-processing.
kwargs: Additional arguments to pass to the redis-py hybrid_search method (e.g. timeout).
Returns:
List[Dict[str, Any]]: The search results ordered by combined score unless otherwise specified.
Notes:
Hybrid search is only available in Redis 8.4.0+, and requires redis-py >= 7.1.0.
See Also:
- `FT.HYBRID command documentation <https://redis.io/docs/latest/commands/ft.hybrid>`_
- `redis-py hybrid_search documentation <https://redis.readthedocs.io/en/stable/redismodules.html#redis.commands.search.commands.SearchCommands.hybrid_search>`_
.. code-block:: python
from redisvl.query import HybridQuery
hybrid_query = HybridQuery(
text="lorem ipsum dolor sit amet",
text_field_name="description",
vector=[0.1, 0.2, 0.3],
vector_field_name="embedding"
)
results = index.query(hybrid_query)
"""
index = self._redis_client.ft(self.schema.index.name)
self._validate_hybrid_query(query)
if not hasattr(index, "hybrid_search"):
raise ImportError(_HYBRID_SEARCH_ERROR_MESSAGE)
results = index.hybrid_search(
query=query.query,
combine_method=query.combination_method,
post_processing=(
query.postprocessing_config
if query.postprocessing_config.build_args()
else None
),
params_substitution=query.params, # type: ignore[arg-type]
**kwargs,
) # type: ignore
return _convert_and_drop_empty_rows(results.results, "hybrid") # type: ignore[union-attr]
[docs]
def batch_query(
self, queries: Sequence[BaseQuery], batch_size: int = 10
) -> list[list[dict[str, Any]]]:
"""Execute a batch of queries and process results."""
results = self.batch_search(
[(query.query, query.params) for query in queries], batch_size=batch_size
)
all_parsed = []
for query, batch_results in zip(queries, results):
parsed = process_results(batch_results, query=query, schema=self.schema)
# Create separate lists of parsed results for each query
# passed in to the batch_search method, so that callers can
# access the results for each query individually
all_parsed.append(parsed)
return all_parsed
def _query(self, query: BaseQuery) -> list[dict[str, Any]]:
"""Execute a query and process results."""
try:
self._validate_query(query)
except QueryValidationError as e:
raise QueryValidationError(f"Invalid query: {str(e)}") from e
results = self.search(query.query, query_params=query.params)
return process_results(results, query=query, schema=self.schema)
[docs]
def query(
self, query: BaseQuery | AggregationQuery | HybridQuery | SQLQuery
) -> list[dict[str, Any]]:
"""Execute a query on the index.
This method takes a BaseQuery, AggregationQuery, or HybridQuery object directly, and
handles post-processing of the search.
Args:
query (Union[BaseQuery, AggregationQuery, HybridQuery]): The query to run.
Returns:
List[Result]: A list of search results.
.. code-block:: python
from redisvl.query import VectorQuery
query = VectorQuery(
vector=[0.16, -0.34, 0.98, 0.23],
vector_field_name="embedding",
num_results=3
)
results = index.query(query)
"""
if isinstance(query, AggregationQuery):
return self._aggregate(query)
elif isinstance(query, SQLQuery):
return self._sql_query(query)
elif isinstance(query, HybridQuery):
return self._hybrid_search(query)
else:
return self._query(query)
[docs]
def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator:
"""Execute a given query against the index and return results in
paginated batches.
This method accepts a RedisVL query instance, enabling pagination of
results which allows for subsequent processing over each batch with a
generator.
Args:
query (BaseQuery): The search query to be executed.
page_size (int, optional): The number of results to return in each
batch. Defaults to 30.
Yields:
A generator yielding non-empty ``SearchResults`` batches.
Raises:
TypeError: If the page_size argument is not of type int.
ValueError: If the page_size argument is less than or equal to zero.
.. code-block:: python
# Iterate over paginated search results in batches of 10
for result_batch in index.paginate(query, page_size=10):
# Process each batch of results
pass
Note:
The page_size parameter controls the number of items each result
batch contains. Adjust this value based on performance
considerations and the expected volume of search results.
Note:
For stable pagination, the query must have a `sort_by` clause on a
**unique** field. Redis documents that ``LIMIT`` without sorting is
non-deterministic, so pages may otherwise repeat or miss documents.
Very deep pagination is also bounded server-side by
``search-max-search-results`` (1,000,000 by default, but 10,000 on
some managed tiers), past which the search errors rather than ending.
Note:
A yielded batch may contain fewer than ``page_size`` documents, and an
empty batch is never yielded. On Redis 8+ a matched document whose
data expires while the search is running is skipped; each batch
reports how many were skipped via ``dropped_count`` (and
``complete``), including skips inherited from a page that was dropped
in its entirety. Pagination still runs to the end of the result set in
that case.
"""
if not isinstance(page_size, int):
raise TypeError("page_size must be an integer")
if page_size <= 0:
raise ValueError("page_size must be greater than 0")
if isinstance(query, CountQuery):
raise TypeError(
"CountQuery cannot be paginated: it returns a match count rather "
"than documents. Use index.query(query) instead."
)
offset = 0
carried_drops = 0
while True:
query.paging(offset, page_size)
results = self._query(query)
if not _page_had_matches(results):
break
if results:
_fold_carried_drops(results, carried_drops)
carried_drops = 0
yield results
else:
# Whole page dropped: keep its count so it reaches the caller on
# the next yielded batch instead of vanishing with the page.
carried_drops += getattr(results, "dropped_count", 0)
# Advance unconditionally, so a page we cannot materialize can never
# wedge the loop.
offset += page_size
[docs]
def listall(self) -> list[str]:
"""List all search indices in Redis database.
Returns:
List[str]: The list of indices in the database.
"""
return convert_bytes(self._redis_client.execute_command("FT._LIST"))
[docs]
def exists(self) -> bool:
"""Check if the index exists in Redis.
Returns:
bool: True if the index exists, False otherwise.
Raises:
RedisSearchError: If the check fails for any reason other than the
index not existing, such as a connection failure or insufficient
permissions. The original ``redis-py`` exception is available as
``__cause__``.
"""
try:
info = self._info(self.schema.index.name, self._redis_client)
except RedisSearchError as e:
if _is_missing_index_error(e):
return False
raise
# FT.INFO answers for an alias's target, so compare the resolved name:
# reporting an alias as existing would let create(overwrite=True) drop
# the index it points at.
return info.get("index_name") == self.schema.index.name
@staticmethod
def _info(name: str, redis_client: SyncRedisClient) -> dict[str, Any]:
"""Run FT.INFO to fetch information about the index."""
try:
if isinstance(redis_client, SyncRedisCluster):
node = redis_client.get_random_node()
values = redis_client.execute_command(
"FT.INFO", name, target_nodes=node
)
info = make_dict(values)
else:
info = redis_client.ft(name).info()
return convert_bytes(info)
except Exception as e:
raise RedisSearchError(
f"Error while fetching {name} index info: {str(e)}"
) from e
[docs]
def info(self, name: str | None = None) -> dict[str, Any]:
"""Get information about the index.
Args:
name (str, optional): Index name to fetch info about.
Defaults to None.
Returns:
dict: A dictionary containing the information about the index.
"""
index_name = name or self.schema.index.name
return self._info(index_name, self._redis_client)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.disconnect()
[docs]
class AsyncSearchIndex(BaseSearchIndex):
"""A search index class for interacting with Redis as a vector database in
async-mode.
The AsyncSearchIndex is instantiated with a reference to a Redis database
and an IndexSchema (YAML path or dictionary object) that describes the
various settings and field configurations.
.. code-block:: python
from redisvl.index import AsyncSearchIndex
# initialize the index object with schema from file
index = AsyncSearchIndex.from_yaml(
"schemas/schema.yaml",
redis_url="redis://localhost:6379",
validate_on_load=True
)
# create the index
await index.create(overwrite=True, drop=False)
# data is an iterable of dictionaries
await index.load(data)
# delete index and data
await index.delete(drop=True)
"""
@deprecated_argument("redis_kwargs", "Use connection_kwargs instead.")
def __init__(
self,
schema: IndexSchema,
*,
redis_url: str | None = None,
redis_client: AsyncRedisClient | None = None,
connection_kwargs: dict[str, Any] | None = None,
validate_on_load: bool = False,
**kwargs,
):
"""Initialize the RedisVL async search index with a schema.
Args:
schema (IndexSchema): Index schema object.
redis_url (Optional[str], optional): The URL of the Redis server to
connect to.
redis_client (Optional[AsyncRedis]): An
instantiated redis client.
connection_kwargs (Optional[Dict[str, Any]]): Redis client connection
args.
validate_on_load (bool, optional): Whether to validate data against schema
when loading. Defaults to False.
"""
if "redis_kwargs" in kwargs:
connection_kwargs = kwargs.pop("redis_kwargs")
# final validation on schema object
if not isinstance(schema, IndexSchema):
raise ValueError("Must provide a valid IndexSchema object")
self.schema = schema
self._validate_on_load = validate_on_load
self._lib_name: str | None = kwargs.pop("lib_name", None)
# Store connection parameters
self._redis_client = redis_client
self._redis_url = redis_url
self._connection_kwargs = connection_kwargs or {}
self._lock = asyncio.Lock()
self._sql_executors: dict[str, Any] = {}
self._validated_client = kwargs.pop("_client_validated", False)
self._owns_redis_client = kwargs.pop("_owns_redis_client", redis_client is None)
self._client_finalizer = None
# Close the owned client when this index is garbage collected. When
# the client is created lazily, registration happens at creation time
# instead (see _get_client).
self._register_client_finalizer(redis_client)
_finalizer_close_client = staticmethod(_close_owned_async_client)
[docs]
@classmethod
async def from_existing(
cls,
name: str,
redis_client: AsyncRedisClient | None = None,
redis_url: str | None = None,
**kwargs,
):
"""
Initialize from an existing search index in Redis by index name.
Args:
name (str): Name of the search index in Redis.
redis_client(Optional[Redis]): An
instantiated redis client.
redis_url (Optional[str]): The URL of the Redis server to
connect to.
"""
if not redis_url and not redis_client:
raise ValueError(
"Must provide either a redis_url or redis_client to fetch Redis index info."
)
init_kwargs, connection_kwargs = _split_from_existing_kwargs(
dict(kwargs),
nested_connection_keys=("connection_kwargs", "redis_kwargs"),
)
lib_name = cast(str | None, init_kwargs.get("lib_name"))
created_redis_client = False
if redis_client:
# Validate client type and set lib name
await RedisConnectionFactory.validate_async_redis(redis_client, lib_name)
# Mark that client was already validated to avoid duplicate calls
init_kwargs["_client_validated"] = True
elif redis_url:
factory_kwargs = {**connection_kwargs}
if lib_name is not None:
factory_kwargs["lib_name"] = lib_name
redis_client = await RedisConnectionFactory._get_aredis_connection(
redis_url=redis_url,
**factory_kwargs,
)
init_kwargs["_client_validated"] = True
created_redis_client = True
if redis_client is None:
raise ValueError(
"Failed to obtain a valid Redis client. "
"Please provide a valid redis_client or redis_url."
)
# Fetch index info and convert to schema
index_info = await cls._info(name, redis_client)
schema_dict = convert_index_info_to_schema(index_info)
schema = IndexSchema.from_dict(schema_dict)
if created_redis_client:
init_kwargs["_owns_redis_client"] = True
return cls(
schema,
redis_client=redis_client,
redis_url=redis_url,
connection_kwargs=connection_kwargs or None,
**init_kwargs,
)
return cls(schema, redis_client=redis_client, **init_kwargs)
@property
def client(self) -> AsyncRedisClient | None:
"""The underlying redis-py client object."""
return self._redis_client
[docs]
@deprecated_function("connect", "Pass connection parameters in __init__.")
async def connect(self, redis_url: str | None = None, **kwargs):
"""[DEPRECATED] Connect to a Redis instance. Use connection parameters in __init__."""
warnings.warn(
"connect() is deprecated; pass connection parameters in __init__",
DeprecationWarning,
)
self.invalidate_sql_schema_cache()
client = await RedisConnectionFactory._get_aredis_connection(
redis_url=redis_url, **kwargs
)
await self.set_client(client)
[docs]
@deprecated_function("set_client", "Pass connection parameters in __init__.")
async def set_client(self, redis_client: AsyncRedisClient | SyncRedisClient):
"""
[DEPRECATED] Manually set the Redis client to use with the search index.
This method is deprecated; please provide connection parameters in __init__.
"""
redis_client = await self._validate_client(redis_client)
self.invalidate_sql_schema_cache()
await self.disconnect()
async with self._lock:
self._redis_client = redis_client
self._register_client_finalizer(redis_client)
return self
async def _get_client(self) -> AsyncRedisClient:
"""Lazily instantiate and return the async Redis client."""
if self._redis_client is None:
async with self._lock:
# Double-check to protect against concurrent access
if self._redis_client is None:
# Pass lib_name to connection factory
kwargs = {**self._connection_kwargs}
if self._redis_url:
kwargs["redis_url"] = self._redis_url
if self._lib_name:
kwargs["lib_name"] = self._lib_name
self._redis_client = (
await RedisConnectionFactory._get_aredis_connection(**kwargs)
)
self._register_client_finalizer(self._redis_client)
if not self._validated_client and self._lib_name:
# Set lib name for user-provided clients
await RedisConnectionFactory.validate_async_redis(
self._redis_client,
self._lib_name,
)
self._validated_client = True
return self._redis_client
async def _validate_client(
self, redis_client: AsyncRedisClient | SyncRedisClient
) -> AsyncRedisClient:
# Handle deprecated sync client conversion
if isinstance(redis_client, (Redis, RedisCluster)):
warnings.warn(
"Passing a sync Redis client to AsyncSearchIndex is deprecated "
"and will be removed in the next major version. Please use an "
"async Redis client instead.",
DeprecationWarning,
)
# Use a new variable name
async_redis_client: AsyncRedisClient = (
RedisConnectionFactory.sync_to_async_redis(redis_client)
)
return async_redis_client # Return the converted client
# Check if it's a valid async client (standard or cluster)
elif not isinstance(redis_client, (AsyncRedis, AsyncRedisCluster)):
raise ValueError(
"Invalid async client type: must be AsyncRedis or AsyncRedisCluster"
)
# If it passed the elif, it's already an AsyncRedisClient
return redis_client
@staticmethod
async def _info(name: str, redis_client: AsyncRedisClient) -> dict[str, Any]:
try:
if isinstance(redis_client, AsyncRedisCluster):
node = redis_client.get_random_node()
values = await redis_client.execute_command(
"FT.INFO", name, target_nodes=node
)
info = make_dict(values)
else:
info = await redis_client.ft(name).info()
return convert_bytes(info)
except Exception as e:
raise RedisSearchError(
f"Error while fetching {name} index info: {str(e)}"
) from e
async def _check_svs_support_async(self) -> None:
"""Validate SVS-VAMANA support.
Raises:
RedisModuleVersionError: If SVS-VAMANA requirements are not met.
"""
client = await self._get_client()
if not await supports_svs_async(client):
raise RedisModuleVersionError.for_svs_vamana(SVS_MIN_REDIS_VERSION)
[docs]
async def create(self, overwrite: bool = False, drop: bool = False) -> None:
"""Asynchronously create an index in Redis with the current schema
and properties.
Args:
overwrite (bool, optional): Whether to overwrite the index if it
already exists. Defaults to False.
drop (bool, optional): Whether to drop all keys associated with the
index in the case of overwriting. Defaults to False.
Raises:
RuntimeError: If the index already exists and 'overwrite' is False.
ValueError: If no fields are defined for the index.
.. code-block:: python
# create an index in Redis; only if one does not exist with given name
await index.create()
# overwrite an index in Redis without dropping associated data
await index.create(overwrite=True)
# overwrite an index in Redis; drop associated data (clean slate)
await index.create(overwrite=True, drop=True)
"""
client = await self._get_client()
redis_fields = self.schema.redis_fields
if not redis_fields:
raise ValueError("No fields defined for index")
if not isinstance(overwrite, bool):
raise TypeError("overwrite must be of type bool")
# Check if schema uses SVS-VAMANA and validate Redis capabilities
if self._uses_svs_vamana():
await self._check_svs_support_async()
if await self.exists():
if not overwrite:
logger.info("Index already exists, not overwriting.")
return None
logger.info("Index already exists, overwriting.")
await self.delete(drop)
try:
# Prefix can be a string or list of strings
prefix = self.schema.index.prefix
prefix_list = prefix if isinstance(prefix, list) else [prefix]
definition = IndexDefinition(
prefix=prefix_list, index_type=self._storage.type
)
# Extract stopwords from schema
stopwords = self.schema.index.stopwords
if isinstance(client, AsyncRedisCluster):
await async_cluster_create_index(
index_name=self.schema.index.name,
client=client,
fields=redis_fields,
definition=definition,
stopwords=stopwords,
)
else:
await client.ft(self.schema.index.name).create_index(
fields=redis_fields,
definition=definition,
stopwords=stopwords,
)
self.invalidate_sql_schema_cache()
except redis.exceptions.RedisError as e:
raise RedisSearchError(
f"Failed to create index '{self.name}' on Redis: {str(e)}"
) from e
except Exception as e:
logger.exception("Error while trying to create the index")
raise RedisSearchError(
f"Unexpected error creating index '{self.name}': {str(e)}"
) from e
[docs]
async def delete(self, drop: bool = True):
"""Delete the search index.
``FT.DROPINDEX`` resolves index aliases, so if this schema's name happens
to be an alias, the index the alias points at is dropped instead -- along
with its documents when ``drop`` is True. :meth:`exists` reports an alias
as absent to keep :meth:`create` away from this path.
Args:
drop (bool, optional): Delete the documents in the index.
Defaults to True.
Raises:
RedisSearchError: If the index cannot be deleted, for example when it
does not exist. The original ``redis-py`` exception is available
as ``__cause__``.
"""
client = await self._get_client()
try:
# For Redis Cluster with drop=True, we need to handle key deletion manually
# to avoid cross-slot errors since we control the keys opaquely
if drop and isinstance(client, AsyncRedisCluster):
# First clear all keys manually (handles cluster compatibility)
await self.clear()
# Then drop the index without the DD flag
cmd_args = ["FT.DROPINDEX", self.schema.index.name]
else:
# Standard approach for non-cluster or when not dropping keys
cmd_args = ["FT.DROPINDEX", self.schema.index.name]
if drop:
cmd_args.append("DD")
if isinstance(client, AsyncRedisCluster):
target_nodes = [client.get_default_node()]
await client.execute_command(*cmd_args, target_nodes=target_nodes)
else:
await client.execute_command(*cmd_args)
self.invalidate_sql_schema_cache()
except Exception as e:
raise RedisSearchError(f"Error while deleting index: {str(e)}") from e
async def _delete_batch(self, batch_keys: list[str]) -> int:
"""Delete a batch of keys from Redis.
For Redis Cluster, keys are deleted individually due to potential
cross-slot limitations. For standalone Redis, keys are deleted in
a single operation for better performance.
Args:
batch_keys (List[str]): List of Redis keys to delete.
Returns:
int: Count of records deleted from Redis.
"""
client = await self._get_client()
is_cluster = isinstance(client, AsyncRedisCluster)
if is_cluster:
records_deleted_in_batch = 0
for key_to_delete in batch_keys:
try:
records_deleted_in_batch += cast(
int, await client.delete(key_to_delete)
)
except redis.exceptions.RedisError as e:
logger.warning(f"Failed to delete key {key_to_delete}: {e}")
else:
records_deleted_in_batch = await client.delete(*batch_keys)
return records_deleted_in_batch
[docs]
async def clear(self) -> int:
"""Clear all keys in Redis associated with the index, leaving the index
available and in-place for future insertions or updates.
NOTE: This method requires custom behavior for Redis Cluster because here,
we can't easily give control of the keys we're clearing to the user so they
can separate them based on hash tag.
See :meth:`SearchIndex.clear` for the sweep's caveats, which apply
identically here.
Returns:
int: Count of records deleted from Redis.
"""
batch_size = 500
matched = cast(int, await self.query(CountQuery(FilterExpression("*"))))
# Runaway backstop sized to the matched count plus slack for concurrent
# inserts -- the same shape as drop_by_filter. CountQuery is FT.SEARCH,
# which the query loop below already needs and which `+@read +@write`
# grants; the FT.INFO this once read for the same purpose is `@search`
# only, so a single call for a loop bound denied the whole method.
max_records = ceil(matched * 1.5) + batch_size
total_records_deleted: int = 0
offset = 0
query = FilterQuery(FilterExpression("*"), return_fields=["id"])
while True:
if total_records_deleted > max_records:
logger.warning(
"clear() of index %s hit its runaway backstop (%d) with "
"documents possibly still indexed; %d records were deleted. "
"Re-run to continue.",
self.schema.index.name,
max_records,
total_records_deleted,
)
break
query.paging(offset, batch_size)
batch = await self._query(query)
if not batch:
break
batch_keys = [record["id"] for record in batch]
records_deleted = await self._delete_batch(batch_keys)
total_records_deleted += records_deleted
if records_deleted:
# Deleted documents leave the index, so the next page of
# survivors is at offset 0 again.
offset = 0
else:
# Nothing in this page could be deleted, most plausibly a
# permission denial swallowed by _delete_batch's cluster branch.
# Page past it: the documents behind may still be deletable.
offset += batch_size
if offset > max_records:
logger.warning(
"clear() of index %s paged past its runaway backstop "
"(%d) without being able to delete; %d records were "
"deleted. Documents remain.",
self.schema.index.name,
max_records,
total_records_deleted,
)
break
self.invalidate_sql_schema_cache()
return total_records_deleted
[docs]
def invalidate_sql_schema_cache(self) -> None:
"""Clear cached sql-redis executors and schema state for this index."""
self._sql_executors.clear()
async def _unlink_batch(self, batch_keys: list[str]) -> int:
"""Unlink a batch of keys from Redis (async).
Mirrors ``_delete_batch`` but uses non-blocking ``UNLINK``. For Redis
Cluster, keys are unlinked individually to avoid cross-slot errors.
"""
client = await self._get_client()
if isinstance(client, AsyncRedisCluster):
unlinked = 0
for key_to_unlink in batch_keys:
try:
unlinked += cast(int, await client.unlink(key_to_unlink))
except redis.exceptions.RedisError as e:
logger.warning(f"Failed to unlink key {key_to_unlink}: {e}")
return unlinked
return cast(int, await client.unlink(*batch_keys))
[docs]
async def drop_keys(
self, keys: str | list[str], batch_size: int = DEFAULT_BULK_BATCH_SIZE
) -> int:
"""Remove a specific entry or entries from the index by it's key ID.
Uses ``UNLINK`` rather than ``DEL`` so memory reclamation runs on a
background thread. This avoids blocking the main thread when a large
number of keys are dropped at once (for example, scope-targeted
``SemanticCache`` invalidation). The returned count is unchanged.
Large key lists are unlinked in chunks of ``batch_size``. On Redis
Cluster, keys are unlinked individually so a chunk that spans hash
slots does not raise ``CROSSSLOT``.
Args:
keys (Union[str, List[str]]): The document ID or IDs to remove from the index.
batch_size (int): Number of keys to unlink per round-trip.
Returns:
int: Count of records deleted from Redis.
"""
client = await self._get_client()
if not isinstance(keys, list):
return await client.unlink(keys)
if not keys:
return 0
total = 0
for i in range(0, len(keys), batch_size):
total += await self._unlink_batch(keys[i : i + batch_size])
return total
[docs]
async def drop_documents(
self, ids: str | list[str], batch_size: int = DEFAULT_BULK_BATCH_SIZE
) -> int:
"""Remove documents from the index by their document IDs.
This method converts document IDs to Redis keys automatically by applying
the index's key prefix and separator configuration.
NOTE: Cluster users will need to incorporate hash tags into their
document IDs and only call this method with documents from a single hash
tag at a time.
Args:
ids (Union[str, List[str]]): The document ID or IDs to remove from the index.
batch_size (int): Number of documents to delete per round-trip
(standalone Redis only; cluster deletes in a single call after
the shared-hash-tag check).
Returns:
int: Count of documents deleted from Redis.
Raises:
ValueError: On Redis Cluster, if the resolved keys do not all share
a hash tag (a cross-slot ``DELETE`` is not permitted).
"""
client = await self._get_client()
if not isinstance(ids, list):
key = self.key(ids)
return await client.delete(key)
if not ids:
return 0
keys = [self.key(id) for id in ids]
# On cluster, a multi-key DELETE must stay within one hash slot, so
# require the keys to share a hash tag and delete them in a single call.
if isinstance(client, AsyncRedisCluster):
if not _keys_share_hash_tag(keys):
raise ValueError(
"All keys must share a hash tag when using Redis Cluster."
)
return await client.delete(*keys)
# Standalone: delete in chunks to bound the size of any single command.
total = 0
for i in range(0, len(keys), batch_size):
batch = keys[i : i + batch_size]
total += cast(int, await client.delete(*batch))
return total
[docs]
async def expire_keys(self, keys: str | list[str], ttl: int) -> int | list[int]:
"""Set the expiration time for a specific entry or entries in Redis.
Args:
keys (Union[str, List[str]]): The entry ID or IDs to set the expiration for.
ttl (int): The time-to-live in seconds.
"""
client = await self._get_client()
if isinstance(keys, list):
pipe = client.pipeline()
for key in keys:
pipe.expire(key, ttl)
return await pipe.execute()
else:
return await client.expire(keys, ttl)
[docs]
async def drop_by_filter(
self,
filter_expression: str | FilterExpression,
*,
batch_size: int = DEFAULT_BULK_BATCH_SIZE,
dry_run: bool = False,
allow_all: bool = False,
on_progress: Callable[[int, int], None] | None = None,
) -> BulkResult:
"""Delete every document matching a filter expression (async).
See :meth:`SearchIndex.drop_by_filter` for full semantics.
"""
_require_specific_filter(filter_expression, allow_all)
matched = cast(int, await self.query(CountQuery(filter_expression)))
if dry_run:
return BulkResult(
matched=matched, processed=matched, completed=True, dry_run=True
)
max_records = ceil(matched * 1.5) + batch_size
total_deleted = 0
completed = True
query = FilterQuery(filter_expression, return_fields=["id"])
query.paging(0, batch_size)
while True:
if total_deleted > max_records:
logger.warning(
"drop_by_filter hit its runaway backstop (%d) with documents "
"possibly still matching; returning an incomplete result. "
"Re-run to continue.",
max_records,
)
completed = False
break
batch = await self._query(query)
if not batch:
break
batch_keys = [record["id"] for record in batch]
total_deleted += await self._unlink_batch(batch_keys)
if on_progress is not None:
on_progress(total_deleted, matched)
self.invalidate_sql_schema_cache()
return BulkResult(matched=matched, processed=total_deleted, completed=completed)
[docs]
async def update_by_filter(
self,
filter_expression: str | FilterExpression,
values: dict[str, Any],
*,
batch_size: int = DEFAULT_BULK_BATCH_SIZE,
dry_run: bool = False,
allow_all: bool = False,
on_progress: Callable[[int, int], None] | None = None,
) -> BulkResult:
"""Set ``values`` on every document matching a filter expression (async).
See :meth:`SearchIndex.update_by_filter` for full semantics.
"""
_require_specific_filter(filter_expression, allow_all)
if not values:
raise ValueError("values must be a non-empty mapping of field to value.")
matched = cast(int, await self.query(CountQuery(filter_expression)))
if dry_run:
return BulkResult(
matched=matched, processed=matched, completed=True, dry_run=True
)
# Phase 1 (read): fully resolve matching keys before mutating (the
# cursor cannot be read while the index is written). on_progress fires
# only in phase 2.
keys: list[str] = []
async for batch in self._iter_keys_by_filter(filter_expression, batch_size):
keys.extend(batch)
# Phase 2 (write): cursor closed, safe to mutate the index.
total_updated = 0
for i in range(0, len(keys), batch_size):
total_updated += await self._apply_update_batch(
keys[i : i + batch_size], values
)
if on_progress is not None:
on_progress(total_updated, matched)
return BulkResult(matched=matched, processed=total_updated, completed=True)
async def _iter_keys_by_filter(
self, filter_expression: str | FilterExpression, batch_size: int
) -> AsyncGenerator[list[str], None]:
"""Yield batches of distinct document keys matching a filter (async).
De-duplicates keys across cursor pages and always releases the
server-side cursor on exit. See the sync counterpart for why a cursor
repeats keys and why that makes memory ``O(match count)``.
"""
client = await self._get_client()
request = (
AggregateRequest(str(filter_expression))
.load("__key")
.cursor(count=batch_size, max_idle=BULK_CURSOR_MAX_IDLE_SECONDS)
.dialect(2)
)
name = self.schema.index.name
ft = client.ft(name)
cid = 0
seen: set[str] = set()
try:
result = await ft.aggregate(request) # type: ignore[arg-type]
while True:
# Capture the cursor id BEFORE parsing rows, so the finally
# block can release the cursor even if row parsing raises.
# A cursor id of 0 signals the server has released the cursor.
cid = result.cursor.cid if result.cursor else 0
keys = _dedupe_keys((_agg_row_to_key(row) for row in result.rows), seen)
if keys:
yield keys
if not cid:
break
result = await ft.aggregate(Cursor(cid))
except redis.exceptions.RedisError as e:
raise RedisSearchError(
f"Error while iterating documents by filter: {str(e)}"
) from e
finally:
if cid:
try:
await client.execute_command("FT.CURSOR", "DEL", name, cid)
except redis.exceptions.RedisError:
pass
async def _apply_update_batch(
self, batch_keys: list[str], values: dict[str, Any]
) -> int:
"""Apply a partial update to a batch of keys, skipping any that no longer
exist (async; see sync counterpart). Returns documents actually written.
"""
client = await self._get_client()
lua, args = _update_script_and_args(
self.storage_type == StorageType.JSON, values
)
if isinstance(client, AsyncRedisCluster):
written = 0
for key in batch_keys:
written += int(cast(int, await client.eval(lua, 1, key, *args))) # type: ignore[misc]
return written
pipe = client.pipeline(transaction=False)
for key in batch_keys:
pipe.eval(lua, 1, key, *args)
results = await pipe.execute()
return sum(int(r) for r in results)
[docs]
@deprecated_argument("concurrency", "Use batch_size instead.")
async def load(
self,
data: Iterable[Any],
id_field: str | None = None,
keys: Iterable[str] | None = None,
ttl: int | None = None,
preprocess: Callable | None = None,
concurrency: int | None = None,
batch_size: int | None = None,
) -> list[str]:
"""Asynchronously load objects to Redis. Returns the list of keys loaded
to Redis.
RedisVL automatically handles constructing the object keys, batching,
optional preprocessing steps, and setting optional expiration
(TTL policies) on keys.
Args:
data (Iterable[Any]): An iterable of objects to store.
id_field (Optional[str], optional): Specified field used as the id
portion of the redis key (after the prefix) for each
object. Defaults to None.
keys (Optional[Iterable[str]], optional): Optional iterable of keys.
Must match the length of objects if provided. Defaults to None.
ttl (Optional[int], optional): Time-to-live in seconds for each key.
Defaults to None.
preprocess (Optional[Callable], optional): A function to
preprocess objects before storage. Defaults to None.
batch_size (Optional[int], optional): Number of objects to write in
a single Redis pipeline execution. Defaults to class's
default batch size.
Returns:
List[str]: List of keys loaded to Redis.
Raises:
SchemaValidationError: If validation fails when validate_on_load is enabled.
RedisVLError: If there's an error loading data to Redis.
.. code-block:: python
data = [{"test": "foo"}, {"test": "bar"}]
# simple case
keys = await index.load(data)
# set 360 second ttl policy on data
keys = await index.load(data, ttl=360)
# load data with predefined keys
keys = await index.load(data, keys=["rvl:foo", "rvl:bar"])
# load data with preprocessing step
def add_field(d):
d["new_field"] = 123
return d
keys = await index.load(data, preprocess=add_field)
"""
client = await self._get_client()
try:
return await self._storage.awrite(
client,
objects=data,
id_field=id_field,
keys=keys,
ttl=ttl,
preprocess=preprocess,
batch_size=batch_size,
validate=self._validate_on_load,
)
except SchemaValidationError:
# Log the detailed validation error with actionable information
logger.error("Data validation failed during load operation")
raise
except Exception as exc:
# Wrap other errors as general RedisVL errors
logger.exception("Error while loading data to Redis")
raise RedisVLError(f"Failed to load data: {str(exc)}") from exc
[docs]
async def fetch(self, id: str) -> dict[str, Any] | None:
"""Asynchronously etch an object from Redis by id. The id is typically
either a unique identifier, or derived from some domain-specific
metadata combination (like a document id or chunk id).
Args:
id (str): The specified unique identifier for a particular
document indexed in Redis.
Returns:
Dict[str, Any]: The fetched object.
"""
client = await self._get_client()
obj = await self._storage.aget(client, [self.key(id)])
if obj:
return convert_bytes(obj[0])
return None
async def _aggregate(
self, aggregation_query: AggregationQuery
) -> list[dict[str, Any]]:
"""Execute an aggregation query and processes the results."""
results = await self.aggregate(
aggregation_query,
query_params=aggregation_query.params, # type: ignore[attr-defined]
)
return process_aggregate_results(
results,
query=aggregation_query,
storage_type=self.schema.index.storage_type,
)
[docs]
async def aggregate(self, *args, **kwargs) -> "AggregateResult":
"""Perform an aggregation operation against the index.
Wrapper around the aggregation API that adds the index name
to the query and passes along the rest of the arguments
to the redis-py ft().aggregate() method.
Returns:
Result: Raw Redis aggregation results.
"""
client = await self._get_client()
try:
return await client.ft(self.schema.index.name).aggregate(*args, **kwargs)
except redis.exceptions.RedisError as e:
if "CROSSSLOT" in str(e):
raise RedisSearchError(
"Cross-slot error during aggregation. Ensure consistent hash tags in your keys."
)
raise RedisSearchError(f"Error while aggregating: {str(e)}") from e
except Exception as e:
raise RedisSearchError(
f"Unexpected error while aggregating: {str(e)}"
) from e
[docs]
async def batch_search(
self, queries: list[SearchParams], batch_size: int = 10
) -> list["Result"]:
"""Asynchronously execute a batch of search queries.
This method takes a list of search queries and executes them in batches
to improve performance when dealing with multiple queries.
NOTE: Cluster users may need to incorporate hash tags into their query
to avoid cross-slot operations.
Args:
queries (List[SearchParams]): A list of search queries to execute.
Each query can be either a string or a tuple of (query, params).
batch_size (int, optional): The number of queries to execute in each
batch. Defaults to 10.
Returns:
List[Result]: A list of search results corresponding to each query.
.. code-block:: python
queries = [
"hello world",
("goodbye world", {"num_results": 5}),
]
results = await index.batch_search(queries)
"""
all_results = []
client = await self._get_client()
search = client.ft(self.schema.index.name)
options = {}
if get_protocol_version(client) not in ["3", 3]:
options[NEVER_DECODE] = True
for i in range(0, len(queries), batch_size):
batch_queries = queries[i : i + batch_size]
# redis-py doesn't support calling `search` in an async pipeline,
# so we need to manually execute each command in a pipeline
# and parse the results
async with client.pipeline(transaction=False) as pipe:
batch_built_queries = []
for query in batch_queries:
if isinstance(query, tuple):
query_args, q = search._mk_query_args( # type: ignore
query[0], query_params=query[1]
)
else:
query_args, q = search._mk_query_args( # type: ignore
query, query_params=None
)
batch_built_queries.append(q)
pipe.execute_command(
"FT.SEARCH",
*query_args,
**options,
)
st = time.time()
results = await pipe.execute()
# We don't know how long each query took, so we'll use the total time
# for all queries in the batch as the duration for each query
duration = (time.time() - st) * 1000.0
for j, query_results in enumerate(results):
_built_query = batch_built_queries[j]
parsed_result = _parse_batch_search_result(
search, query_results, _built_query, duration
)
# Return a parsed Result object for each query
all_results.append(parsed_result)
return all_results
[docs]
async def search(self, *args, **kwargs) -> "Result":
"""Perform an async search against the index.
Wrapper around the search API that adds the index name
to the query and passes along the rest of the arguments
to the redis-py ft().search() method.
Returns:
Result: Raw Redis search results.
"""
try:
client = await self._get_client()
if isinstance(client, AsyncRedisCluster):
# Use special cluster search for AsyncRedisCluster
return await async_cluster_search(
client.ft(self.schema.index.name),
*args,
**kwargs, # type: ignore
)
else:
return await client.ft(self.schema.index.name).search(*args, **kwargs) # type: ignore
except redis.exceptions.RedisError as e:
if "CROSSSLOT" in str(e):
raise RedisSearchError(
"Cross-slot error during search. Ensure consistent hash tags in your keys."
)
raise RedisSearchError(f"Error while searching: {str(e)}") from e
except Exception as e:
raise RedisSearchError(f"Unexpected error while searching: {str(e)}") from e
async def _hybrid_search(
self, query: HybridQuery, **kwargs
) -> list[dict[str, Any]]:
"""Perform a hybrid search against the index, combining text and vector search.
Args:
query (HybridQuery): The text+vector search query to be performed, with configurable fusion methods and
post-processing.
kwargs: Additional arguments to pass to the redis-py hybrid_search method (e.g. timeout).
Returns:
List[Dict[str, Any]]: The search results ordered by combined score unless otherwise specified.
Notes:
Hybrid search is only available in Redis 8.4.0+, and requires redis-py >= 7.1.0.
.. code-block:: python
from redisvl.query import HybridQuery
hybrid_query = HybridQuery(
text="lorem ipsum dolor sit amet",
text_field_name="description",
vector=[0.1, 0.2, 0.3],
vector_field_name="embedding"
)
results = await async_index.query(hybrid_query)
"""
client = await self._get_client()
index = client.ft(self.schema.index.name)
self._validate_hybrid_query(query)
if not hasattr(index, "hybrid_search"):
raise ImportError(_HYBRID_SEARCH_ERROR_MESSAGE)
results = await index.hybrid_search(
query=query.query,
combine_method=query.combination_method,
post_processing=(
query.postprocessing_config
if query.postprocessing_config.build_args()
else None
),
params_substitution=query.params, # type: ignore[arg-type]
**kwargs,
) # type: ignore
return _convert_and_drop_empty_rows(results.results, "hybrid") # type: ignore[union-attr]
[docs]
async def batch_query(
self, queries: list[BaseQuery], batch_size: int = 10
) -> list[list[dict[str, Any]]]:
"""Asynchronously execute a batch of queries and process results."""
results = await self.batch_search(
[(query.query, query.params) for query in queries], batch_size=batch_size
)
all_parsed = []
for query, batch_results in zip(queries, results):
parsed = process_results(
batch_results,
query=query,
schema=self.schema,
)
# Create separate lists of parsed results for each query
# passed in to the batch_search method, so that callers can
# access the results for each query individually
all_parsed.append(parsed)
return all_parsed
async def _query(self, query: BaseQuery) -> list[dict[str, Any]]:
"""Asynchronously execute a query and process results."""
try:
self._validate_query(query)
except QueryValidationError as e:
raise QueryValidationError(f"Invalid query: {str(e)}") from e
results = await self.search(query.query, query_params=query.params)
return process_results(results, query=query, schema=self.schema)
async def _sql_query(self, sql_query: SQLQuery) -> list[dict[str, Any]]:
"""Asynchronously execute a SQL query and return results.
Args:
sql_query: The SQLQuery object containing the SQL statement.
Returns:
List of dictionaries containing the query results.
Raises:
ImportError: If sql-redis package is not installed.
"""
try:
from sql_redis import create_async_executor
except ImportError:
raise ImportError(
"sql-redis is required for SQL query support. "
"Install it with: pip install redisvl[sql-redis]"
)
client = await self._get_client()
sql_redis_options = _get_sql_redis_options(sql_query)
cache_key = _sql_executor_cache_key(sql_redis_options)
sql_executors_lock = getattr(self, "_sql_executors_lock", None)
if sql_executors_lock is None:
sql_executors_lock = asyncio.Lock()
existing_sql_executors_lock = getattr(self, "_sql_executors_lock", None)
if existing_sql_executors_lock is None:
self._sql_executors_lock = sql_executors_lock
else:
sql_executors_lock = existing_sql_executors_lock
async with sql_executors_lock:
executor = self._sql_executors.get(cache_key)
if executor is None:
executor = await create_async_executor(client, **sql_redis_options)
self._sql_executors[cache_key] = executor
# Execute the query with any params asynchronously
result = await executor.execute(sql_query.sql, params=sql_query.params)
return _convert_and_drop_empty_rows(result.rows, "SQL")
[docs]
async def query(
self, query: BaseQuery | AggregationQuery | HybridQuery | SQLQuery
) -> list[dict[str, Any]]:
"""Asynchronously execute a query on the index.
This method takes a BaseQuery, AggregationQuery, HybridQuery, or SQLQuery object
directly, runs the search, and handles post-processing of the search.
Args:
query (Union[BaseQuery, AggregationQuery, HybridQuery, SQLQuery]): The query to run.
Returns:
List[Result]: A list of search results.
.. code-block:: python
from redisvl.query import VectorQuery
query = VectorQuery(
vector=[0.16, -0.34, 0.98, 0.23],
vector_field_name="embedding",
num_results=3
)
results = await index.query(query)
"""
if isinstance(query, AggregationQuery):
return await self._aggregate(query)
elif isinstance(query, SQLQuery):
return await self._sql_query(query)
elif isinstance(query, HybridQuery):
return await self._hybrid_search(query)
else:
return await self._query(query)
[docs]
async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerator:
"""Execute a given query against the index and return results in
paginated batches.
This method accepts a RedisVL query instance, enabling async pagination
of results which allows for subsequent processing over each batch with a
generator.
Args:
query (BaseQuery): The search query to be executed.
page_size (int, optional): The number of results to return in each
batch. Defaults to 30.
Yields:
An async generator yielding non-empty ``SearchResults`` batches.
Raises:
TypeError: If the page_size argument is not of type int.
ValueError: If the page_size argument is less than or equal to zero.
.. code-block:: python
# Iterate over paginated search results in batches of 10
async for result_batch in index.paginate(query, page_size=10):
# Process each batch of results
pass
Note:
The page_size parameter controls the number of items each result
batch contains. Adjust this value based on performance
considerations and the expected volume of search results.
Note:
For stable pagination, the query must have a `sort_by` clause on a
**unique** field. Redis documents that ``LIMIT`` without sorting is
non-deterministic, so pages may otherwise repeat or miss documents.
Very deep pagination is also bounded server-side by
``search-max-search-results`` (1,000,000 by default, but 10,000 on
some managed tiers), past which the search errors rather than ending.
Note:
A yielded batch may contain fewer than ``page_size`` documents, and an
empty batch is never yielded. On Redis 8+ a matched document whose
data expires while the search is running is skipped; each batch
reports how many were skipped via ``dropped_count`` (and
``complete``), including skips inherited from a page that was dropped
in its entirety. Pagination still runs to the end of the result set in
that case.
"""
if not isinstance(page_size, int):
raise TypeError("page_size must be of type int")
if page_size <= 0:
raise ValueError("page_size must be greater than 0")
if isinstance(query, CountQuery):
raise TypeError(
"CountQuery cannot be paginated: it returns a match count rather "
"than documents. Use index.query(query) instead."
)
first = 0
carried_drops = 0
while True:
query.paging(first, page_size)
results = await self._query(query)
if not _page_had_matches(results):
break
if results:
_fold_carried_drops(results, carried_drops)
carried_drops = 0
yield results
else:
# Whole page dropped: keep its count so it reaches the caller on
# the next yielded batch instead of vanishing with the page.
carried_drops += getattr(results, "dropped_count", 0)
# Advance unconditionally, so a page we cannot materialize can never
# wedge the loop.
first += page_size
[docs]
async def listall(self) -> list[str]:
"""List all search indices in Redis database.
Returns:
List[str]: The list of indices in the database.
"""
client = await self._get_client()
if isinstance(client, AsyncRedisCluster):
target_nodes = client.get_random_node()
return convert_bytes(await target_nodes.execute_command("FT._LIST"))
else:
return convert_bytes(await client.execute_command("FT._LIST"))
[docs]
async def exists(self) -> bool:
"""Check if the index exists in Redis.
Returns:
bool: True if the index exists, False otherwise.
Raises:
RedisSearchError: If the check fails for any reason other than the
index not existing, such as a connection failure or insufficient
permissions. The original ``redis-py`` exception is available as
``__cause__``.
"""
client = await self._get_client()
try:
info = await self._info(self.schema.index.name, client)
except RedisSearchError as e:
if _is_missing_index_error(e):
return False
raise
# FT.INFO answers for an alias's target, so compare the resolved name:
# reporting an alias as existing would let create(overwrite=True) drop
# the index it points at.
return info.get("index_name") == self.schema.index.name
[docs]
async def info(self, name: str | None = None) -> dict[str, Any]:
"""Get information about the index.
Args:
name (str, optional): Index name to fetch info about.
Defaults to None.
Returns:
dict: A dictionary containing the information about the index.
"""
client = await self._get_client()
index_name = name or self.schema.index.name
return await self._info(index_name, client)
[docs]
async def disconnect(self):
self.invalidate_sql_schema_cache()
if self._owns_redis_client is False:
return
self._detach_client_finalizer()
if self._redis_client is not None:
await self._redis_client.aclose()
self._redis_client = None
def disconnect_sync(self):
if self._redis_client is None or self._owns_redis_client is False:
return
sync_wrapper(self.disconnect)()
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.disconnect()
