Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py # 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.	787 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 document.py # 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 from typing import ( TYPE_CHECKING, Any, AsyncIterable, Dict, List, Optional, Tuple, Union, cast, ) from typing_extensions import Self, dataclass_transform from elasticsearch.exceptions import NotFoundError, RequestError from elasticsearch.helpers import async_bulk from .._async.index import AsyncIndex from ..async_connections import get_connection from ..document_base import DocumentBase, DocumentMeta, mapped_field from ..exceptions import IllegalOperation from ..utils import DOC_META_FIELDS, META_FIELDS, AsyncUsingType, merge from .search import AsyncSearch if TYPE_CHECKING: from elasticsearch import AsyncElasticsearch class AsyncIndexMeta(DocumentMeta): _index: AsyncIndex # global flag to guard us from associating an Index with the base Document # class, only user defined subclasses should have an _index attr _document_initialized = False def __new__( cls, name: str, bases: Tuple[type, ...], attrs: Dict[str, Any] ) -> "AsyncIndexMeta": new_cls = super().__new__(cls, name, bases, attrs) if cls._document_initialized: index_opts = attrs.pop("Index", None) index = cls.construct_index(index_opts, bases) new_cls._index = index index.document(new_cls) cls._document_initialized = True return cast(AsyncIndexMeta, new_cls) @classmethod def construct_index( cls, opts: Dict[str, Any], bases: Tuple[type, ...] ) -> AsyncIndex: if opts is None: for b in bases: if hasattr(b, "_index"): return b._index # Set None as Index name so it will set _all while making the query return AsyncIndex(name=None) i = AsyncIndex( getattr(opts, "name", ""), using=getattr(opts, "using", "default") ) i.settings(getattr(opts, "settings", {})) i.aliases(getattr(opts, "aliases", {})) for a in getattr(opts, "analyzers", ()): i.analyzer(a) return i @dataclass_transform(field_specifiers=(mapped_field,)) class AsyncDocument(DocumentBase, metaclass=AsyncIndexMeta): """ Model-like class for persisting documents in elasticsearch. """ if TYPE_CHECKING: _index: AsyncIndex @classmethod def _get_using(cls, using: Optional[AsyncUsingType] = None) -> AsyncUsingType: return cast(AsyncUsingType, using or cls._index._using) @classmethod def _get_connection( cls, using: Optional[AsyncUsingType] = None ) -> "AsyncElasticsearch": return get_connection(cls._get_using(using)) @classmethod async def init( cls, index: Optional[str] = None, using: Optional[AsyncUsingType] = None ) -> None: """ Create the index and populate the mappings in elasticsearch. """ i = cls._index if index: i = i.clone(name=index) await i.save(using=using) @classmethod def search( cls, using: Optional[AsyncUsingType] = None, index: Optional[str] = None ) -> AsyncSearch[Self]: """ Create an :class:`~elasticsearch.dsl.Search` instance that will search over this ``Document``. """ return AsyncSearch( using=cls._get_using(using), index=cls._default_index(index), doc_type=[cls] ) @classmethod async def get( cls, id: str, using: Optional[AsyncUsingType] = None, index: Optional[str] = None, kwargs: Any, ) -> Optional[Self]: """ Retrieve a single document from elasticsearch using its ``id``. :arg id: ``id`` of the document to be retrieved :arg index: elasticsearch index to use, if the ``Document`` is associated with an index this can be omitted. :arg using: connection alias to use, defaults to ``'default'`` Any additional keyword arguments will be passed to ``Elasticsearch.get`` unchanged. """ es = cls._get_connection(using) doc = await es.get(index=cls._default_index(index), id=id, kwargs) if not doc.get("found", False): return None return cls.from_es(doc) @classmethod async def exists( cls, id: str, using: Optional[AsyncUsingType] = None, index: Optional[str] = None, kwargs: Any, ) -> bool: """ check if exists a single document from elasticsearch using its ``id``. :arg id: ``id`` of the document to check if exists :arg index: elasticsearch index to use, if the ``Document`` is associated with an index this can be omitted. :arg using: connection alias to use, defaults to ``'default'`` Any additional keyword arguments will be passed to ``Elasticsearch.exists`` unchanged. """ es = cls._get_connection(using) return bool(await es.exists(index=cls._default_index(index), id=id, kwargs)) @classmethod async def mget( cls, docs: List[Dict[str, Any]], using: Optional[AsyncUsingType] = None, index: Optional[str] = None, raise_on_error: bool = True, missing: str = "none", kwargs: Any, ) -> List[Optional[Self]]: r""" Retrieve multiple document by their ``id``\s. Returns a list of instances in the same order as requested. :arg docs: list of ``id``\s of the documents to be retrieved or a list of document specifications as per https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-multi-get.html :arg index: elasticsearch index to use, if the ``Document`` is associated with an index this can be omitted. :arg using: connection alias to use, defaults to ``'default'`` :arg missing: what to do when one of the documents requested is not found. Valid options are ``'none'`` (use ``None``), ``'raise'`` (raise ``NotFoundError``) or ``'skip'`` (ignore the missing document). Any additional keyword arguments will be passed to ``Elasticsearch.mget`` unchanged. """ if missing not in ("raise", "skip", "none"): raise ValueError("'missing' must be 'raise', 'skip', or 'none'.") es = cls._get_connection(using) body = { "docs": [ doc if isinstance(doc, collections.abc.Mapping) else {"_id": doc} for doc in docs ] } results = await es.mget(index=cls._default_index(index), body=body, kwargs) objs: List[Optional[Self]] = [] error_docs: List[Self] = [] missing_docs: List[Self] = [] for doc in results["docs"]: if doc.get("found"): if error_docs or missing_docs: # We're going to raise an exception anyway, so avoid an # expensive call to cls.from_es(). continue objs.append(cls.from_es(doc)) elif doc.get("error"): if raise_on_error: error_docs.append(doc) if missing == "none": objs.append(None) # The doc didn't cause an error, but the doc also wasn't found. elif missing == "raise": missing_docs.append(doc) elif missing == "none": objs.append(None) if error_docs: error_ids = [doc["_id"] for doc in error_docs] message = "Required routing not provided for documents %s." message %= ", ".join(error_ids) raise RequestError(400, message, error_docs) # type: ignore[arg-type] if missing_docs: missing_ids = [doc["_id"] for doc in missing_docs] message = f"Documents {', '.join(missing_ids)} not found." raise NotFoundError(404, message, {"docs": missing_docs}) # type: ignore[arg-type] return objs async def delete( self, using: Optional[AsyncUsingType] = None, index: Optional[str] = None, kwargs: Any, ) -> None: """ Delete the instance in elasticsearch. :arg index: elasticsearch index to use, if the ``Document`` is associated with an index this can be omitted. :arg using: connection alias to use, defaults to ``'default'`` Any additional keyword arguments will be passed to ``Elasticsearch.delete`` unchanged. """ es = self._get_connection(using) # extract routing etc from meta doc_meta = {k: self.meta[k] for k in DOC_META_FIELDS if k in self.meta} # Optimistic concurrency control if "seq_no" in self.meta and "primary_term" in self.meta: doc_meta["if_seq_no"] = self.meta["seq_no"] doc_meta["if_primary_term"] = self.meta["primary_term"] doc_meta.update(kwargs) i = self._get_index(index) assert i is not None await es.delete(index=i, doc_meta) async def update( self, using: Optional[AsyncUsingType] = None, index: Optional[str] = None, detect_noop: bool = True, doc_as_upsert: bool = False, refresh: bool = False, retry_on_conflict: Optional[int] = None, script: Optional[Union[str, Dict[str, Any]]] = None, script_id: Optional[str] = None, scripted_upsert: bool = False, upsert: Optional[Dict[str, Any]] = None, return_doc_meta: bool = False, fields: Any, ) -> Any: """ Partial update of the document, specify fields you wish to update and both the instance and the document in elasticsearch will be updated:: doc = MyDocument(title='Document Title!') doc.save() doc.update(title='New Document Title!') :arg index: elasticsearch index to use, if the ``Document`` is associated with an index this can be omitted. :arg using: connection alias to use, defaults to ``'default'`` :arg detect_noop: Set to ``False`` to disable noop detection. :arg refresh: Control when the changes made by this request are visible to search. Set to ``True`` for immediate effect. :arg retry_on_conflict: In between the get and indexing phases of the update, it is possible that another process might have already updated the same document. By default, the update will fail with a version conflict exception. The retry_on_conflict parameter controls how many times to retry the update before finally throwing an exception. :arg doc_as_upsert: Instead of sending a partial doc plus an upsert doc, setting doc_as_upsert to true will use the contents of doc as the upsert value :arg script: the source code of the script as a string, or a dictionary with script attributes to update. :arg return_doc_meta: set to ``True`` to return all metadata from the index API call instead of only the operation result :return: operation result noop/updated """ body: Dict[str, Any] = { "doc_as_upsert": doc_as_upsert, "detect_noop": detect_noop, } # scripted update if script or script_id: if upsert is not None: body["upsert"] = upsert if script: if isinstance(script, str): script = {"source": script} else: script = {"id": script_id} if "params" not in script: script["params"] = fields else: script["params"].update(fields) body["script"] = script body["scripted_upsert"] = scripted_upsert # partial document update else: if not fields: raise IllegalOperation( "You cannot call update() without updating individual fields or a script. " "If you wish to update the entire object use save()." ) # update given fields locally merge(self, fields) # prepare data for ES values = self.to_dict(skip_empty=False) # if fields were given: partial update body["doc"] = {k: values.get(k) for k in fields.keys()} # extract routing etc from meta doc_meta = {k: self.meta[k] for k in DOC_META_FIELDS if k in self.meta} if retry_on_conflict is not None: doc_meta["retry_on_conflict"] = retry_on_conflict # Optimistic concurrency control if ( retry_on_conflict in (None, 0) and "seq_no" in self.meta and "primary_term" in self.meta ): doc_meta["if_seq_no"] = self.meta["seq_no"] doc_meta["if_primary_term"] = self.meta["primary_term"] i = self._get_index(index) assert i is not None meta = await self._get_connection(using).update( index=i, body=body, refresh=refresh, doc_meta ) # update meta information from ES for k in META_FIELDS: if "_" + k in meta: setattr(self.meta, k, meta["_" + k]) return meta if return_doc_meta else meta["result"] async def save( self, using: Optional[AsyncUsingType] = None, index: Optional[str] = None, validate: bool = True, skip_empty: bool = True, return_doc_meta: bool = False, kwargs: Any, ) -> Any: """ Save the document into elasticsearch. If the document doesn't exist it is created, it is overwritten otherwise. Returns ``True`` if this operations resulted in new document being created. :arg index: elasticsearch index to use, if the ``Document`` is associated with an index this can be omitted. :arg using: connection alias to use, defaults to ``'default'`` :arg validate: set to ``False`` to skip validating the document :arg skip_empty: if set to ``False`` will cause empty values (``None``, ``[]``, ``{}``) to be left on the document. Those values will be stripped out otherwise as they make no difference in elasticsearch. :arg return_doc_meta: set to ``True`` to return all metadata from the update API call instead of only the operation result Any additional keyword arguments will be passed to ``Elasticsearch.index`` unchanged. :return: operation result created/updated """ if validate: self.full_clean() es = self._get_connection(using) # extract routing etc from meta doc_meta = {k: self.meta[k] for k in DOC_META_FIELDS if k in self.meta} # Optimistic concurrency control if "seq_no" in self.meta and "primary_term" in self.meta: doc_meta["if_seq_no"] = self.meta["seq_no"] doc_meta["if_primary_term"] = self.meta["primary_term"] doc_meta.update(kwargs) i = self._get_index(index) assert i is not None meta = await es.index( index=i, body=self.to_dict(skip_empty=skip_empty), doc_meta, ) # update meta information from ES for k in META_FIELDS: if "_" + k in meta: setattr(self.meta, k, meta["_" + k]) return meta if return_doc_meta else meta["result"] @classmethod async def bulk( cls, actions: AsyncIterable[Union[Self, Dict[str, Any]]], using: Optional[AsyncUsingType] = None, index: Optional[str] = None, validate: bool = True, skip_empty: bool = True, kwargs: Any, ) -> Tuple[int, Union[int, List[Any]]]: """ Allows to perform multiple indexing operations in a single request. :arg actions: a generator that returns document instances to be indexed, bulk operation dictionaries. :arg using: connection alias to use, defaults to ``'default'`` :arg index: Elasticsearch index to use, if the ``Document`` is associated with an index this can be omitted. :arg validate: set to ``False`` to skip validating the documents :arg skip_empty: if set to ``False`` will cause empty values (``None``, ``[]``, ``{}``) to be left on the document. Those values will be stripped out otherwise as they make no difference in Elasticsearch. Any additional keyword arguments will be passed to ``Elasticsearch.bulk`` unchanged. :return: bulk operation results """ es = cls._get_connection(using) i = cls._default_index(index) assert i is not None class Generate: def __init__( self, doc_iterator: AsyncIterable[Union[AsyncDocument, Dict[str, Any]]], ): self.doc_iterator = doc_iterator.__aiter__() def __aiter__(self) -> Self: return self async def __anext__(self) -> Dict[str, Any]: doc: Optional[Union[AsyncDocument, Dict[str, Any]]] = ( await self.doc_iterator.__anext__() ) if isinstance(doc, dict): action = doc doc = None if "_source" in action and isinstance( action["_source"], AsyncDocument ): doc = action["_source"] if validate: # pragma: no cover doc.full_clean() action["_source"] = doc.to_dict( include_meta=False, skip_empty=skip_empty ) elif doc is not None: if validate: # pragma: no cover doc.full_clean() action = doc.to_dict(include_meta=True, skip_empty=skip_empty) if "_index" not in action: action["_index"] = i return action return await async_bulk(es, Generate(actions), *kwargs)	18.89 KB	2025-05-25 12:26
📄 faceted_search.py # 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. from typing import TYPE_CHECKING from ..faceted_search_base import FacetedResponse, FacetedSearchBase from ..utils import _R from .search import AsyncSearch if TYPE_CHECKING: from ..response import Response class AsyncFacetedSearch(FacetedSearchBase[_R]): _s: AsyncSearch[_R] async def count(self) -> int: return await self._s.count() def search(self) -> AsyncSearch[_R]: """ Returns the base Search object to which the facets are added. You can customize the query by overriding this method and returning a modified search object. """ s = AsyncSearch[_R](doc_type=self.doc_types, index=self.index, using=self.using) return s.response_class(FacetedResponse) async def execute(self) -> "Response[_R]": """ Execute the search and return the response. """ r = await self._s.execute() r._faceted_search = self return r	1.7 KB	2025-05-25 12:26
📄 index.py # 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. from typing import TYPE_CHECKING, Any, Dict, Optional from typing_extensions import Self from ..async_connections import get_connection from ..exceptions import IllegalOperation from ..index_base import IndexBase from ..utils import AsyncUsingType from .mapping import AsyncMapping from .search import AsyncSearch from .update_by_query import AsyncUpdateByQuery if TYPE_CHECKING: from elastic_transport import ObjectApiResponse from elasticsearch import AsyncElasticsearch class AsyncIndexTemplate: def __init__( self, name: str, template: str, index: Optional["AsyncIndex"] = None, order: Optional[int] = None, kwargs: Any, ): if index is None: self._index = AsyncIndex(template, kwargs) else: if kwargs: raise ValueError( "You cannot specify options for Index when" " passing an Index instance." ) self._index = index.clone() self._index._name = template self._template_name = name self.order = order def __getattr__(self, attr_name: str) -> Any: return getattr(self._index, attr_name) def to_dict(self) -> Dict[str, Any]: d = self._index.to_dict() d["index_patterns"] = [self._index._name] if self.order is not None: d["order"] = self.order return d async def save( self, using: Optional[AsyncUsingType] = None ) -> "ObjectApiResponse[Any]": es = get_connection(using or self._index._using) return await es.indices.put_template( name=self._template_name, body=self.to_dict() ) class AsyncComposableIndexTemplate: def __init__( self, name: str, template: str, index: Optional["AsyncIndex"] = None, priority: Optional[int] = None, kwargs: Any, ): if index is None: self._index = AsyncIndex(template, kwargs) else: if kwargs: raise ValueError( "You cannot specify options for Index when" " passing an Index instance." ) self._index = index.clone() self._index._name = template self._template_name = name self.priority = priority def __getattr__(self, attr_name: str) -> Any: return getattr(self._index, attr_name) def to_dict(self) -> Dict[str, Any]: d: Dict[str, Any] = {"template": self._index.to_dict()} d["index_patterns"] = [self._index._name] if self.priority is not None: d["priority"] = self.priority return d async def save( self, using: Optional[AsyncUsingType] = None ) -> "ObjectApiResponse[Any]": es = get_connection(using or self._index._using) return await es.indices.put_index_template( name=self._template_name, self.to_dict() ) class AsyncIndex(IndexBase): _using: AsyncUsingType if TYPE_CHECKING: def get_or_create_mapping(self) -> AsyncMapping: ... def __init__(self, name: str, using: AsyncUsingType = "default"): """ :arg name: name of the index :arg using: connection alias to use, defaults to ``'default'`` """ super().__init__(name, AsyncMapping, using=using) def _get_connection( self, using: Optional[AsyncUsingType] = None ) -> "AsyncElasticsearch": if self._name is None: raise ValueError("You cannot perform API calls on the default index.") return get_connection(using or self._using) connection = property(_get_connection) def as_template( self, template_name: str, pattern: Optional[str] = None, order: Optional[int] = None, ) -> AsyncIndexTemplate: return AsyncIndexTemplate( template_name, pattern or self._name, index=self, order=order ) def as_composable_template( self, template_name: str, pattern: Optional[str] = None, priority: Optional[int] = None, ) -> AsyncComposableIndexTemplate: return AsyncComposableIndexTemplate( template_name, pattern or self._name, index=self, priority=priority ) async def load_mappings(self, using: Optional[AsyncUsingType] = None) -> None: await self.get_or_create_mapping().update_from_es( self._name, using=using or self._using ) def clone( self, name: Optional[str] = None, using: Optional[AsyncUsingType] = None ) -> Self: """ Create a copy of the instance with another name or connection alias. Useful for creating multiple indices with shared configuration:: i = Index('base-index') i.settings(number_of_shards=1) i.create() i2 = i.clone('other-index') i2.create() :arg name: name of the index :arg using: connection alias to use, defaults to ``'default'`` """ i = self.__class__(name or self._name, using=using or self._using) i._settings = self._settings.copy() i._aliases = self._aliases.copy() i._analysis = self._analysis.copy() i._doc_types = self._doc_types[:] if self._mapping is not None: i._mapping = self._mapping._clone() return i def search(self, using: Optional[AsyncUsingType] = None) -> AsyncSearch: """ Return a :class:`~elasticsearch.dsl.Search` object searching over the index (or all the indices belonging to this template) and its ``Document``\\s. """ return AsyncSearch( using=using or self._using, index=self._name, doc_type=self._doc_types ) def updateByQuery( self, using: Optional[AsyncUsingType] = None ) -> AsyncUpdateByQuery: """ Return a :class:`~elasticsearch.dsl.UpdateByQuery` object searching over the index (or all the indices belonging to this template) and updating Documents that match the search criteria. For more information, see here: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html """ return AsyncUpdateByQuery( using=using or self._using, index=self._name, ) async def create( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Creates the index in elasticsearch. Any additional keyword arguments will be passed to ``Elasticsearch.indices.create`` unchanged. """ return await self._get_connection(using).indices.create( index=self._name, body=self.to_dict(), kwargs ) async def is_closed(self, using: Optional[AsyncUsingType] = None) -> bool: state = await self._get_connection(using).cluster.state( index=self._name, metric="metadata" ) return bool(state["metadata"]["indices"][self._name]["state"] == "close") async def save( self, using: Optional[AsyncUsingType] = None ) -> "Optional[ObjectApiResponse[Any]]": """ Sync the index definition with elasticsearch, creating the index if it doesn't exist and updating its settings and mappings if it does. Note some settings and mapping changes cannot be done on an open index (or at all on an existing index) and for those this method will fail with the underlying exception. """ if not await self.exists(using=using): return await self.create(using=using) body = self.to_dict() settings = body.pop("settings", {}) analysis = settings.pop("analysis", None) current_settings = (await self.get_settings(using=using))[self._name][ "settings" ]["index"] if analysis: if await self.is_closed(using=using): # closed index, update away settings["analysis"] = analysis else: # compare analysis definition, if all analysis objects are # already defined as requested, skip analysis update and # proceed, otherwise raise IllegalOperation existing_analysis = current_settings.get("analysis", {}) if any( existing_analysis.get(section, {}).get(k, None) != analysis[section][k] for section in analysis for k in analysis[section] ): raise IllegalOperation( "You cannot update analysis configuration on an open index, " "you need to close index %s first." % self._name ) # try and update the settings if settings: settings = settings.copy() for k, v in list(settings.items()): if k in current_settings and current_settings[k] == str(v): del settings[k] if settings: await self.put_settings(using=using, body=settings) # update the mappings, any conflict in the mappings will result in an # exception mappings = body.pop("mappings", {}) if mappings: return await self.put_mapping(using=using, body=mappings) return None async def analyze( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Perform the analysis process on a text and return the tokens breakdown of the text. Any additional keyword arguments will be passed to ``Elasticsearch.indices.analyze`` unchanged. """ return await self._get_connection(using).indices.analyze( index=self._name, kwargs ) async def refresh( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Performs a refresh operation on the index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.refresh`` unchanged. """ return await self._get_connection(using).indices.refresh( index=self._name, kwargs ) async def flush( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Performs a flush operation on the index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.flush`` unchanged. """ return await self._get_connection(using).indices.flush( index=self._name, kwargs ) async def get( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ The get index API allows to retrieve information about the index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.get`` unchanged. """ return await self._get_connection(using).indices.get(index=self._name, kwargs) async def open( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Opens the index in elasticsearch. Any additional keyword arguments will be passed to ``Elasticsearch.indices.open`` unchanged. """ return await self._get_connection(using).indices.open( index=self._name, kwargs ) async def close( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Closes the index in elasticsearch. Any additional keyword arguments will be passed to ``Elasticsearch.indices.close`` unchanged. """ return await self._get_connection(using).indices.close( index=self._name, kwargs ) async def delete( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Deletes the index in elasticsearch. Any additional keyword arguments will be passed to ``Elasticsearch.indices.delete`` unchanged. """ return await self._get_connection(using).indices.delete( index=self._name, kwargs ) async def exists( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> bool: """ Returns ``True`` if the index already exists in elasticsearch. Any additional keyword arguments will be passed to ``Elasticsearch.indices.exists`` unchanged. """ return bool( await self._get_connection(using).indices.exists(index=self._name, kwargs) ) async def put_mapping( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Register specific mapping definition for a specific type. Any additional keyword arguments will be passed to ``Elasticsearch.indices.put_mapping`` unchanged. """ return await self._get_connection(using).indices.put_mapping( index=self._name, kwargs ) async def get_mapping( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Retrieve specific mapping definition for a specific type. Any additional keyword arguments will be passed to ``Elasticsearch.indices.get_mapping`` unchanged. """ return await self._get_connection(using).indices.get_mapping( index=self._name, kwargs ) async def get_field_mapping( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Retrieve mapping definition of a specific field. Any additional keyword arguments will be passed to ``Elasticsearch.indices.get_field_mapping`` unchanged. """ return await self._get_connection(using).indices.get_field_mapping( index=self._name, kwargs ) async def put_alias( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Create an alias for the index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.put_alias`` unchanged. """ return await self._get_connection(using).indices.put_alias( index=self._name, kwargs ) async def exists_alias( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> bool: """ Return a boolean indicating whether given alias exists for this index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.exists_alias`` unchanged. """ return bool( await self._get_connection(using).indices.exists_alias( index=self._name, kwargs ) ) async def get_alias( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Retrieve a specified alias. Any additional keyword arguments will be passed to ``Elasticsearch.indices.get_alias`` unchanged. """ return await self._get_connection(using).indices.get_alias( index=self._name, kwargs ) async def delete_alias( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Delete specific alias. Any additional keyword arguments will be passed to ``Elasticsearch.indices.delete_alias`` unchanged. """ return await self._get_connection(using).indices.delete_alias( index=self._name, kwargs ) async def get_settings( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Retrieve settings for the index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.get_settings`` unchanged. """ return await self._get_connection(using).indices.get_settings( index=self._name, kwargs ) async def put_settings( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Change specific index level settings in real time. Any additional keyword arguments will be passed to ``Elasticsearch.indices.put_settings`` unchanged. """ return await self._get_connection(using).indices.put_settings( index=self._name, kwargs ) async def stats( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Retrieve statistics on different operations happening on the index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.stats`` unchanged. """ return await self._get_connection(using).indices.stats( index=self._name, kwargs ) async def segments( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Provide low level segments information that a Lucene index (shard level) is built with. Any additional keyword arguments will be passed to ``Elasticsearch.indices.segments`` unchanged. """ return await self._get_connection(using).indices.segments( index=self._name, kwargs ) async def validate_query( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Validate a potentially expensive query without executing it. Any additional keyword arguments will be passed to ``Elasticsearch.indices.validate_query`` unchanged. """ return await self._get_connection(using).indices.validate_query( index=self._name, kwargs ) async def clear_cache( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Clear all caches or specific cached associated with the index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.clear_cache`` unchanged. """ return await self._get_connection(using).indices.clear_cache( index=self._name, kwargs ) async def recovery( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ The indices recovery API provides insight into on-going shard recoveries for the index. Any additional keyword arguments will be passed to ``Elasticsearch.indices.recovery`` unchanged. """ return await self._get_connection(using).indices.recovery( index=self._name, kwargs ) async def shard_stores( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ Provides store information for shard copies of the index. Store information reports on which nodes shard copies exist, the shard copy version, indicating how recent they are, and any exceptions encountered while opening the shard index or from earlier engine failure. Any additional keyword arguments will be passed to ``Elasticsearch.indices.shard_stores`` unchanged. """ return await self._get_connection(using).indices.shard_stores( index=self._name, kwargs ) async def forcemerge( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ The force merge API allows to force merging of the index through an API. The merge relates to the number of segments a Lucene index holds within each shard. The force merge operation allows to reduce the number of segments by merging them. This call will block until the merge is complete. If the http connection is lost, the request will continue in the background, and any new requests will block until the previous force merge is complete. Any additional keyword arguments will be passed to ``Elasticsearch.indices.forcemerge`` unchanged. """ return await self._get_connection(using).indices.forcemerge( index=self._name, kwargs ) async def shrink( self, using: Optional[AsyncUsingType] = None, kwargs: Any ) -> "ObjectApiResponse[Any]": """ The shrink index API allows you to shrink an existing index into a new index with fewer primary shards. The number of primary shards in the target index must be a factor of the shards in the source index. For example an index with 8 primary shards can be shrunk into 4, 2 or 1 primary shards or an index with 15 primary shards can be shrunk into 5, 3 or 1. If the number of shards in the index is a prime number it can only be shrunk into a single primary shard. Before shrinking, a (primary or replica) copy of every shard in the index must be present on the same node. Any additional keyword arguments will be passed to ``Elasticsearch.indices.shrink`` unchanged. """ return await self._get_connection(using).indices.shrink( index=self._name, **kwargs )	22.23 KB	2025-05-25 12:26
📄 mapping.py # 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. from typing import List, Optional, Union from typing_extensions import Self from ..async_connections import get_connection from ..mapping_base import MappingBase from ..utils import AsyncUsingType class AsyncMapping(MappingBase): @classmethod async def from_es( cls, index: Optional[Union[str, List[str]]], using: AsyncUsingType = "default" ) -> Self: m = cls() await m.update_from_es(index, using) return m async def update_from_es( self, index: Optional[Union[str, List[str]]], using: AsyncUsingType = "default" ) -> None: es = get_connection(using) raw = await es.indices.get_mapping(index=index) _, raw = raw.popitem() self._update_from_dict(raw["mappings"]) async def save(self, index: str, using: AsyncUsingType = "default") -> None: from .index import AsyncIndex i = AsyncIndex(index, using=using) i.mapping(self) await i.save()	1.72 KB	2025-05-25 12:26
📄 search.py # 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 contextlib from typing import ( TYPE_CHECKING, Any, AsyncIterator, Dict, Iterator, List, Optional, cast, ) from typing_extensions import Self from elasticsearch.exceptions import ApiError from elasticsearch.helpers import async_scan from ..async_connections import get_connection from ..response import Response from ..search_base import MultiSearchBase, SearchBase from ..utils import _R, AsyncUsingType, AttrDict class AsyncSearch(SearchBase[_R]): _using: AsyncUsingType def __aiter__(self) -> AsyncIterator[_R]: """ Iterate over the hits. """ class ResultsIterator(AsyncIterator[_R]): def __init__(self, search: AsyncSearch[_R]): self.search = search self.iterator: Optional[Iterator[_R]] = None async def __anext__(self) -> _R: if self.iterator is None: self.iterator = iter(await self.search.execute()) try: return next(self.iterator) except StopIteration: raise StopAsyncIteration() return ResultsIterator(self) async def count(self) -> int: """ Return the number of hits matching the query and filters. Note that only the actual number is returned. """ if hasattr(self, "_response") and self._response.hits.total.relation == "eq": # type: ignore[attr-defined] return cast(int, self._response.hits.total.value) # type: ignore[attr-defined] es = get_connection(self._using) d = self.to_dict(count=True) # TODO: failed shards detection resp = await es.count( index=self._index, query=cast(Optional[Dict[str, Any]], d.get("query", None)), self._params, ) return cast(int, resp["count"]) async def execute(self, ignore_cache: bool = False) -> Response[_R]: """ Execute the search and return an instance of ``Response`` wrapping all the data. :arg ignore_cache: if set to ``True``, consecutive calls will hit ES, while cached result will be ignored. Defaults to `False` """ if ignore_cache or not hasattr(self, "_response"): es = get_connection(self._using) self._response = self._response_class( self, ( await es.search( index=self._index, body=self.to_dict(), self._params ) ).body, ) return self._response async def scan(self) -> AsyncIterator[_R]: """ Turn the search into a scan search and return a generator that will iterate over all the documents matching the query. Use the ``params`` method to specify any additional arguments you wish to pass to the underlying ``scan`` helper from ``elasticsearch-py`` - https://elasticsearch-py.readthedocs.io/en/latest/helpers.html#scan The ``iterate()`` method should be preferred, as it provides similar functionality using an Elasticsearch point in time. """ es = get_connection(self._using) async for hit in async_scan( es, query=self.to_dict(), index=self._index, self._params ): yield self._get_result(cast(AttrDict[Any], hit)) async def delete(self) -> AttrDict[Any]: """ ``delete()`` executes the query by delegating to ``delete_by_query()``. Use the ``params`` method to specify any additional arguments you wish to pass to the underlying ``delete_by_query`` helper from ``elasticsearch-py`` - https://elasticsearch-py.readthedocs.io/en/latest/api/elasticsearch.html#elasticsearch.Elasticsearch.delete_by_query """ es = get_connection(self._using) assert self._index is not None return AttrDict( cast( Dict[str, Any], await es.delete_by_query( index=self._index, body=self.to_dict(), self._params ), ) ) @contextlib.asynccontextmanager async def point_in_time(self, keep_alive: str = "1m") -> AsyncIterator[Self]: """ Open a point in time (pit) that can be used across several searches. This method implements a context manager that returns a search object configured to operate within the created pit. :arg keep_alive: the time to live for the point in time, renewed with each search request """ es = get_connection(self._using) pit = await es.open_point_in_time( index=self._index or "", keep_alive=keep_alive ) search = self.index().extra(pit={"id": pit["id"], "keep_alive": keep_alive}) if not search._sort: search = search.sort("_shard_doc") yield search await es.close_point_in_time(id=pit["id"]) async def iterate(self, keep_alive: str = "1m") -> AsyncIterator[_R]: """ Return a generator that iterates over all the documents matching the query. This method uses a point in time to provide consistent results even when the index is changing. It should be preferred over ``scan()``. :arg keep_alive: the time to live for the point in time, renewed with each new search request """ async with self.point_in_time(keep_alive=keep_alive) as s: while True: r = await s.execute() for hit in r: yield hit if len(r.hits) == 0: break s = s.search_after() class AsyncMultiSearch(MultiSearchBase[_R]): """ Combine multiple :class:`~elasticsearch.dsl.Search` objects into a single request. """ _using: AsyncUsingType if TYPE_CHECKING: def add(self, search: AsyncSearch[_R]) -> Self: ... # type: ignore[override] async def execute( self, ignore_cache: bool = False, raise_on_error: bool = True ) -> List[Response[_R]]: """ Execute the multi search request and return a list of search results. """ if ignore_cache or not hasattr(self, "_response"): es = get_connection(self._using) responses = await es.msearch( index=self._index, body=self.to_dict(), *self._params ) out: List[Response[_R]] = [] for s, r in zip(self._searches, responses["responses"]): if r.get("error", False): if raise_on_error: raise ApiError("N/A", meta=responses.meta, body=r) r = None else: r = Response(s, r) out.append(r) self._response = out return self._response class AsyncEmptySearch(AsyncSearch[_R]): async def count(self) -> int: return 0 async def execute(self, ignore_cache: bool = False) -> Response[_R]: return self._response_class(self, {"hits": {"total": 0, "hits": []}}) async def scan(self) -> AsyncIterator[_R]: return yield # a bit strange, but this forces an empty generator function async def delete(self) -> AttrDict[Any]: return AttrDict[Any]({})	8.03 KB	2025-05-25 12:26
📄 update_by_query.py # 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. from typing import TYPE_CHECKING from ..async_connections import get_connection from ..update_by_query_base import UpdateByQueryBase from ..utils import _R, AsyncUsingType if TYPE_CHECKING: from ..response import UpdateByQueryResponse class AsyncUpdateByQuery(UpdateByQueryBase[_R]): _using: AsyncUsingType async def execute(self) -> "UpdateByQueryResponse[_R]": """ Execute the search and return an instance of ``Response`` wrapping all the data. """ es = get_connection(self._using) assert self._index is not None self._response = self._response_class( self, ( await es.update_by_query( index=self._index, self.to_dict(), self._params ) ).body, ) return self._response	1.6 KB	2025-05-25 12:26

Terminal