# Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import collections.abc import copy from typing import ( TYPE_CHECKING, Any, Callable, Dict, Generic, Iterator, List, Optional, Protocol, Tuple, Type, Union, cast, overload, ) from typing_extensions import Self, TypeVar from .aggs import A, Agg, AggBase from .document_base import InstrumentedField from .exceptions import IllegalOperation from .query import Bool, Q, Query from .response import Hit, Response from .utils import _R, AnyUsingType, AttrDict, DslBase, recursive_to_dict if TYPE_CHECKING: from .field import Field, Object class SupportsClone(Protocol): def _clone(self) -> Self: ... _S = TypeVar("_S", bound=SupportsClone) class QueryProxy(Generic[_S]): """ Simple proxy around DSL objects (queries) that can be called (to add query/post_filter) and also allows attribute access which is proxied to the wrapped query. """ def __init__(self, search: _S, attr_name: str): self._search = search self._proxied: Optional[Query] = None self._attr_name = attr_name def __nonzero__(self) -> bool: return self._proxied is not None __bool__ = __nonzero__ def __call__(self, *args: Any, **kwargs: Any) -> _S: """ Add a query. """ s = self._search._clone() # we cannot use self._proxied since we just cloned self._search and # need to access the new self on the clone proxied = getattr(s, self._attr_name) if proxied._proxied is None: proxied._proxied = Q(*args, **kwargs) else: proxied._proxied &= Q(*args, **kwargs) # always return search to be chainable return s def __getattr__(self, attr_name: str) -> Any: return getattr(self._proxied, attr_name) def __setattr__(self, attr_name: str, value: Any) -> None: if not attr_name.startswith("_"): if self._proxied is not None: self._proxied = Q(self._proxied.to_dict()) setattr(self._proxied, attr_name, value) super().__setattr__(attr_name, value) def __getstate__(self) -> Tuple[_S, Optional[Query], str]: return self._search, self._proxied, self._attr_name def __setstate__(self, state: Tuple[_S, Optional[Query], str]) -> None: self._search, self._proxied, self._attr_name = state class ProxyDescriptor(Generic[_S]): """ Simple descriptor to enable setting of queries and filters as: s = Search() s.query = Q(...) """ def __init__(self, name: str): self._attr_name = f"_{name}_proxy" def __get__(self, instance: Any, owner: object) -> QueryProxy[_S]: return cast(QueryProxy[_S], getattr(instance, self._attr_name)) def __set__(self, instance: _S, value: Dict[str, Any]) -> None: proxy: QueryProxy[_S] = getattr(instance, self._attr_name) proxy._proxied = Q(value) class AggsProxy(AggBase[_R], DslBase): name = "aggs" def __init__(self, search: "SearchBase[_R]"): self._base = cast("Agg[_R]", self) self._search = search self._params = {"aggs": {}} def to_dict(self) -> Dict[str, Any]: return cast(Dict[str, Any], super().to_dict().get("aggs", {})) class Request(Generic[_R]): def __init__( self, using: AnyUsingType = "default", index: Optional[Union[str, List[str]]] = None, doc_type: Optional[ Union[type, str, List[Union[type, str]], Dict[str, Union[type, str]]] ] = None, extra: Optional[Dict[str, Any]] = None, ): self._using = using self._index = None if isinstance(index, (tuple, list)): self._index = list(index) elif index: self._index = [index] self._doc_type: List[Union[type, str]] = [] self._doc_type_map: Dict[str, Any] = {} if isinstance(doc_type, (tuple, list)): self._doc_type.extend(doc_type) elif isinstance(doc_type, collections.abc.Mapping): self._doc_type.extend(doc_type.keys()) self._doc_type_map.update(doc_type) elif doc_type: self._doc_type.append(doc_type) self._params: Dict[str, Any] = {} self._extra: Dict[str, Any] = extra or {} def __eq__(self, other: Any) -> bool: return ( isinstance(other, Request) and other._params == self._params and other._index == self._index and other._doc_type == self._doc_type and other.to_dict() == self.to_dict() ) def __copy__(self) -> Self: return self._clone() def params(self, **kwargs: Any) -> Self: """ Specify query params to be used when executing the search. All the keyword arguments will override the current values. See https://elasticsearch-py.readthedocs.io/en/latest/api/elasticsearch.html#elasticsearch.Elasticsearch.search for all available parameters. Example:: s = Search() s = s.params(routing='user-1', preference='local') """ s = self._clone() s._params.update(kwargs) return s def index(self, *index: Union[str, List[str], Tuple[str, ...]]) -> Self: """ Set the index for the search. If called empty it will remove all information. Example:: s = Search() s = s.index('twitter-2015.01.01', 'twitter-2015.01.02') s = s.index(['twitter-2015.01.01', 'twitter-2015.01.02']) """ # .index() resets s = self._clone() if not index: s._index = None else: indexes = [] for i in index: if isinstance(i, str): indexes.append(i) elif isinstance(i, list): indexes += i elif isinstance(i, tuple): indexes += list(i) s._index = (self._index or []) + indexes return s def _resolve_field(self, path: str) -> Optional["Field"]: for dt in self._doc_type: if not hasattr(dt, "_index"): continue field = dt._index.resolve_field(path) if field is not None: return cast("Field", field) return None def _resolve_nested( self, hit: AttrDict[Any], parent_class: Optional[type] = None ) -> Type[_R]: doc_class = Hit nested_path = [] nesting = hit["_nested"] while nesting and "field" in nesting: nested_path.append(nesting["field"]) nesting = nesting.get("_nested") nested_path_str = ".".join(nested_path) nested_field: Optional["Object"] if parent_class is not None and hasattr(parent_class, "_index"): nested_field = cast( Optional["Object"], parent_class._index.resolve_field(nested_path_str) ) else: nested_field = cast( Optional["Object"], self._resolve_field(nested_path_str) ) if nested_field is not None: return cast(Type[_R], nested_field._doc_class) return cast(Type[_R], doc_class) def _get_result( self, hit: AttrDict[Any], parent_class: Optional[type] = None ) -> _R: doc_class: Any = Hit dt = hit.get("_type") if "_nested" in hit: doc_class = self._resolve_nested(hit, parent_class) elif dt in self._doc_type_map: doc_class = self._doc_type_map[dt] else: for doc_type in self._doc_type: if hasattr(doc_type, "_matches") and doc_type._matches(hit): doc_class = doc_type break for t in hit.get("inner_hits", ()): hit["inner_hits"][t] = Response[_R]( self, hit["inner_hits"][t], doc_class=doc_class ) callback = getattr(doc_class, "from_es", doc_class) return cast(_R, callback(hit)) def doc_type( self, *doc_type: Union[type, str], **kwargs: Callable[[AttrDict[Any]], Any] ) -> Self: """ Set the type to search through. You can supply a single value or multiple. Values can be strings or subclasses of ``Document``. You can also pass in any keyword arguments, mapping a doc_type to a callback that should be used instead of the Hit class. If no doc_type is supplied any information stored on the instance will be erased. Example: s = Search().doc_type('product', 'store', User, custom=my_callback) """ # .doc_type() resets s = self._clone() if not doc_type and not kwargs: s._doc_type = [] s._doc_type_map = {} else: s._doc_type.extend(doc_type) s._doc_type.extend(kwargs.keys()) s._doc_type_map.update(kwargs) return s def using(self, client: AnyUsingType) -> Self: """ Associate the search request with an elasticsearch client. A fresh copy will be returned with current instance remaining unchanged. :arg client: an instance of ``elasticsearch.Elasticsearch`` to use or an alias to look up in ``elasticsearch.dsl.connections`` """ s = self._clone() s._using = client return s def extra(self, **kwargs: Any) -> Self: """ Add extra keys to the request body. Mostly here for backwards compatibility. """ s = self._clone() if "from_" in kwargs: kwargs["from"] = kwargs.pop("from_") s._extra.update(kwargs) return s def _clone(self) -> Self: s = self.__class__( using=self._using, index=self._index, doc_type=self._doc_type ) s._doc_type_map = self._doc_type_map.copy() s._extra = self._extra.copy() s._params = self._params.copy() return s if TYPE_CHECKING: def to_dict(self) -> Dict[str, Any]: ... class SearchBase(Request[_R]): query = ProxyDescriptor[Self]("query") post_filter = ProxyDescriptor[Self]("post_filter") _response: Response[_R] def __init__( self, using: AnyUsingType = "default", index: Optional[Union[str, List[str]]] = None, **kwargs: Any, ): """ Search request to elasticsearch. :arg using: `Elasticsearch` instance to use :arg index: limit the search to index All the parameters supplied (or omitted) at creation type can be later overridden by methods (`using`, `index` and `doc_type` respectively). """ super().__init__(using=using, index=index, **kwargs) self.aggs = AggsProxy[_R](self) self._sort: List[Union[str, Dict[str, Dict[str, str]]]] = [] self._knn: List[Dict[str, Any]] = [] self._rank: Dict[str, Any] = {} self._collapse: Dict[str, Any] = {} self._source: Optional[Union[bool, List[str], Dict[str, List[str]]]] = None self._highlight: Dict[str, Any] = {} self._highlight_opts: Dict[str, Any] = {} self._suggest: Dict[str, Any] = {} self._script_fields: Dict[str, Any] = {} self._response_class = Response[_R] self._query_proxy = QueryProxy(self, "query") self._post_filter_proxy = QueryProxy(self, "post_filter") def filter(self, *args: Any, **kwargs: Any) -> Self: """ Add a query in filter context. """ return self.query(Bool(filter=[Q(*args, **kwargs)])) def exclude(self, *args: Any, **kwargs: Any) -> Self: """ Add a negative query in filter context. """ return self.query(Bool(filter=[~Q(*args, **kwargs)])) def __getitem__(self, n: Union[int, slice]) -> Self: """ Support slicing the `Search` instance for pagination. Slicing equates to the from/size parameters. E.g.:: s = Search().query(...)[0:25] is equivalent to:: s = Search().query(...).extra(from_=0, size=25) """ s = self._clone() if isinstance(n, slice): # If negative slicing, abort. if n.start and n.start < 0 or n.stop and n.stop < 0: raise ValueError("Search does not support negative slicing.") slice_start = n.start slice_stop = n.stop else: # This is an index lookup, equivalent to slicing by [n:n+1]. # If negative index, abort. if n < 0: raise ValueError("Search does not support negative indexing.") slice_start = n slice_stop = n + 1 old_from = s._extra.get("from") old_to = None if "size" in s._extra: old_to = (old_from or 0) + s._extra["size"] new_from = old_from if slice_start is not None: new_from = (old_from or 0) + slice_start new_to = old_to if slice_stop is not None: new_to = (old_from or 0) + slice_stop if old_to is not None and old_to < new_to: new_to = old_to if new_from is not None: s._extra["from"] = new_from if new_to is not None: s._extra["size"] = max(0, new_to - (new_from or 0)) return s @classmethod def from_dict(cls, d: Dict[str, Any]) -> Self: """ Construct a new `Search` instance from a raw dict containing the search body. Useful when migrating from raw dictionaries. Example:: s = Search.from_dict({ "query": { "bool": { "must": [...] } }, "aggs": {...} }) s = s.filter('term', published=True) """ s = cls() s.update_from_dict(d) return s def _clone(self) -> Self: """ Return a clone of the current search request. Performs a shallow copy of all the underlying objects. Used internally by most state modifying APIs. """ s = super()._clone() s._response_class = self._response_class s._knn = [knn.copy() for knn in self._knn] s._rank = self._rank.copy() s._collapse = self._collapse.copy() s._sort = self._sort[:] s._source = copy.copy(self._source) if self._source is not None else None s._highlight = self._highlight.copy() s._highlight_opts = self._highlight_opts.copy() s._suggest = self._suggest.copy() s._script_fields = self._script_fields.copy() for x in ("query", "post_filter"): getattr(s, x)._proxied = getattr(self, x)._proxied # copy top-level bucket definitions if self.aggs._params.get("aggs"): s.aggs._params = {"aggs": self.aggs._params["aggs"].copy()} return s def response_class(self, cls: Type[Response[_R]]) -> Self: """ Override the default wrapper used for the response. """ s = self._clone() s._response_class = cls return s def update_from_dict(self, d: Dict[str, Any]) -> Self: """ Apply options from a serialized body to the current instance. Modifies the object in-place. Used mostly by ``from_dict``. """ d = d.copy() if "query" in d: self.query._proxied = Q(d.pop("query")) if "post_filter" in d: self.post_filter._proxied = Q(d.pop("post_filter")) aggs = d.pop("aggs", d.pop("aggregations", {})) if aggs: self.aggs._params = { "aggs": {name: A(value) for (name, value) in aggs.items()} } if "knn" in d: self._knn = d.pop("knn") if isinstance(self._knn, dict): self._knn = [self._knn] if "rank" in d: self._rank = d.pop("rank") if "collapse" in d: self._collapse = d.pop("collapse") if "sort" in d: self._sort = d.pop("sort") if "_source" in d: self._source = d.pop("_source") if "highlight" in d: high = d.pop("highlight").copy() self._highlight = high.pop("fields") self._highlight_opts = high if "suggest" in d: self._suggest = d.pop("suggest") if "text" in self._suggest: text = self._suggest.pop("text") for s in self._suggest.values(): s.setdefault("text", text) if "script_fields" in d: self._script_fields = d.pop("script_fields") self._extra.update(d) return self def script_fields(self, **kwargs: Any) -> Self: """ Define script fields to be calculated on hits. See https://www.elastic.co/guide/en/elasticsearch/reference/current/search-request-script-fields.html for more details. Example:: s = Search() s = s.script_fields(times_two="doc['field'].value * 2") s = s.script_fields( times_three={ 'script': { 'lang': 'painless', 'source': "doc['field'].value * params.n", 'params': {'n': 3} } } ) """ s = self._clone() for name in kwargs: if isinstance(kwargs[name], str): kwargs[name] = {"script": kwargs[name]} s._script_fields.update(kwargs) return s def knn( self, field: Union[str, "InstrumentedField"], k: int, num_candidates: int, query_vector: Optional[List[float]] = None, query_vector_builder: Optional[Dict[str, Any]] = None, boost: Optional[float] = None, filter: Optional[Query] = None, similarity: Optional[float] = None, inner_hits: Optional[Dict[str, Any]] = None, ) -> Self: """ Add a k-nearest neighbor (kNN) search. :arg field: the vector field to search against as a string or document class attribute :arg k: number of nearest neighbors to return as top hits :arg num_candidates: number of nearest neighbor candidates to consider per shard :arg query_vector: the vector to search for :arg query_vector_builder: A dictionary indicating how to build a query vector :arg boost: A floating-point boost factor for kNN scores :arg filter: query to filter the documents that can match :arg similarity: the minimum similarity required for a document to be considered a match, as a float value :arg inner_hits: retrieve hits from nested field Example:: s = Search() s = s.knn(field='embedding', k=5, num_candidates=10, query_vector=vector, filter=Q('term', category='blog'))) """ s = self._clone() s._knn.append( { "field": str(field), # str() is for InstrumentedField instances "k": k, "num_candidates": num_candidates, } ) if query_vector is None and query_vector_builder is None: raise ValueError("one of query_vector and query_vector_builder is required") if query_vector is not None and query_vector_builder is not None: raise ValueError( "only one of query_vector and query_vector_builder must be given" ) if query_vector is not None: s._knn[-1]["query_vector"] = cast(Any, query_vector) if query_vector_builder is not None: s._knn[-1]["query_vector_builder"] = query_vector_builder if boost is not None: s._knn[-1]["boost"] = boost if filter is not None: if isinstance(filter, Query): s._knn[-1]["filter"] = filter.to_dict() else: s._knn[-1]["filter"] = filter if similarity is not None: s._knn[-1]["similarity"] = similarity if inner_hits is not None: s._knn[-1]["inner_hits"] = inner_hits return s def rank(self, rrf: Optional[Union[bool, Dict[str, Any]]] = None) -> Self: """ Defines a method for combining and ranking results sets from a combination of searches. Requires a minimum of 2 results sets. :arg rrf: Set to ``True`` or an options dictionary to set the rank method to reciprocal rank fusion (RRF). Example:: s = Search() s = s.query('match', content='search text') s = s.knn(field='embedding', k=5, num_candidates=10, query_vector=vector) s = s.rank(rrf=True) Note: This option is in technical preview and may change in the future. The syntax will likely change before GA. """ s = self._clone() s._rank = {} if rrf is not None and rrf is not False: s._rank["rrf"] = {} if rrf is True else rrf return s def source( self, fields: Optional[ Union[ bool, str, "InstrumentedField", List[Union[str, "InstrumentedField"]], Dict[str, List[Union[str, "InstrumentedField"]]], ] ] = None, **kwargs: Any, ) -> Self: """ Selectively control how the _source field is returned. :arg fields: field name, wildcard string, list of field names or wildcards, or dictionary of includes and excludes :arg kwargs: ``includes`` or ``excludes`` arguments, when ``fields`` is ``None``. When no arguments are given, the entire document will be returned for each hit. If ``fields`` is a string or list of strings, the field names or field wildcards given will be included. If ``fields`` is a dictionary with keys of 'includes' and/or 'excludes' the fields will be either included or excluded appropriately. Calling this multiple times with the same named parameter will override the previous values with the new ones. Example:: s = Search() s = s.source(includes=['obj1.*'], excludes=["*.description"]) s = Search() s = s.source(includes=['obj1.*']).source(excludes=["*.description"]) """ s = self._clone() if fields and kwargs: raise ValueError("You cannot specify fields and kwargs at the same time.") @overload def ensure_strings(fields: str) -> str: ... @overload def ensure_strings(fields: "InstrumentedField") -> str: ... @overload def ensure_strings( fields: List[Union[str, "InstrumentedField"]], ) -> List[str]: ... @overload def ensure_strings( fields: Dict[str, List[Union[str, "InstrumentedField"]]], ) -> Dict[str, List[str]]: ... def ensure_strings( fields: Union[ str, "InstrumentedField", List[Union[str, "InstrumentedField"]], Dict[str, List[Union[str, "InstrumentedField"]]], ], ) -> Union[str, List[str], Dict[str, List[str]]]: if isinstance(fields, dict): return {k: ensure_strings(v) for k, v in fields.items()} elif not isinstance(fields, (str, InstrumentedField)): # we assume that if `fields` is not a any of [dict, str, # InstrumentedField] then it is an iterable of strings or # InstrumentedFields, so we convert them to a plain list of # strings return [str(f) for f in fields] else: return str(fields) if fields is not None: s._source = fields if isinstance(fields, bool) else ensure_strings(fields) # type: ignore[assignment] return s if kwargs and not isinstance(s._source, dict): s._source = {} if isinstance(s._source, dict): for key, value in kwargs.items(): if value is None: try: del s._source[key] except KeyError: pass else: s._source[key] = ensure_strings(value) return s def sort( self, *keys: Union[str, "InstrumentedField", Dict[str, Dict[str, str]]] ) -> Self: """ Add sorting information to the search request. If called without arguments it will remove all sort requirements. Otherwise it will replace them. Acceptable arguments are:: 'some.field' '-some.other.field' {'different.field': {'any': 'dict'}} so for example:: s = Search().sort( 'category', '-title', {"price" : {"order" : "asc", "mode" : "avg"}} ) will sort by ``category``, ``title`` (in descending order) and ``price`` in ascending order using the ``avg`` mode. The API returns a copy of the Search object and can thus be chained. """ s = self._clone() s._sort = [] for k in keys: if not isinstance(k, dict): sort_field = str(k) if sort_field.startswith("-"): if sort_field[1:] == "_score": raise IllegalOperation("Sorting by `-_score` is not allowed.") s._sort.append({sort_field[1:]: {"order": "desc"}}) else: s._sort.append(sort_field) else: s._sort.append(k) return s def collapse( self, field: Optional[Union[str, "InstrumentedField"]] = None, inner_hits: Optional[Dict[str, Any]] = None, max_concurrent_group_searches: Optional[int] = None, ) -> Self: """ Add collapsing information to the search request. If called without providing ``field``, it will remove all collapse requirements, otherwise it will replace them with the provided arguments. The API returns a copy of the Search object and can thus be chained. """ s = self._clone() s._collapse = {} if field is None: return s s._collapse["field"] = str(field) if inner_hits: s._collapse["inner_hits"] = inner_hits if max_concurrent_group_searches: s._collapse["max_concurrent_group_searches"] = max_concurrent_group_searches return s def highlight_options(self, **kwargs: Any) -> Self: """ Update the global highlighting options used for this request. For example:: s = Search() s = s.highlight_options(order='score') """ s = self._clone() s._highlight_opts.update(kwargs) return s def highlight( self, *fields: Union[str, "InstrumentedField"], **kwargs: Any ) -> Self: """ Request highlighting of some fields. All keyword arguments passed in will be used as parameters for all the fields in the ``fields`` parameter. Example:: Search().highlight('title', 'body', fragment_size=50) will produce the equivalent of:: { "highlight": { "fields": { "body": {"fragment_size": 50}, "title": {"fragment_size": 50} } } } If you want to have different options for different fields you can call ``highlight`` twice:: Search().highlight('title', fragment_size=50).highlight('body', fragment_size=100) which will produce:: { "highlight": { "fields": { "body": {"fragment_size": 100}, "title": {"fragment_size": 50} } } } """ s = self._clone() for f in fields: s._highlight[str(f)] = kwargs return s def suggest( self, name: str, text: Optional[str] = None, regex: Optional[str] = None, **kwargs: Any, ) -> Self: """ Add a suggestions request to the search. :arg name: name of the suggestion :arg text: text to suggest on All keyword arguments will be added to the suggestions body. For example:: s = Search() s = s.suggest('suggestion-1', 'Elasticsearch', term={'field': 'body'}) # regex query for Completion Suggester s = Search() s = s.suggest('suggestion-1', regex='py[thon|py]', completion={'field': 'body'}) """ if text is None and regex is None: raise ValueError('You have to pass "text" or "regex" argument.') if text and regex: raise ValueError('You can only pass either "text" or "regex" argument.') if regex and "completion" not in kwargs: raise ValueError( '"regex" argument must be passed with "completion" keyword argument.' ) s = self._clone() if regex: s._suggest[name] = {"regex": regex} elif text: if "completion" in kwargs: s._suggest[name] = {"prefix": text} else: s._suggest[name] = {"text": text} s._suggest[name].update(kwargs) return s def search_after(self) -> Self: """ Return a ``Search`` instance that retrieves the next page of results. This method provides an easy way to paginate a long list of results using the ``search_after`` option. For example:: page_size = 20 s = Search()[:page_size].sort("date") while True: # get a page of results r = await s.execute() # do something with this page of results # exit the loop if we reached the end if len(r.hits) < page_size: break # get a search object with the next page of results s = s.search_after() Note that the ``search_after`` option requires the search to have an explicit ``sort`` order. """ if not hasattr(self, "_response"): raise ValueError("A search must be executed before using search_after") return cast(Self, self._response.search_after()) def to_dict(self, count: bool = False, **kwargs: Any) -> Dict[str, Any]: """ Serialize the search into the dictionary that will be sent over as the request's body. :arg count: a flag to specify if we are interested in a body for count - no aggregations, no pagination bounds etc. All additional keyword arguments will be included into the dictionary. """ d = {} if self.query: d["query"] = recursive_to_dict(self.query) if self._knn: if len(self._knn) == 1: d["knn"] = self._knn[0] else: d["knn"] = self._knn if self._rank: d["rank"] = self._rank # count request doesn't care for sorting and other things if not count: if self.post_filter: d["post_filter"] = recursive_to_dict(self.post_filter.to_dict()) if self.aggs.aggs: d.update(recursive_to_dict(self.aggs.to_dict())) if self._sort: d["sort"] = self._sort if self._collapse: d["collapse"] = self._collapse d.update(recursive_to_dict(self._extra)) if self._source not in (None, {}): d["_source"] = self._source if self._highlight: d["highlight"] = {"fields": self._highlight} d["highlight"].update(self._highlight_opts) if self._suggest: d["suggest"] = self._suggest if self._script_fields: d["script_fields"] = self._script_fields d.update(recursive_to_dict(kwargs)) return d class MultiSearchBase(Request[_R]): """ Combine multiple :class:`~elasticsearch.dsl.Search` objects into a single request. """ def __init__(self, **kwargs: Any): super().__init__(**kwargs) self._searches: List[SearchBase[_R]] = [] def __getitem__(self, key: Union[int, slice]) -> Any: return self._searches[key] def __iter__(self) -> Iterator[SearchBase[_R]]: return iter(self._searches) def _clone(self) -> Self: ms = super()._clone() ms._searches = self._searches[:] return ms def add(self, search: SearchBase[_R]) -> Self: """ Adds a new :class:`~elasticsearch.dsl.Search` object to the request:: ms = MultiSearch(index='my-index') ms = ms.add(Search(doc_type=Category).filter('term', category='python')) ms = ms.add(Search(doc_type=Blog)) """ ms = self._clone() ms._searches.append(search) return ms def to_dict(self) -> List[Dict[str, Any]]: # type: ignore[override] out: List[Dict[str, Any]] = [] for s in self._searches: meta: Dict[str, Any] = {} if s._index: meta["index"] = cast(Any, s._index) meta.update(s._params) out.append(meta) out.append(s.to_dict()) return out