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
📁 ..	-	2026-03-27 02:58
📄 __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. """Transport classes and utilities shared among Python Elastic client libraries""" import logging from ._async_transport import AsyncTransport as AsyncTransport from ._exceptions import ( ApiError, ConnectionError, ConnectionTimeout, SecurityWarning, SerializationError, SniffingError, TlsError, TransportError, TransportWarning, ) from ._models import ApiResponseMeta, HttpHeaders, NodeConfig, SniffOptions from ._node import ( AiohttpHttpNode, BaseAsyncNode, BaseNode, HttpxAsyncHttpNode, RequestsHttpNode, Urllib3HttpNode, ) from ._node_pool import NodePool, NodeSelector, RandomSelector, RoundRobinSelector from ._otel import OpenTelemetrySpan from ._response import ApiResponse as ApiResponse from ._response import BinaryApiResponse as BinaryApiResponse from ._response import HeadApiResponse as HeadApiResponse from ._response import ListApiResponse as ListApiResponse from ._response import ObjectApiResponse as ObjectApiResponse from ._response import TextApiResponse as TextApiResponse from ._serializer import ( JsonSerializer, NdjsonSerializer, Serializer, SerializerCollection, TextSerializer, ) from ._transport import Transport as Transport from ._transport import TransportApiResponse from ._utils import fixup_module_metadata from ._version import __version__ as __version__ # noqa __all__ = [ "AiohttpHttpNode", "ApiError", "ApiResponse", "ApiResponseMeta", "AsyncTransport", "BaseAsyncNode", "BaseNode", "BinaryApiResponse", "ConnectionError", "ConnectionTimeout", "HeadApiResponse", "HttpHeaders", "HttpxAsyncHttpNode", "JsonSerializer", "ListApiResponse", "NdjsonSerializer", "NodeConfig", "NodePool", "NodeSelector", "ObjectApiResponse", "OpenTelemetrySpan", "RandomSelector", "RequestsHttpNode", "RoundRobinSelector", "SecurityWarning", "SerializationError", "Serializer", "SerializerCollection", "SniffOptions", "SniffingError", "TextApiResponse", "TextSerializer", "TlsError", "Transport", "TransportApiResponse", "TransportError", "TransportWarning", "Urllib3HttpNode", ] try: from elastic_transport._serializer import OrjsonSerializer # noqa: F401 __all__.append("OrjsonSerializer") except ImportError: pass _logger = logging.getLogger("elastic_transport") _logger.addHandler(logging.NullHandler()) del _logger fixup_module_metadata(__name__, globals()) del fixup_module_metadata def debug_logging() -> None: """Enables logging on all ``elastic_transport.*`` loggers and attaches a :class:`logging.StreamHandler` instance to each. This is an easy way to visualize the network activity occurring on the client or debug a client issue. """ handler = logging.StreamHandler() formatter = logging.Formatter( "[%(asctime)s] %(message)s", datefmt="%Y-%m-%dT%H:%M:%S" ) handler.setFormatter(formatter) for logger in ( logging.getLogger("elastic_transport.node"), logging.getLogger("elastic_transport.node_pool"), logging.getLogger("elastic_transport.transport"), ): logger.addHandler(handler) logger.setLevel(logging.DEBUG)	3.98 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 _async_transport.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 asyncio import logging from typing import ( Any, Awaitable, Callable, Collection, List, Mapping, Optional, Tuple, Type, Union, ) from ._compat import await_if_coro from ._exceptions import ( ConnectionError, ConnectionTimeout, SniffingError, TransportError, ) from ._models import DEFAULT, DefaultType, HttpHeaders, NodeConfig, SniffOptions from ._node import AiohttpHttpNode, BaseAsyncNode from ._node_pool import NodePool, NodeSelector from ._otel import OpenTelemetrySpan from ._serializer import Serializer from ._transport import ( DEFAULT_CLIENT_META_SERVICE, NOT_DEAD_NODE_HTTP_STATUSES, Transport, TransportApiResponse, validate_sniffing_options, ) from .client_utils import resolve_default _logger = logging.getLogger("elastic_transport.transport") class AsyncTransport(Transport): """ Encapsulation of transport-related to logic. Handles instantiation of the individual nodes as well as creating a node pool to hold them. Main interface is the :meth:`elastic_transport.Transport.perform_request` method. """ def __init__( self, node_configs: List[NodeConfig], node_class: Union[str, Type[BaseAsyncNode]] = AiohttpHttpNode, node_pool_class: Type[NodePool] = NodePool, randomize_nodes_in_pool: bool = True, node_selector_class: Optional[Union[str, Type[NodeSelector]]] = None, dead_node_backoff_factor: Optional[float] = None, max_dead_node_backoff: Optional[float] = None, serializers: Optional[Mapping[str, Serializer]] = None, default_mimetype: str = "application/json", max_retries: int = 3, retry_on_status: Collection[int] = (429, 502, 503, 504), retry_on_timeout: bool = False, sniff_on_start: bool = False, sniff_before_requests: bool = False, sniff_on_node_failure: bool = False, sniff_timeout: Optional[float] = 0.5, min_delay_between_sniffing: float = 10.0, sniff_callback: Optional[ Callable[ ["AsyncTransport", "SniffOptions"], Union[List[NodeConfig], Awaitable[List[NodeConfig]]], ] ] = None, meta_header: bool = True, client_meta_service: Tuple[str, str] = DEFAULT_CLIENT_META_SERVICE, ): """ :arg node_configs: List of 'NodeConfig' instances to create initial set of nodes. :arg node_class: subclass of :class:`~elastic_transport.BaseNode` to use or the name of the Connection (ie 'urllib3', 'requests') :arg node_pool_class: subclass of :class:`~elastic_transport.NodePool` to use :arg randomize_nodes_in_pool: Set to false to not randomize nodes within the pool. Defaults to true. :arg node_selector_class: Class to be used to select nodes within the :class:`~elastic_transport.NodePool`. :arg dead_node_backoff_factor: Exponential backoff factor to calculate the amount of time to timeout a node after an unsuccessful API call. :arg max_dead_node_backoff: Maximum amount of time to timeout a node after an unsuccessful API call. :arg serializers: optional dict of serializer instances that will be used for deserializing data coming from the server. (key is the mimetype) :arg max_retries: Maximum number of retries for an API call. Set to 0 to disable retries. Defaults to ``0``. :arg retry_on_status: set of HTTP status codes on which we should retry on a different node. defaults to ``(429, 502, 503, 504)`` :arg retry_on_timeout: should timeout trigger a retry on different node? (default ``False``) :arg sniff_on_start: If ``True`` will sniff for additional nodes as soon as possible, guaranteed before the first request. :arg sniff_on_node_failure: If ``True`` will sniff for additional nodees after a node is marked as dead in the pool. :arg sniff_before_requests: If ``True`` will occasionally sniff for additional nodes as requests are sent. :arg sniff_timeout: Timeout value in seconds to use for sniffing requests. Defaults to 1 second. :arg min_delay_between_sniffing: Number of seconds to wait between calls to :meth:`elastic_transport.Transport.sniff` to avoid sniffing too frequently. Defaults to 10 seconds. :arg sniff_callback: Function that is passed a :class:`elastic_transport.Transport` and :class:`elastic_transport.SniffOptions` and should do node discovery and return a list of :class:`elastic_transport.NodeConfig` instances or a coroutine that returns the list. """ # Since we don't pass all the sniffing options to super().__init__() # we want to validate the sniffing options here too. validate_sniffing_options( node_configs=node_configs, sniff_on_start=sniff_on_start, sniff_before_requests=sniff_before_requests, sniff_on_node_failure=sniff_on_node_failure, sniff_callback=sniff_callback, ) super().__init__( node_configs=node_configs, node_class=node_class, node_pool_class=node_pool_class, randomize_nodes_in_pool=randomize_nodes_in_pool, node_selector_class=node_selector_class, dead_node_backoff_factor=dead_node_backoff_factor, max_dead_node_backoff=max_dead_node_backoff, serializers=serializers, default_mimetype=default_mimetype, max_retries=max_retries, retry_on_status=retry_on_status, retry_on_timeout=retry_on_timeout, sniff_timeout=sniff_timeout, min_delay_between_sniffing=min_delay_between_sniffing, meta_header=meta_header, client_meta_service=client_meta_service, ) self._sniff_on_start = sniff_on_start self._sniff_before_requests = sniff_before_requests self._sniff_on_node_failure = sniff_on_node_failure self._sniff_timeout = sniff_timeout self._sniff_callback = sniff_callback # type: ignore self._sniffing_task: Optional["asyncio.Task[Any]"] = None self._last_sniffed_at = 0.0 # We set this to 'None' here but it'll never be None by the # time it's needed. Gets set within '_async_call()' which should # precede all logic within async calls. self._loop: asyncio.AbstractEventLoop = None # type: ignore[assignment] # AsyncTransport doesn't require a thread lock for # sniffing. Uses '_sniffing_task' instead. self._sniffing_lock = None # type: ignore[assignment] async def perform_request( # type: ignore[override, return] self, method: str, target: str, *, body: Optional[Any] = None, headers: Union[Mapping[str, Any], DefaultType] = DEFAULT, max_retries: Union[int, DefaultType] = DEFAULT, retry_on_status: Union[Collection[int], DefaultType] = DEFAULT, retry_on_timeout: Union[bool, DefaultType] = DEFAULT, request_timeout: Union[Optional[float], DefaultType] = DEFAULT, client_meta: Union[Tuple[Tuple[str, str], ...], DefaultType] = DEFAULT, otel_span: Union[OpenTelemetrySpan, DefaultType] = DEFAULT, ) -> TransportApiResponse: """ Perform the actual request. Retrieve a node from the node pool, pass all the information to it's perform_request method and return the data. If an exception was raised, mark the node as failed and retry (up to ``max_retries`` times). If the operation was successful and the node used was previously marked as dead, mark it as live, resetting it's failure count. :arg method: HTTP method to use :arg target: HTTP request target :arg body: body of the request, will be serialized using serializer and passed to the node :arg headers: Additional headers to send with the request. :arg max_retries: Maximum number of retries before giving up on a request. Set to ``0`` to disable retries. :arg retry_on_status: Collection of HTTP status codes to retry. :arg retry_on_timeout: Set to true to retry after timeout errors. :arg request_timeout: Amount of time to wait for a response to fail with a timeout error. :arg client_meta: Extra client metadata key-value pairs to send in the client meta header. :arg otel_span: OpenTelemetry span used to add metadata to the span. :returns: Tuple of the :class:`elastic_transport.ApiResponseMeta` with the deserialized response. """ await self._async_call() if headers is DEFAULT: request_headers = HttpHeaders() else: request_headers = HttpHeaders(headers) max_retries = resolve_default(max_retries, self.max_retries) retry_on_timeout = resolve_default(retry_on_timeout, self.retry_on_timeout) retry_on_status = resolve_default(retry_on_status, self.retry_on_status) otel_span = resolve_default(otel_span, OpenTelemetrySpan(None)) if self.meta_header: request_headers["x-elastic-client-meta"] = ",".join( f"{k}={v}" for k, v in self._transport_client_meta + resolve_default(client_meta, ()) ) # Serialize the request body to bytes based on the given mimetype. request_body: Optional[bytes] if body is not None: if "content-type" not in request_headers: raise ValueError( "Must provide a 'Content-Type' header to requests with bodies" ) request_body = self.serializers.dumps( body, mimetype=request_headers["content-type"] ) otel_span.set_db_statement(request_body) else: request_body = None # Errors are stored from (oldest->newest) errors: List[Exception] = [] for attempt in range(max_retries + 1): # If we sniff before requests are made we want to do so before # 'node_pool.get()' is called so our sniffed nodes show up in the pool. if self._sniff_before_requests: await self.sniff(False) retry = False node_failure = False last_response: Optional[TransportApiResponse] = None node: BaseAsyncNode = self.node_pool.get() # type: ignore[assignment] start_time = self._loop.time() try: otel_span.set_node_metadata(node.host, node.port, node.base_url, target) resp = await node.perform_request( method, target, body=request_body, headers=request_headers, request_timeout=request_timeout, ) _logger.info( "%s %s%s [status:%s duration:%.3fs]" % ( method, node.base_url, target, resp.meta.status, self._loop.time() - start_time, ) ) if method != "HEAD": body = self.serializers.loads(resp.body, resp.meta.mimetype) else: body = None if resp.meta.status in retry_on_status: retry = True # Keep track of the last response we see so we can return # it in case the retried request returns with a transport error. last_response = TransportApiResponse(resp.meta, body) except TransportError as e: _logger.info( "%s %s%s [status:%s duration:%.3fs]" % ( method, node.base_url, target, "N/A", self._loop.time() - start_time, ) ) if isinstance(e, ConnectionTimeout): retry = retry_on_timeout node_failure = True elif isinstance(e, ConnectionError): retry = True node_failure = True # If the error was determined to be a node failure # we mark it dead in the node pool to allow for # other nodes to be retried. if node_failure: self.node_pool.mark_dead(node) if self._sniff_on_node_failure: try: await self.sniff(False) except TransportError: # If sniffing on failure, it could fail too. Catch the # exception not to interrupt the retries. pass if not retry or attempt >= max_retries: # Since we're exhausted but we have previously # received some sort of response from the API # we should forward that along instead of the # transport error. Likely to be more actionable. if last_response is not None: return last_response e.errors = tuple(errors) raise else: _logger.warning( "Retrying request after failure (attempt %d of %d)", attempt, max_retries, exc_info=e, ) errors.append(e) else: # If we got back a response we need to check if that status # is indicative of a healthy node even if it's a non-2XX status if ( 200 <= resp.meta.status < 299 or resp.meta.status in NOT_DEAD_NODE_HTTP_STATUSES ): self.node_pool.mark_live(node) else: self.node_pool.mark_dead(node) if self._sniff_on_node_failure: try: await self.sniff(False) except TransportError: # If sniffing on failure, it could fail too. Catch the # exception not to interrupt the retries. pass # We either got a response we're happy with or # we've exhausted all of our retries so we return it. if not retry or attempt >= max_retries: return TransportApiResponse(resp.meta, body) else: _logger.warning( "Retrying request after non-successful status %d (attempt %d of %d)", resp.meta.status, attempt, max_retries, ) async def sniff(self, is_initial_sniff: bool = False) -> None: # type: ignore[override] await self._async_call() task = self._create_sniffing_task(is_initial_sniff) # Only block on the task if this is the initial sniff. # Otherwise we do the sniffing in the background. if is_initial_sniff and task: await task async def close(self) -> None: # type: ignore[override] """ Explicitly closes all nodes in the transport's pool """ node: BaseAsyncNode for node in self.node_pool.all(): # type: ignore[assignment] await node.close() def _should_sniff(self, is_initial_sniff: bool) -> bool: """Decide if we should sniff or not. _async_init() must be called before using this function.The async implementation doesn't have a lock. """ if is_initial_sniff: return True # Only start a new sniff if the previous run is completed. if self._sniffing_task: if not self._sniffing_task.done(): return False # If there was a previous run we collect the sniffing task's # result as it could have failed with an exception. self._sniffing_task.result() return ( self._loop.time() - self._last_sniffed_at >= self._min_delay_between_sniffing ) def _create_sniffing_task( self, is_initial_sniff: bool ) -> Optional["asyncio.Task[Any]"]: """Creates a sniffing task if one should be created and returns the task if created.""" task = None if self._should_sniff(is_initial_sniff): _logger.info("Started sniffing for additional nodes") # 'self._sniffing_task' is unset within the task implementation. task = self._loop.create_task(self._sniffing_task_impl(is_initial_sniff)) self._sniffing_task = task return task async def _sniffing_task_impl(self, is_initial_sniff: bool) -> None: """Implementation of the sniffing task""" previously_sniffed_at = self._last_sniffed_at try: self._last_sniffed_at = self._loop.time() options = SniffOptions( is_initial_sniff=is_initial_sniff, sniff_timeout=self._sniff_timeout ) assert self._sniff_callback is not None node_configs = await await_if_coro(self._sniff_callback(self, options)) if not node_configs and is_initial_sniff: raise SniffingError( "No viable nodes were discovered on the initial sniff attempt" ) prev_node_pool_size = len(self.node_pool) for node_config in node_configs: self.node_pool.add(node_config) # Do some math to log which nodes are new/existing sniffed_nodes = len(node_configs) new_nodes = sniffed_nodes - (len(self.node_pool) - prev_node_pool_size) existing_nodes = sniffed_nodes - new_nodes _logger.debug( "Discovered %d nodes during sniffing (%d new nodes, %d already in pool)", sniffed_nodes, new_nodes, existing_nodes, ) # If sniffing failed for any reason we # want to allow retrying immediately. except BaseException: self._last_sniffed_at = previously_sniffed_at raise async def _async_call(self) -> None: """Async constructor which is called on the first call to perform_request() because we're not guaranteed to be within an active asyncio event loop when __init__() is called. """ if self._loop is not None: return # Call at most once! self._loop = asyncio.get_running_loop() if self._sniff_on_start: await self.sniff(True)	19.74 KB	2025-05-25 12:26
📄 _compat.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 inspect import sys from pathlib import Path from typing import Any, Awaitable, TypeVar, Union from urllib.parse import quote as _quote from urllib.parse import urlencode, urlparse string_types = (str, bytes) T = TypeVar("T") async def await_if_coro(coro: Union[T, Awaitable[T]]) -> T: if inspect.iscoroutine(coro): return await coro # type: ignore return coro # type: ignore _QUOTE_ALWAYS_SAFE = frozenset( "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_.-~" ) def quote(string: str, safe: str = "/") -> str: # Redefines 'urllib.parse.quote()' to always have the '~' character # within the 'ALWAYS_SAFE' list. The character was added in Python 3.7 safe = "".join(_QUOTE_ALWAYS_SAFE.union(set(safe))) return _quote(string, safe) try: from threading import Lock except ImportError: class Lock: # type: ignore def __enter__(self) -> None: pass def __exit__(self, *_: Any) -> None: pass def acquire(self, _: bool = True) -> bool: return True def release(self) -> None: pass def warn_stacklevel() -> int: """Dynamically determine warning stacklevel for warnings based on the call stack""" try: # Grab the root module from the current module '__name__' module_name = __name__.partition(".")[0] module_path = Path(sys.modules[module_name].__file__) # type: ignore[arg-type] # If the module is a folder we're looking at # subdirectories, otherwise we're looking for # an exact match. module_is_folder = module_path.name == "__init__.py" if module_is_folder: module_path = module_path.parent # Look through frames until we find a file that # isn't a part of our module, then return that stacklevel. for level, frame in enumerate(inspect.stack()): # Garbage collecting frames frame_filename = Path(frame.filename) del frame if ( # If the module is a folder we look at subdirectory module_is_folder and module_path not in frame_filename.parents ) or ( # Otherwise we're looking for an exact match. not module_is_folder and module_path != frame_filename ): return level except KeyError: pass return 0 __all__ = [ "await_if_coro", "quote", "urlparse", "urlencode", "string_types", "Lock", ]	3.29 KB	2025-05-25 12:26
📄 _exceptions.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 Any, Tuple from ._models import ApiResponseMeta class TransportWarning(Warning): """Generic warning for the 'elastic-transport' package.""" class SecurityWarning(TransportWarning): """Warning for potentially insecure configurations.""" class TransportError(Exception): """Generic exception for the 'elastic-transport' package. For the 'errors' attribute, errors are ordered from most recently raised (index=0) to least recently raised (index=N) If an HTTP status code is available with the error it will be stored under 'status'. If HTTP headers are available they are stored under 'headers'. """ def __init__(self, message: Any, errors: Tuple[Exception, ...] = ()): super().__init__(message) self.errors = tuple(errors) self.message = message def __repr__(self) -> str: parts = [repr(self.message)] if self.errors: parts.append(f"errors={self.errors!r}") return "{}({})".format(self.__class__.__name__, ", ".join(parts)) def __str__(self) -> str: return str(self.message) class SniffingError(TransportError): """Error that occurs during the sniffing of nodes""" class SerializationError(TransportError): """Error that occurred during the serialization or deserialization of an HTTP message body """ class ConnectionError(TransportError): """Error raised by the HTTP connection""" def __str__(self) -> str: if self.errors: return f"Connection error caused by: {self.errors[0].__class__.__name__}({self.errors[0]})" return "Connection error" class TlsError(ConnectionError): """Error raised by during the TLS handshake""" def __str__(self) -> str: if self.errors: return f"TLS error caused by: {self.errors[0].__class__.__name__}({self.errors[0]})" return "TLS error" class ConnectionTimeout(TransportError): """Connection timed out during an operation""" def __str__(self) -> str: if self.errors: return f"Connection timeout caused by: {self.errors[0].__class__.__name__}({self.errors[0]})" return "Connection timed out" class ApiError(Exception): """Base-class for clients that raise errors due to a response such as '404 Not Found'""" def __init__( self, message: str, meta: ApiResponseMeta, body: Any, errors: Tuple[Exception, ...] = (), ): super().__init__(message) self.message = message self.errors = errors self.meta = meta self.body = body def __repr__(self) -> str: parts = [repr(self.message)] if self.meta: parts.append(f"meta={self.meta!r}") if self.errors: parts.append(f"errors={self.errors!r}") if self.body is not None: parts.append(f"body={self.body!r}") return "{}({})".format(self.__class__.__name__, ", ".join(parts)) def __str__(self) -> str: return f"[{self.meta.status}] {self.message}"	3.78 KB	2025-05-25 12:26
📄 _models.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 dataclasses import enum import re import ssl from dataclasses import dataclass, field from typing import ( TYPE_CHECKING, Any, Collection, Dict, Iterator, KeysView, Mapping, MutableMapping, Optional, Tuple, TypeVar, Union, ValuesView, ) if TYPE_CHECKING: from typing import Final class DefaultType(enum.Enum): """ Sentinel used as a default value when ``None`` has special meaning like timeouts. The only comparisons that are supported for this type are ``is``. """ value = 0 def __repr__(self) -> str: return "<DEFAULT>" def __str__(self) -> str: return "<DEFAULT>" DEFAULT: "Final[DefaultType]" = DefaultType.value T = TypeVar("T") _TYPE_SSL_VERSION = Union[int, ssl.TLSVersion] class HttpHeaders(MutableMapping[str, str]): """HTTP headers Behaves like a Python dictionary. Can be used like this:: headers = HttpHeaders() headers["foo"] = "bar" headers["foo"] = "baz" print(headers["foo"]) # prints "baz" """ __slots__ = ("_internal", "_frozen") def __init__( self, initial: Optional[Union[Mapping[str, str], Collection[Tuple[str, str]]]] = None, ) -> None: self._internal = {} self._frozen = False if initial: for key, val in dict(initial).items(): self._internal[self._normalize_key(key)] = (key, val) def __setitem__(self, key: str, value: str) -> None: if self._frozen: raise ValueError("Can't modify headers that have been frozen") self._internal[self._normalize_key(key)] = (key, value) def __getitem__(self, item: str) -> str: return self._internal[self._normalize_key(item)][1] def __delitem__(self, key: str) -> None: if self._frozen: raise ValueError("Can't modify headers that have been frozen") del self._internal[self._normalize_key(key)] def __eq__(self, other: object) -> bool: if not isinstance(other, Mapping): return NotImplemented if not isinstance(other, HttpHeaders): other = HttpHeaders(other) return {k: v for k, (_, v) in self._internal.items()} == { k: v for k, (_, v) in other._internal.items() } def __ne__(self, other: object) -> bool: if not isinstance(other, Mapping): return NotImplemented return not self == other def __iter__(self) -> Iterator[str]: return iter(self.keys()) def __len__(self) -> int: return len(self._internal) def __bool__(self) -> bool: return bool(self._internal) def __contains__(self, item: object) -> bool: return isinstance(item, str) and self._normalize_key(item) in self._internal def __repr__(self) -> str: return repr(self._dict_hide_auth()) def __str__(self) -> str: return str(self._dict_hide_auth()) def __hash__(self) -> int: if not self._frozen: raise ValueError("Can't calculate the hash of headers that aren't frozen") return hash(tuple((k, v) for k, (_, v) in sorted(self._internal.items()))) def get(self, key: str, default: Optional[str] = None) -> Optional[str]: # type: ignore[override] return self._internal.get(self._normalize_key(key), (None, default))[1] def keys(self) -> KeysView[str]: return self._internal.keys() def values(self) -> ValuesView[str]: return {"": v for _, v in self._internal.values()}.values() def items(self) -> Collection[Tuple[str, str]]: # type: ignore[override] return [(key, val) for _, (key, val) in self._internal.items()] def freeze(self) -> "HttpHeaders": """Freezes the current set of headers so they can be used in hashes. Returns the same instance, doesn't make a copy. """ self._frozen = True return self @property def frozen(self) -> bool: return self._frozen def copy(self) -> "HttpHeaders": return HttpHeaders(self.items()) def _normalize_key(self, key: str) -> str: try: return key.lower() except AttributeError: return key def _dict_hide_auth(self) -> Dict[str, str]: def hide_auth(val: str) -> str: # Hides only the authentication value, not the method. match = re.match(r"^(ApiKey\|Basic\|Bearer) ", val) if match: return f"{match.group(1)} <hidden>" return "<hidden>" return { key: hide_auth(val) if key.lower() == "authorization" else val for key, val in self.items() } @dataclass class ApiResponseMeta: """Metadata that is returned from Transport.perform_request() :ivar int status: HTTP status code :ivar str http_version: HTTP version being used :ivar HttpHeaders headers: HTTP headers :ivar float duration: Number of seconds from start of request to start of response :ivar NodeConfig node: Node which handled the request :ivar typing.Optional[str] mimetype: Mimetype to be used by the serializer to decode the raw response bytes. """ status: int http_version: str headers: HttpHeaders duration: float node: "NodeConfig" @property def mimetype(self) -> Optional[str]: try: content_type = self.headers["content-type"] return content_type.partition(";")[0] or None except KeyError: return None def _empty_frozen_http_headers() -> HttpHeaders: """Used for the 'default_factory' of the 'NodeConfig.headers'""" return HttpHeaders().freeze() @dataclass(repr=True) class NodeConfig: """Configuration options available for every node.""" #: Protocol in use to connect to the node scheme: str #: IP address or hostname to connect to host: str #: IP port to connect to port: int #: Prefix to add to the path of every request path_prefix: str = "" #: Default HTTP headers to add to every request headers: Union[HttpHeaders, Mapping[str, str]] = field( default_factory=_empty_frozen_http_headers ) #: Number of concurrent connections that are #: able to be open at one time for this node. #: Having multiple connections per node allows #: for higher concurrency of requests. connections_per_node: int = 10 #: Number of seconds to wait before a request should timeout. request_timeout: Optional[float] = 10.0 #: Set to ``True`` to enable HTTP compression #: of request and response bodies via gzip. http_compress: Optional[bool] = False #: Set to ``True`` to verify the node's TLS certificate against 'ca_certs' #: Setting to ``False`` will disable verifying the node's certificate. verify_certs: Optional[bool] = True #: Path to a CA bundle or directory containing bundles. By default #: If the ``certifi`` package is installed and ``verify_certs`` is #: set to ``True`` this value will be set to ``certifi.where()``. ca_certs: Optional[str] = None #: Path to a client certificate for TLS client authentication. client_cert: Optional[str] = None #: Path to a client private key for TLS client authentication. client_key: Optional[str] = None #: Hostname or IP address to verify on the node's certificate. #: This is useful if the certificate contains a different value #: than the one supplied in ``host``. An example of this situation #: is connecting to an IP address instead of a hostname. #: Set to ``False`` to disable certificate hostname verification. ssl_assert_hostname: Optional[str] = None #: SHA-256 fingerprint of the node's certificate. If this value is #: given then root-of-trust verification isn't done and only the #: node's certificate fingerprint is verified. #: #: On CPython 3.10+ this also verifies if any certificate in the #: chain including the Root CA matches this fingerprint. However #: because this requires using private APIs support for this is #: experimental. ssl_assert_fingerprint: Optional[str] = None #: Minimum TLS version to use to connect to the node. Can be either #: :class:`ssl.TLSVersion` or one of the deprecated #: ``ssl.PROTOCOL_TLSvX`` instances. ssl_version: Optional[_TYPE_SSL_VERSION] = None #: Pre-configured :class:`ssl.SSLContext` object. If this value #: is given then no other TLS options (besides ``ssl_assert_fingerprint``) #: can be set on the :class:`elastic_transport.NodeConfig`. ssl_context: Optional[ssl.SSLContext] = field(default=None, hash=False) #: Set to ``False`` to disable the :class:`elastic_transport.SecurityWarning` #: issued when using ``verify_certs=False``. ssl_show_warn: bool = True #: Extras that can be set to anything, typically used #: for annotating this node with additional information for #: future decisions like sniffing, instance roles, etc. #: Third-party keys should start with an underscore and prefix. _extras: Dict[str, Any] = field(default_factory=dict, hash=False) def replace(self, kwargs: Any) -> "NodeConfig": if not kwargs: return self return dataclasses.replace(self, kwargs) def __post_init__(self) -> None: if not isinstance(self.headers, HttpHeaders) or not self.headers.frozen: self.headers = HttpHeaders(self.headers).freeze() if self.scheme != self.scheme.lower(): raise ValueError("'scheme' must be lowercase") if "[" in self.host or "]" in self.host: raise ValueError("'host' must not have square braces") if self.port < 0: raise ValueError("'port' must be a positive integer") if self.connections_per_node <= 0: raise ValueError("'connections_per_node' must be a positive integer") if self.path_prefix: self.path_prefix = ( ("/" + self.path_prefix.strip("/")) if self.path_prefix else "" ) tls_options = [ "ca_certs", "client_cert", "client_key", "ssl_assert_hostname", "ssl_assert_fingerprint", "ssl_context", ] # Disallow setting TLS options on non-HTTPS connections. if self.scheme != "https": if any(getattr(self, attr) is not None for attr in tls_options): raise ValueError("TLS options require scheme to be 'https'") elif self.scheme == "https": # It's not valid to set 'ssl_context' and any other # TLS option, the SSLContext object must be configured # the way the user wants already. def tls_option_filter(attr: object) -> bool: return ( isinstance(attr, str) and attr not in ("ssl_context", "ssl_assert_fingerprint") and getattr(self, attr) is not None ) if self.ssl_context is not None and any( filter( tls_option_filter, tls_options, ) ): raise ValueError( "The 'ssl_context' option can't be combined with other TLS options" ) def __eq__(self, other: object) -> bool: if not isinstance(other, NodeConfig): return NotImplemented return ( self.scheme == other.scheme and self.host == other.host and self.port == other.port and self.path_prefix == other.path_prefix ) def __ne__(self, other: object) -> bool: if not isinstance(other, NodeConfig): return NotImplemented return not self == other def __hash__(self) -> int: return hash( ( self.scheme, self.host, self.port, self.path_prefix, ) ) @dataclass() class SniffOptions: """Options which are passed to Transport.sniff_callback""" is_initial_sniff: bool sniff_timeout: Optional[float]	12.76 KB	2025-05-25 12:26
📁 _node	-	2025-05-25 12:26
📄 _node_pool.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 logging import random import threading import time from collections import defaultdict from queue import Empty, PriorityQueue from typing import ( TYPE_CHECKING, Dict, List, Optional, Sequence, Set, Tuple, Type, Union, overload, ) from ._compat import Lock from ._models import NodeConfig from ._node import BaseNode if TYPE_CHECKING: from typing import Literal _logger = logging.getLogger("elastic_transport.node_pool") class NodeSelector: """ Simple class used to select a node from a list of currently live node instances. In init time it is passed a dictionary containing all the nodes options which it can then use during the selection process. When the ``select()`` method is called it is given a list of currently live nodes to choose from. The selector is initialized with the list of seed nodes that the NodePool was initialized with. This list of seed nodes can be used to make decisions within ``select()`` Example of where this would be useful is a zone-aware selector that would only select connections from it's own zones and only fall back to other connections where there would be none in its zones. """ def __init__(self, node_configs: List[NodeConfig]): """ :arg node_configs: List of NodeConfig instances """ self.node_configs = node_configs def select(self, nodes: Sequence[BaseNode]) -> BaseNode: # pragma: nocover """ Select a nodes from the given list. :arg nodes: list of live nodes to choose from """ raise NotImplementedError() class RandomSelector(NodeSelector): """Randomly select a node""" def select(self, nodes: Sequence[BaseNode]) -> BaseNode: return random.choice(nodes) class RoundRobinSelector(NodeSelector): """Select a node using round-robin""" def __init__(self, node_configs: List[NodeConfig]): super().__init__(node_configs) self._thread_local = threading.local() def select(self, nodes: Sequence[BaseNode]) -> BaseNode: self._thread_local.rr = (getattr(self._thread_local, "rr", -1) + 1) % len(nodes) return nodes[self._thread_local.rr] _SELECTOR_CLASS_NAMES: Dict[str, Type[NodeSelector]] = { "round_robin": RoundRobinSelector, "random": RandomSelector, } class NodePool: """ Container holding the :class:`~elastic_transport.BaseNode` instances, managing the selection process (via a :class:`~elastic_transport.NodeSelector`) and dead connections. It's only interactions are with the :class:`~elastic_transport.Transport` class that drives all the actions within ``NodePool``. Initially nodes are stored on the class as a list and, along with the connection options, get passed to the ``NodeSelector`` instance for future reference. Upon each request the ``Transport`` will ask for a ``BaseNode`` via the ``get_node`` method. If the connection fails (it's `perform_request` raises a `ConnectionError`) it will be marked as dead (via `mark_dead`) and put on a timeout (if it fails N times in a row the timeout is exponentially longer - the formula is `default_timeout * 2 ** (fail_count - 1)`). When the timeout is over the connection will be resurrected and returned to the live pool. A connection that has been previously marked as dead and succeeds will be marked as live (its fail count will be deleted). """ def __init__( self, node_configs: List[NodeConfig], node_class: Type[BaseNode], dead_node_backoff_factor: float = 1.0, max_dead_node_backoff: float = 30.0, node_selector_class: Union[str, Type[NodeSelector]] = RoundRobinSelector, randomize_nodes: bool = True, ): """ :arg node_configs: List of initial NodeConfigs to use :arg node_class: Type to use when creating nodes :arg dead_node_backoff_factor: Number of seconds used as a factor in calculating the amount of "backoff" time we should give a node after an unsuccessful request. The formula is calculated as follows where N is the number of consecutive failures: ``min(dead_backoff_factor * (2 ** (N - 1)), max_dead_backoff)`` :arg max_dead_node_backoff: Maximum number of seconds to wait when calculating the "backoff" time for a dead node. :arg node_selector_class: :class:`~elastic_transport.NodeSelector` subclass to use if more than one connection is live :arg randomize_nodes: shuffle the list of nodes upon instantiation to avoid dog-piling effect across processes """ if not node_configs: raise ValueError("Must specify at least one NodeConfig") node_configs = list( node_configs ) # Make a copy so we don't have side-effects outside. if any(not isinstance(node_config, NodeConfig) for node_config in node_configs): raise TypeError("NodePool must be passed a list of NodeConfig instances") if isinstance(node_selector_class, str): if node_selector_class not in _SELECTOR_CLASS_NAMES: raise ValueError( "Unknown option for selector_class: '%s'. " "Available options are: '%s'" % ( node_selector_class, "', '".join(sorted(_SELECTOR_CLASS_NAMES.keys())), ) ) node_selector_class = _SELECTOR_CLASS_NAMES[node_selector_class] if randomize_nodes: # randomize the list of nodes to avoid hammering the same node # if a large set of clients are created all at once. random.shuffle(node_configs) # Initial set of nodes that the NodePool was initialized with. # This set of nodes can never be removed. self._seed_nodes: Tuple[NodeConfig, ...] = tuple(set(node_configs)) if len(self._seed_nodes) != len(node_configs): raise ValueError("Cannot use duplicate NodeConfigs within a NodePool") self._node_class = node_class self._node_selector = node_selector_class(node_configs) # _all_nodes relies on dict insert order self._all_nodes: Dict[NodeConfig, BaseNode] = {} for node_config in node_configs: self._all_nodes[node_config] = self._node_class(node_config) # Lock that is used to protect writing to 'all_nodes' self._all_nodes_write_lock = Lock() # Flag which tells NodePool.get() that there's only one node # which allows for optimizations. Setting this flag is also # protected by the above write lock. self._all_nodes_len_1 = len(self._all_nodes) == 1 # Collection of currently-alive nodes. This is an ordered # dict so round-robin actually works. self._alive_nodes: Dict[NodeConfig, BaseNode] = dict(self._all_nodes) # PriorityQueue for thread safety and ease of timeout management self._dead_nodes: PriorityQueue[Tuple[float, BaseNode]] = PriorityQueue() self._dead_consecutive_failures: Dict[NodeConfig, int] = defaultdict(int) # Nodes that have been marked as 'removed' to be thread-safe. self._removed_nodes: Set[NodeConfig] = set() # default timeout after which to try resurrecting a connection self._dead_node_backoff_factor = dead_node_backoff_factor self._max_dead_node_backoff = max_dead_node_backoff @property def node_class(self) -> Type[BaseNode]: return self._node_class @property def node_selector(self) -> NodeSelector: return self._node_selector @property def dead_node_backoff_factor(self) -> float: return self._dead_node_backoff_factor @property def max_dead_node_backoff(self) -> float: return self._max_dead_node_backoff def mark_dead(self, node: BaseNode, _now: Optional[float] = None) -> None: """ Mark the node as dead (failed). Remove it from the live pool and put it on a timeout. :arg node: The failed node. """ now: float = _now if _now is not None else time.time() try: del self._alive_nodes[node.config] except KeyError: pass consecutive_failures = self._dead_consecutive_failures[node.config] + 1 self._dead_consecutive_failures[node.config] = consecutive_failures try: timeout = min( self._dead_node_backoff_factor * (2 ** (consecutive_failures - 1)), self._max_dead_node_backoff, ) except OverflowError: timeout = self._max_dead_node_backoff self._dead_nodes.put((now + timeout, node)) _logger.warning( "Node %r has failed for %i times in a row, putting on %i second timeout", node, consecutive_failures, timeout, ) def mark_live(self, node: BaseNode) -> None: """ Mark node as healthy after a resurrection. Resets the fail counter for the node. :arg node: The ``BaseNode`` instance to mark as alive. """ try: del self._dead_consecutive_failures[node.config] except KeyError: # race condition, safe to ignore pass else: self._alive_nodes.setdefault(node.config, node) _logger.warning( "Node %r has been marked alive after a successful request", node, ) @overload def resurrect(self, force: "Literal[True]" = ...) -> BaseNode: ... @overload def resurrect(self, force: "Literal[False]" = ...) -> Optional[BaseNode]: ... def resurrect(self, force: bool = False) -> Optional[BaseNode]: """ Attempt to resurrect a node from the dead queue. It will try to locate one (not all) eligible (it's timeout is over) node to return to the live pool. Any resurrected node is also returned. :arg force: resurrect a node even if there is none eligible (used when we have no live nodes). If force is 'True'' resurrect always returns a node. """ node: Optional[BaseNode] mark_node_alive_after: float = 0.0 try: # Try to resurrect a dead node if any. mark_node_alive_after, node = self._dead_nodes.get(block=False) except Empty: # No dead nodes. if force: # If we're being forced to return a node we randomly # pick between alive and dead nodes. return random.choice(list(self._all_nodes.values())) node = None if node is not None and not force and mark_node_alive_after > time.time(): # return it back if not eligible and not forced self._dead_nodes.put((mark_node_alive_after, node)) node = None # either we were forced or the node is eligible to be retried if node is not None: self._alive_nodes[node.config] = node _logger.info("Resurrected node %r (force=%s)", node, force) return node def add(self, node_config: NodeConfig) -> None: try: # If the node was previously removed we mark it as "in the pool" self._removed_nodes.remove(node_config) except KeyError: pass with self._all_nodes_write_lock: # We don't error when trying to add a duplicate node # to the pool because threading+sniffing can call # .add() on the same NodeConfig. if node_config not in self._all_nodes: node = self._node_class(node_config) self._all_nodes[node.config] = node # Update the flag to disable optimizations. Also ensures that # .resurrect() starts getting called so our added node makes # it way into the alive nodes. self._all_nodes_len_1 = False # Start the node as dead because 'dead_nodes' is thread-safe. # The node will be resurrected on the next call to .get() self._dead_consecutive_failures[node.config] = 0 self._dead_nodes.put((time.time(), node)) def remove(self, node_config: NodeConfig) -> None: # Can't mark a seed node as removed. if node_config not in self._seed_nodes: self._removed_nodes.add(node_config) def get(self) -> BaseNode: """ Return a node from the pool using the ``NodeSelector`` instance. It tries to resurrect eligible nodes, forces a resurrection when no nodes are available and passes the list of live nodes to the selector instance to choose from. """ # Even with the optimization below we want to participate in the # dead/alive cycle in case more nodes join after sniffing, for example. self.resurrect() # Flag that short-circuits the extra logic if we have only one node. # The only way this flag can be set to 'True' is if there were only # one node defined within 'seed_nodes' so we know this good to do. if self._all_nodes_len_1: return self._all_nodes[self._seed_nodes[0]] # Filter nodes in 'alive_nodes' to ones not marked as removed. nodes = [ node for node_config, node in self._alive_nodes.items() if node_config not in self._removed_nodes ] # No live nodes, resurrect one by force and return it if not nodes: return self.resurrect(force=True) # Only call selector if we have a choice to make if len(nodes) > 1: return self._node_selector.select(nodes) return nodes[0] def all(self) -> List[BaseNode]: return list(self._all_nodes.values()) def __repr__(self) -> str: return "<NodePool>" def __len__(self) -> int: return len(self._all_nodes)	14.64 KB	2025-05-25 12:26
📄 _otel.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 __future__ import annotations from typing import TYPE_CHECKING, Literal, Mapping if TYPE_CHECKING: from opentelemetry.trace import Span # A list of the Elasticsearch endpoints that qualify as "search" endpoints. The search query in # the request body may be captured for these endpoints, depending on the body capture strategy. SEARCH_ENDPOINTS = ( "search", "async_search.submit", "msearch", "eql.search", "esql.query", "terms_enum", "search_template", "msearch_template", "render_search_template", ) class OpenTelemetrySpan: def __init__( self, otel_span: Span \| None, endpoint_id: str \| None = None, body_strategy: Literal["omit", "raw"] = "omit", ): self.otel_span = otel_span self.body_strategy = body_strategy self.endpoint_id = endpoint_id def set_node_metadata( self, host: str, port: int, base_url: str, target: str ) -> None: if self.otel_span is None: return # url.full does not contain auth info which is passed as headers self.otel_span.set_attribute("url.full", base_url + target) self.otel_span.set_attribute("server.address", host) self.otel_span.set_attribute("server.port", port) def set_elastic_cloud_metadata(self, headers: Mapping[str, str]) -> None: if self.otel_span is None: return cluster_name = headers.get("X-Found-Handling-Cluster") if cluster_name is not None: self.otel_span.set_attribute("db.elasticsearch.cluster.name", cluster_name) node_name = headers.get("X-Found-Handling-Instance") if node_name is not None: self.otel_span.set_attribute("db.elasticsearch.node.name", node_name) def set_db_statement(self, serialized_body: bytes) -> None: if self.otel_span is None: return if self.body_strategy == "omit": return elif self.body_strategy == "raw" and self.endpoint_id in SEARCH_ENDPOINTS: self.otel_span.set_attribute( "db.statement", serialized_body.decode("utf-8") )	2.88 KB	2025-05-25 12:26
📄 _response.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 ( Any, Dict, Generic, Iterator, List, NoReturn, Tuple, TypeVar, Union, overload, ) from ._models import ApiResponseMeta _BodyType = TypeVar("_BodyType") _ObjectBodyType = TypeVar("_ObjectBodyType") _ListItemBodyType = TypeVar("_ListItemBodyType") class ApiResponse(Generic[_BodyType]): """Base class for all API response classes""" __slots__ = ("_body", "_meta") def __init__( self, args: Any, *kwargs: Any, ): def _raise_typeerror() -> NoReturn: raise TypeError("Must pass 'meta' and 'body' to ApiResponse") from None # Working around pre-releases of elasticsearch-python # that would use raw=... instead of body=... try: if bool(args) == bool(kwargs): _raise_typeerror() elif args and len(args) == 2: body, meta = args elif kwargs and "raw" in kwargs: body = kwargs.pop("raw") meta = kwargs.pop("meta") kwargs.pop("body_cls", None) elif kwargs and "body" in kwargs: body = kwargs.pop("body") meta = kwargs.pop("meta") kwargs.pop("body_cls", None) else: _raise_typeerror() except KeyError: _raise_typeerror() # If there are still kwargs left over # and we're not in positional mode... if not args and kwargs: _raise_typeerror() self._body = body self._meta = meta def __repr__(self) -> str: return f"{type(self).__name__}({self.body!r})" def __contains__(self, item: Any) -> bool: return item in self._body def __eq__(self, other: object) -> bool: if isinstance(other, ApiResponse): other = other.body return self._body == other # type: ignore[no-any-return] def __ne__(self, other: object) -> bool: if isinstance(other, ApiResponse): other = other.body return self._body != other # type: ignore[no-any-return] def __getitem__(self, item: Any) -> Any: return self._body[item] def __getattr__(self, attr: str) -> Any: return getattr(self._body, attr) def __getstate__(self) -> Tuple[_BodyType, ApiResponseMeta]: return self._body, self._meta def __setstate__(self, state: Tuple[_BodyType, ApiResponseMeta]) -> None: self._body, self._meta = state def __len__(self) -> int: return len(self._body) def __iter__(self) -> Iterator[Any]: return iter(self._body) def __str__(self) -> str: return str(self._body) def __bool__(self) -> bool: return bool(self._body) @property def meta(self) -> ApiResponseMeta: """Response metadata""" return self._meta # type: ignore[no-any-return] @property def body(self) -> _BodyType: """User-friendly view into the raw response with type hints if applicable""" return self._body # type: ignore[no-any-return] @property def raw(self) -> _BodyType: return self.body class TextApiResponse(ApiResponse[str]): """API responses which are text such as 'text/plain' or 'text/csv'""" def __iter__(self) -> Iterator[str]: return iter(self.body) def __getitem__(self, item: Union[int, slice]) -> str: return self.body[item] @property def body(self) -> str: return self._body # type: ignore[no-any-return] class BinaryApiResponse(ApiResponse[bytes]): """API responses which are a binary response such as Mapbox vector tiles""" def __iter__(self) -> Iterator[int]: return iter(self.body) @overload def __getitem__(self, item: slice) -> bytes: ... @overload def __getitem__(self, item: int) -> int: ... def __getitem__(self, item: Union[int, slice]) -> Union[int, bytes]: return self.body[item] @property def body(self) -> bytes: return self._body # type: ignore[no-any-return] class HeadApiResponse(ApiResponse[bool]): """API responses which are for an 'exists' / HEAD API request""" def __init__(self, meta: ApiResponseMeta): super().__init__(body=200 <= meta.status < 300, meta=meta) def __bool__(self) -> bool: return 200 <= self.meta.status < 300 @property def body(self) -> bool: return bool(self) class ObjectApiResponse(Generic[_ObjectBodyType], ApiResponse[Dict[str, Any]]): """API responses which are for a JSON object""" def __getitem__(self, item: str) -> Any: return self.body[item] # type: ignore[index] def __iter__(self) -> Iterator[str]: return iter(self._body) @property def body(self) -> _ObjectBodyType: # type: ignore[override] return self._body # type: ignore[no-any-return] class ListApiResponse( Generic[_ListItemBodyType], ApiResponse[List[Any]], ): """API responses which are a list of items. Can be NDJSON or a JSON list""" @overload def __getitem__(self, item: slice) -> List[_ListItemBodyType]: ... @overload def __getitem__(self, item: int) -> _ListItemBodyType: ... def __getitem__( self, item: Union[int, slice] ) -> Union[_ListItemBodyType, List[_ListItemBodyType]]: return self.body[item] def __iter__(self) -> Iterator[_ListItemBodyType]: return iter(self.body) @property def body(self) -> List[_ListItemBodyType]: return self._body # type: ignore[no-any-return]	6.25 KB	2025-05-25 12:26
📄 _serializer.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 json import re import uuid from datetime import date from decimal import Decimal from typing import Any, ClassVar, Mapping, Optional from ._exceptions import SerializationError try: import orjson except ModuleNotFoundError: orjson = None # type: ignore[assignment] class Serializer: """Serializer interface.""" mimetype: ClassVar[str] def loads(self, data: bytes) -> Any: # pragma: nocover raise NotImplementedError() def dumps(self, data: Any) -> bytes: # pragma: nocover raise NotImplementedError() class TextSerializer(Serializer): """Text serializer to and from UTF-8.""" mimetype: ClassVar[str] = "text/" def loads(self, data: bytes) -> str: if isinstance(data, str): return data try: return data.decode("utf-8", "surrogatepass") except UnicodeError as e: raise SerializationError( f"Unable to deserialize as text: {data!r}", errors=(e,) ) def dumps(self, data: str) -> bytes: # The body is already encoded to bytes # so we forward the request body along. if isinstance(data, bytes): return data try: return data.encode("utf-8", "surrogatepass") except (AttributeError, UnicodeError, TypeError) as e: raise SerializationError( f"Unable to serialize to text: {data!r}", errors=(e,) ) class JsonSerializer(Serializer): """JSON serializer relying on the standard library json module.""" mimetype: ClassVar[str] = "application/json" def default(self, data: Any) -> Any: if isinstance(data, date): return data.isoformat() elif isinstance(data, uuid.UUID): return str(data) elif isinstance(data, Decimal): return float(data) raise SerializationError( message=f"Unable to serialize to JSON: {data!r} (type: {type(data).__name__})", ) def json_dumps(self, data: Any) -> bytes: return json.dumps( data, default=self.default, ensure_ascii=False, separators=(",", ":") ).encode("utf-8", "surrogatepass") def json_loads(self, data: bytes) -> Any: return json.loads(data) def loads(self, data: bytes) -> Any: # Sometimes responses use Content-Type: json but actually # don't contain any data. We should return something instead # of erroring in these cases. if data == b"": return None try: return self.json_loads(data) except (ValueError, TypeError) as e: raise SerializationError( message=f"Unable to deserialize as JSON: {data!r}", errors=(e,) ) def dumps(self, data: Any) -> bytes: # The body is already encoded to bytes # so we forward the request body along. if isinstance(data, str): return data.encode("utf-8", "surrogatepass") elif isinstance(data, bytes): return data try: return self.json_dumps(data) # This should be captured by the .default() # call but just in case we also wrap these. except (ValueError, UnicodeError, TypeError) as e: # pragma: nocover raise SerializationError( message=f"Unable to serialize to JSON: {data!r} (type: {type(data).__name__})", errors=(e,), ) if orjson is not None: class OrjsonSerializer(JsonSerializer): """JSON serializer relying on the orjson package. Only available if orjson if installed. It is faster, especially for vectors, but is also stricter. """ def json_dumps(self, data: Any) -> bytes: return orjson.dumps( data, default=self.default, option=orjson.OPT_SERIALIZE_NUMPY ) def json_loads(self, data: bytes) -> Any: return orjson.loads(data) class NdjsonSerializer(JsonSerializer): """Newline delimited JSON (NDJSON) serializer relying on the standard library json module.""" mimetype: ClassVar[str] = "application/x-ndjson" def loads(self, data: bytes) -> Any: ndjson = [] for line in re.split(b"[\n\r]", data): if not line: continue try: ndjson.append(self.json_loads(line)) except (ValueError, TypeError) as e: raise SerializationError( message=f"Unable to deserialize as NDJSON: {data!r}", errors=(e,) ) return ndjson def dumps(self, data: Any) -> bytes: # The body is already encoded to bytes # so we forward the request body along. if isinstance(data, (bytes, str)): data = (data,) buffer = bytearray() for line in data: if isinstance(line, str): line = line.encode("utf-8", "surrogatepass") if isinstance(line, bytes): buffer += line # Ensure that there is always a final newline if not line.endswith(b"\n"): buffer += b"\n" else: try: buffer += self.json_dumps(line) buffer += b"\n" # This should be captured by the .default() # call but just in case we also wrap these. except (ValueError, UnicodeError, TypeError) as e: # pragma: nocover raise SerializationError( message=f"Unable to serialize to NDJSON: {data!r} (type: {type(data).__name__})", errors=(e,), ) return bytes(buffer) DEFAULT_SERIALIZERS = { JsonSerializer.mimetype: JsonSerializer(), TextSerializer.mimetype: TextSerializer(), NdjsonSerializer.mimetype: NdjsonSerializer(), } class SerializerCollection: """Collection of serializers that can be fetched by mimetype. Used by :class:`elastic_transport.Transport` to serialize and deserialize native Python types into bytes before passing to a node. """ def __init__( self, serializers: Optional[Mapping[str, Serializer]] = None, default_mimetype: str = "application/json", ): if serializers is None: serializers = DEFAULT_SERIALIZERS try: self.default_serializer = serializers[default_mimetype] except KeyError: raise ValueError( f"Must configure a serializer for the default mimetype {default_mimetype!r}" ) from None self.serializers = dict(serializers) def dumps(self, data: Any, mimetype: Optional[str] = None) -> bytes: return self.get_serializer(mimetype).dumps(data) def loads(self, data: bytes, mimetype: Optional[str] = None) -> Any: return self.get_serializer(mimetype).loads(data) def get_serializer(self, mimetype: Optional[str]) -> Serializer: # split out charset if mimetype is None: serializer = self.default_serializer else: mimetype, _, _ = mimetype.partition(";") try: serializer = self.serializers[mimetype] except KeyError: # Try for '<mimetype-supertype>/' types after the specific type fails. try: mimetype_supertype = mimetype.partition("/")[0] serializer = self.serializers[f"{mimetype_supertype}/*"] except KeyError: raise SerializationError( f"Unknown mimetype, not able to serialize or deserialize: {mimetype}" ) from None return serializer	8.41 KB	2025-05-25 12:26
📄 _transport.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 dataclasses import inspect import logging import time import warnings from platform import python_version from typing import ( Any, Callable, Collection, Dict, List, Mapping, NamedTuple, Optional, Tuple, Type, Union, cast, ) from ._compat import Lock, warn_stacklevel from ._exceptions import ( ConnectionError, ConnectionTimeout, SniffingError, TransportError, TransportWarning, ) from ._models import ( DEFAULT, ApiResponseMeta, DefaultType, HttpHeaders, NodeConfig, SniffOptions, ) from ._node import ( AiohttpHttpNode, BaseNode, HttpxAsyncHttpNode, RequestsHttpNode, Urllib3HttpNode, ) from ._node_pool import NodePool, NodeSelector from ._otel import OpenTelemetrySpan from ._serializer import DEFAULT_SERIALIZERS, Serializer, SerializerCollection from ._version import __version__ from .client_utils import client_meta_version, resolve_default # Allows for using a node_class by name rather than import. NODE_CLASS_NAMES: Dict[str, Type[BaseNode]] = { "urllib3": Urllib3HttpNode, "requests": RequestsHttpNode, "aiohttp": AiohttpHttpNode, "httpxasync": HttpxAsyncHttpNode, } # These are HTTP status errors that shouldn't be considered # 'errors' for marking a node as dead. These errors typically # mean everything is fine server-wise and instead the API call # in question responded successfully. NOT_DEAD_NODE_HTTP_STATUSES = {None, 400, 401, 402, 403, 404, 409} DEFAULT_CLIENT_META_SERVICE = ("et", client_meta_version(__version__)) _logger = logging.getLogger("elastic_transport.transport") class TransportApiResponse(NamedTuple): meta: ApiResponseMeta body: Any class Transport: """ Encapsulation of transport-related to logic. Handles instantiation of the individual nodes as well as creating a node pool to hold them. Main interface is the :meth:`elastic_transport.Transport.perform_request` method. """ def __init__( self, node_configs: List[NodeConfig], node_class: Union[str, Type[BaseNode]] = Urllib3HttpNode, node_pool_class: Type[NodePool] = NodePool, randomize_nodes_in_pool: bool = True, node_selector_class: Optional[Union[str, Type[NodeSelector]]] = None, dead_node_backoff_factor: Optional[float] = None, max_dead_node_backoff: Optional[float] = None, serializers: Optional[Mapping[str, Serializer]] = None, default_mimetype: str = "application/json", max_retries: int = 3, retry_on_status: Collection[int] = (429, 502, 503, 504), retry_on_timeout: bool = False, sniff_on_start: bool = False, sniff_before_requests: bool = False, sniff_on_node_failure: bool = False, sniff_timeout: Optional[float] = 0.5, min_delay_between_sniffing: float = 10.0, sniff_callback: Optional[ Callable[ ["Transport", "SniffOptions"], Union[List[NodeConfig], List[NodeConfig]], ] ] = None, meta_header: bool = True, client_meta_service: Tuple[str, str] = DEFAULT_CLIENT_META_SERVICE, ): """ :arg node_configs: List of 'NodeConfig' instances to create initial set of nodes. :arg node_class: subclass of :class:`~elastic_transport.BaseNode` to use or the name of the Connection (ie 'urllib3', 'requests') :arg node_pool_class: subclass of :class:`~elastic_transport.NodePool` to use :arg randomize_nodes_in_pool: Set to false to not randomize nodes within the pool. Defaults to true. :arg node_selector_class: Class to be used to select nodes within the :class:`~elastic_transport.NodePool`. :arg dead_node_backoff_factor: Exponential backoff factor to calculate the amount of time to timeout a node after an unsuccessful API call. :arg max_dead_node_backoff: Maximum amount of time to timeout a node after an unsuccessful API call. :arg serializers: optional dict of serializer instances that will be used for deserializing data coming from the server. (key is the mimetype) :arg max_retries: Maximum number of retries for an API call. Set to 0 to disable retries. Defaults to ``0``. :arg retry_on_status: set of HTTP status codes on which we should retry on a different node. defaults to ``(429, 502, 503, 504)`` :arg retry_on_timeout: should timeout trigger a retry on different node? (default ``False``) :arg sniff_on_start: If ``True`` will sniff for additional nodes as soon as possible, guaranteed before the first request. :arg sniff_on_node_failure: If ``True`` will sniff for additional nodees after a node is marked as dead in the pool. :arg sniff_before_requests: If ``True`` will occasionally sniff for additional nodes as requests are sent. :arg sniff_timeout: Timeout value in seconds to use for sniffing requests. Defaults to 1 second. :arg min_delay_between_sniffing: Number of seconds to wait between calls to :meth:`elastic_transport.Transport.sniff` to avoid sniffing too frequently. Defaults to 10 seconds. :arg sniff_callback: Function that is passed a :class:`elastic_transport.Transport` and :class:`elastic_transport.SniffOptions` and should do node discovery and return a list of :class:`elastic_transport.NodeConfig` instances. :arg meta_header: If set to False the ``X-Elastic-Client-Meta`` HTTP header won't be sent. Defaults to True. :arg client_meta_service: Key-value pair for the service field of the client metadata header. Defaults to the service key-value for Elastic Transport. """ if isinstance(node_class, str): if node_class not in NODE_CLASS_NAMES: options = "', '".join(sorted(NODE_CLASS_NAMES.keys())) raise ValueError( f"Unknown option for node_class: '{node_class}'. " f"Available options are: '{options}'" ) node_class = NODE_CLASS_NAMES[node_class] # Verify that the node_class we're passed is # async/sync the same as the transport is. is_transport_async = inspect.iscoroutinefunction(self.perform_request) is_node_async = inspect.iscoroutinefunction(node_class.perform_request) if is_transport_async != is_node_async: raise ValueError( f"Specified 'node_class' {'is' if is_node_async else 'is not'} async, " f"should be {'async' if is_transport_async else 'sync'} instead" ) validate_sniffing_options( node_configs=node_configs, sniff_on_start=sniff_on_start, sniff_before_requests=sniff_before_requests, sniff_on_node_failure=sniff_on_node_failure, sniff_callback=sniff_callback, ) # Create the default metadata for the x-elastic-client-meta # HTTP header. Only requires adding the (service, service_version) # tuple to the beginning of the client_meta self._transport_client_meta: Tuple[Tuple[str, str], ...] = ( client_meta_service, ("py", client_meta_version(python_version())), ("t", client_meta_version(__version__)), ) # Grab the 'HTTP_CLIENT_META' property from the node class http_client_meta = cast( Optional[Tuple[str, str]], getattr(node_class, "_CLIENT_META_HTTP_CLIENT", None), ) if http_client_meta: self._transport_client_meta += (http_client_meta,) if not isinstance(meta_header, bool): raise TypeError("'meta_header' must be of type bool") self.meta_header = meta_header # serialization config _serializers = DEFAULT_SERIALIZERS.copy() # if custom serializers map has been supplied, override the defaults with it if serializers: _serializers.update(serializers) # Create our collection of serializers self.serializers = SerializerCollection( _serializers, default_mimetype=default_mimetype ) # Set of default request options self.max_retries = max_retries self.retry_on_status = retry_on_status self.retry_on_timeout = retry_on_timeout # Build the NodePool from all the options node_pool_kwargs: Dict[str, Any] = {} if node_selector_class is not None: node_pool_kwargs["node_selector_class"] = node_selector_class if dead_node_backoff_factor is not None: node_pool_kwargs["dead_node_backoff_factor"] = dead_node_backoff_factor if max_dead_node_backoff is not None: node_pool_kwargs["max_dead_node_backoff"] = max_dead_node_backoff self.node_pool: NodePool = node_pool_class( node_configs, node_class=node_class, randomize_nodes=randomize_nodes_in_pool, *node_pool_kwargs, ) self._sniff_on_start = sniff_on_start self._sniff_before_requests = sniff_before_requests self._sniff_on_node_failure = sniff_on_node_failure self._sniff_timeout = sniff_timeout self._sniff_callback = sniff_callback self._sniffing_lock = Lock() # Used to track whether we're currently sniffing. self._min_delay_between_sniffing = min_delay_between_sniffing self._last_sniffed_at = 0.0 if sniff_on_start: self.sniff(True) def perform_request( # type: ignore[return] self, method: str, target: str, , body: Optional[Any] = None, headers: Union[Mapping[str, Any], DefaultType] = DEFAULT, max_retries: Union[int, DefaultType] = DEFAULT, retry_on_status: Union[Collection[int], DefaultType] = DEFAULT, retry_on_timeout: Union[bool, DefaultType] = DEFAULT, request_timeout: Union[Optional[float], DefaultType] = DEFAULT, client_meta: Union[Tuple[Tuple[str, str], ...], DefaultType] = DEFAULT, otel_span: Union[OpenTelemetrySpan, DefaultType] = DEFAULT, ) -> TransportApiResponse: """ Perform the actual request. Retrieve a node from the node pool, pass all the information to it's perform_request method and return the data. If an exception was raised, mark the node as failed and retry (up to ``max_retries`` times). If the operation was successful and the node used was previously marked as dead, mark it as live, resetting it's failure count. :arg method: HTTP method to use :arg target: HTTP request target :arg body: body of the request, will be serialized using serializer and passed to the node :arg headers: Additional headers to send with the request. :arg max_retries: Maximum number of retries before giving up on a request. Set to ``0`` to disable retries. :arg retry_on_status: Collection of HTTP status codes to retry. :arg retry_on_timeout: Set to true to retry after timeout errors. :arg request_timeout: Amount of time to wait for a response to fail with a timeout error. :arg client_meta: Extra client metadata key-value pairs to send in the client meta header. :arg otel_span: OpenTelemetry span used to add metadata to the span. :returns: Tuple of the :class:`elastic_transport.ApiResponseMeta` with the deserialized response. """ if headers is DEFAULT: request_headers = HttpHeaders() else: request_headers = HttpHeaders(headers) max_retries = resolve_default(max_retries, self.max_retries) retry_on_timeout = resolve_default(retry_on_timeout, self.retry_on_timeout) retry_on_status = resolve_default(retry_on_status, self.retry_on_status) otel_span = resolve_default(otel_span, OpenTelemetrySpan(None)) if self.meta_header: request_headers["x-elastic-client-meta"] = ",".join( f"{k}={v}" for k, v in self._transport_client_meta + resolve_default(client_meta, ()) ) # Serialize the request body to bytes based on the given mimetype. request_body: Optional[bytes] if body is not None: if "content-type" not in request_headers: raise ValueError( "Must provide a 'Content-Type' header to requests with bodies" ) request_body = self.serializers.dumps( body, mimetype=request_headers["content-type"] ) otel_span.set_db_statement(request_body) else: request_body = None # Errors are stored from (oldest->newest) errors: List[Exception] = [] for attempt in range(max_retries + 1): # If we sniff before requests are made we want to do so before # 'node_pool.get()' is called so our sniffed nodes show up in the pool. if self._sniff_before_requests: self.sniff(False) retry = False node_failure = False last_response: Optional[TransportApiResponse] = None node = self.node_pool.get() start_time = time.time() try: otel_span.set_node_metadata(node.host, node.port, node.base_url, target) resp = node.perform_request( method, target, body=request_body, headers=request_headers, request_timeout=request_timeout, ) _logger.info( "%s %s%s [status:%s duration:%.3fs]" % ( method, node.base_url, target, resp.meta.status, time.time() - start_time, ) ) if method != "HEAD": body = self.serializers.loads(resp.body, resp.meta.mimetype) else: body = None if resp.meta.status in retry_on_status: retry = True # Keep track of the last response we see so we can return # it in case the retried request returns with a transport error. last_response = TransportApiResponse(resp.meta, body) except TransportError as e: _logger.info( "%s %s%s [status:%s duration:%.3fs]" % ( method, node.base_url, target, "N/A", time.time() - start_time, ) ) if isinstance(e, ConnectionTimeout): retry = retry_on_timeout node_failure = True elif isinstance(e, ConnectionError): retry = True node_failure = True # If the error was determined to be a node failure # we mark it dead in the node pool to allow for # other nodes to be retried. if node_failure: self.node_pool.mark_dead(node) if self._sniff_on_node_failure: try: self.sniff(False) except TransportError: # If sniffing on failure, it could fail too. Catch the # exception not to interrupt the retries. pass if not retry or attempt >= max_retries: # Since we're exhausted but we have previously # received some sort of response from the API # we should forward that along instead of the # transport error. Likely to be more actionable. if last_response is not None: return last_response e.errors = tuple(errors) raise else: _logger.warning( "Retrying request after failure (attempt %d of %d)", attempt, max_retries, exc_info=e, ) errors.append(e) else: # If we got back a response we need to check if that status # is indicative of a healthy node even if it's a non-2XX status if ( 200 <= resp.meta.status < 299 or resp.meta.status in NOT_DEAD_NODE_HTTP_STATUSES ): self.node_pool.mark_live(node) else: self.node_pool.mark_dead(node) if self._sniff_on_node_failure: try: self.sniff(False) except TransportError: # If sniffing on failure, it could fail too. Catch the # exception not to interrupt the retries. pass # We either got a response we're happy with or # we've exhausted all of our retries so we return it. if not retry or attempt >= max_retries: return TransportApiResponse(resp.meta, body) else: _logger.warning( "Retrying request after non-successful status %d (attempt %d of %d)", resp.meta.status, attempt, max_retries, ) def sniff(self, is_initial_sniff: bool = False) -> None: previously_sniffed_at = self._last_sniffed_at should_sniff = self._should_sniff(is_initial_sniff) try: if should_sniff: _logger.info("Started sniffing for additional nodes") self._last_sniffed_at = time.time() options = SniffOptions( is_initial_sniff=is_initial_sniff, sniff_timeout=self._sniff_timeout ) assert self._sniff_callback is not None node_configs = self._sniff_callback(self, options) if not node_configs and is_initial_sniff: raise SniffingError( "No viable nodes were discovered on the initial sniff attempt" ) prev_node_pool_size = len(self.node_pool) for node_config in node_configs: self.node_pool.add(node_config) # Do some math to log which nodes are new/existing sniffed_nodes = len(node_configs) new_nodes = sniffed_nodes - (len(self.node_pool) - prev_node_pool_size) existing_nodes = sniffed_nodes - new_nodes _logger.debug( "Discovered %d nodes during sniffing (%d new nodes, %d already in pool)", sniffed_nodes, new_nodes, existing_nodes, ) # If sniffing failed for any reason we # want to allow retrying immediately. except Exception as e: _logger.warning("Encountered an error during sniffing", exc_info=e) self._last_sniffed_at = previously_sniffed_at raise # If we started a sniff we need to release the lock. finally: if should_sniff: self._sniffing_lock.release() def close(self) -> None: """ Explicitly closes all nodes in the transport's pool """ for node in self.node_pool.all(): node.close() def _should_sniff(self, is_initial_sniff: bool) -> bool: """Decide if we should sniff or not. If we return ``True`` from this method the caller has a responsibility to unlock the ``_sniffing_lock`` """ if not is_initial_sniff and ( time.time() - self._last_sniffed_at < self._min_delay_between_sniffing ): return False return self._sniffing_lock.acquire(False) def validate_sniffing_options( *, node_configs: List[NodeConfig], sniff_before_requests: bool, sniff_on_start: bool, sniff_on_node_failure: bool, sniff_callback: Optional[Any], ) -> None: """Validates the Transport configurations for sniffing""" sniffing_enabled = sniff_before_requests or sniff_on_start or sniff_on_node_failure if sniffing_enabled and not sniff_callback: raise ValueError("Enabling sniffing requires specifying a 'sniff_callback'") if not sniffing_enabled and sniff_callback: raise ValueError( "Using 'sniff_callback' requires enabling sniffing via 'sniff_on_start', " "'sniff_before_requests' or 'sniff_on_node_failure'" ) # If we're sniffing we want to warn the user for non-homogenous NodeConfigs. if sniffing_enabled and len(node_configs) > 1: warn_if_varying_node_config_options(node_configs) def warn_if_varying_node_config_options(node_configs: List[NodeConfig]) -> None: """Function which detects situations when sniffing may produce incorrect configs""" exempt_attrs = {"host", "port", "connections_per_node", "_extras", "ssl_context"} match_attr_dict = None for node_config in node_configs: attr_dict = { field.name: getattr(node_config, field.name) for field in dataclasses.fields(node_config) if field.name not in exempt_attrs } if match_attr_dict is None: match_attr_dict = attr_dict # Detected two nodes that have different config, warn the user. elif match_attr_dict != attr_dict: warnings.warn( "Detected NodeConfig instances with different options. " "It's recommended to keep all options except for " "'host' and 'port' the same for sniffing to work reliably.", category=TransportWarning, stacklevel=warn_stacklevel(), )	23.01 KB	2025-05-25 12:26
📄 _utils.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 re from typing import Any, Dict, Union def fixup_module_metadata(module_name: str, namespace: Dict[str, Any]) -> None: # Yoinked from python-trio/outcome, thanks Nathaniel! License: MIT def fix_one(obj: Any) -> None: mod = getattr(obj, "__module__", None) if mod is not None and mod.startswith("elastic_transport."): obj.__module__ = module_name if isinstance(obj, type): for attr_value in obj.__dict__.values(): fix_one(attr_value) for objname in namespace["__all__"]: obj = namespace[objname] fix_one(obj) IPV4_PAT = r"(?:[0-9]{1,3}\.){3}[0-9]{1,3}" IPV4_RE = re.compile("^" + IPV4_PAT + "$") HEX_PAT = "[0-9A-Fa-f]{1,4}" LS32_PAT = "(?:{hex}:{hex}\|{ipv4})".format(hex=HEX_PAT, ipv4=IPV4_PAT) _subs = {"hex": HEX_PAT, "ls32": LS32_PAT} _variations = [ # 6( h16 ":" ) ls32 "(?:%(hex)s:){6}%(ls32)s", # "::" 5( h16 ":" ) ls32 "::(?:%(hex)s:){5}%(ls32)s", # [ h16 ] "::" 4( h16 ":" ) ls32 "(?:%(hex)s)?::(?:%(hex)s:){4}%(ls32)s", # [ 1( h16 ":" ) h16 ] "::" 3( h16 ":" ) ls32 "(?:(?:%(hex)s:)?%(hex)s)?::(?:%(hex)s:){3}%(ls32)s", # [ 2( h16 ":" ) h16 ] "::" 2( h16 ":" ) ls32 "(?:(?:%(hex)s:){0,2}%(hex)s)?::(?:%(hex)s:){2}%(ls32)s", # [ 3( h16 ":" ) h16 ] "::" h16 ":" ls32 "(?:(?:%(hex)s:){0,3}%(hex)s)?::%(hex)s:%(ls32)s", # [ 4( h16 ":" ) h16 ] "::" ls32 "(?:(?:%(hex)s:){0,4}%(hex)s)?::%(ls32)s", # [ 5( h16 ":" ) h16 ] "::" h16 "(?:(?:%(hex)s:){0,5}%(hex)s)?::%(hex)s", # [ 6( h16 ":" ) h16 ] "::" "(?:(?:%(hex)s:){0,6}%(hex)s)?::", ] IPV6_PAT = "(?:" + "\|".join([x % _subs for x in _variations]) + ")" UNRESERVED_PAT = r"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._!\-~" ZONE_ID_PAT = "(?:%25\|%)(?:[" + UNRESERVED_PAT + "]\|%[a-fA-F0-9]{2})+" BRACELESS_IPV6_ADDRZ_PAT = IPV6_PAT + r"(?:" + ZONE_ID_PAT + r")?" BRACELESS_IPV6_ADDRZ_RE = re.compile("^" + BRACELESS_IPV6_ADDRZ_PAT + "$") def is_ipaddress(hostname: Union[str, bytes]) -> bool: """Detects whether the hostname given is an IPv4 or IPv6 address. Also detects IPv6 addresses with Zone IDs. """ # Copied from urllib3. License: MIT if isinstance(hostname, bytes): # IDN A-label bytes are ASCII compatible. hostname = hostname.decode("ascii") hostname = hostname.strip("[]") return bool(IPV4_RE.match(hostname) or BRACELESS_IPV6_ADDRZ_RE.match(hostname))	3.28 KB	2025-05-25 12:26
📄 _version.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. __version__ = "8.17.1"	811 B	2025-05-25 12:26
📄 client_utils.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 base64 import binascii import dataclasses import re import urllib.parse from platform import python_version from typing import Optional, Tuple, TypeVar, Union from urllib.parse import quote as _quote from urllib3.exceptions import LocationParseError from urllib3.util import parse_url from ._models import DEFAULT, DefaultType, NodeConfig from ._utils import fixup_module_metadata from ._version import __version__ __all__ = [ "CloudId", "DEFAULT", "DefaultType", "basic_auth_to_header", "client_meta_version", "create_user_agent", "dataclasses", "parse_cloud_id", "percent_encode", "resolve_default", "to_bytes", "to_str", "url_to_node_config", ] T = TypeVar("T") def resolve_default(val: Union[DefaultType, T], default: T) -> T: """Resolves a value that could be the ``DEFAULT`` sentinel into either the given value or the default value. """ return val if val is not DEFAULT else default def create_user_agent(name: str, version: str) -> str: """Creates the 'User-Agent' header given the library name and version""" return ( f"{name}/{version} (Python/{python_version()}; elastic-transport/{__version__})" ) def client_meta_version(version: str) -> str: """Converts a Python version into a version string compatible with the ``X-Elastic-Client-Meta`` HTTP header. """ match = re.match(r"^([0-9][0-9.][0-9]\|[0-9])(.)$", version) if match is None: raise ValueError( "Version {version!r} not formatted like a Python version string" ) version, version_suffix = match.groups() # Don't treat post-releases as pre-releases. if re.search(r"^\.post[0-9]*$", version_suffix): return version if version_suffix: version += "p" return version @dataclasses.dataclass(frozen=True, repr=True) class CloudId: #: Name of the cluster in Elastic Cloud cluster_name: str #: Host and port of the Elasticsearch instance es_address: Optional[Tuple[str, int]] #: Host and port of the Kibana instance kibana_address: Optional[Tuple[str, int]] def parse_cloud_id(cloud_id: str) -> CloudId: """Parses an Elastic Cloud ID into its components""" try: cloud_id = to_str(cloud_id) cluster_name, _, cloud_id = cloud_id.partition(":") parts = to_str(binascii.a2b_base64(to_bytes(cloud_id, "ascii")), "ascii").split( "$" ) parent_dn = parts[0] if not parent_dn: raise ValueError() # Caught and re-raised properly below es_uuid: Optional[str] kibana_uuid: Optional[str] try: es_uuid = parts[1] except IndexError: es_uuid = None try: kibana_uuid = parts[2] or None except IndexError: kibana_uuid = None if ":" in parent_dn: parent_dn, _, parent_port = parent_dn.rpartition(":") port = int(parent_port) else: port = 443 except (ValueError, IndexError, UnicodeError): raise ValueError("Cloud ID is not properly formatted") from None es_host = f"{es_uuid}.{parent_dn}" if es_uuid else None kibana_host = f"{kibana_uuid}.{parent_dn}" if kibana_uuid else None return CloudId( cluster_name=cluster_name, es_address=(es_host, port) if es_host else None, kibana_address=(kibana_host, port) if kibana_host else None, ) def to_str( value: Union[str, bytes], encoding: str = "utf-8", errors: str = "strict" ) -> str: if isinstance(value, bytes): return value.decode(encoding, errors) return value def to_bytes( value: Union[str, bytes], encoding: str = "utf-8", errors: str = "strict" ) -> bytes: if isinstance(value, str): return value.encode(encoding, errors) return value # Python 3.7 added '~' to the safe list for urllib.parse.quote() _QUOTE_ALWAYS_SAFE = frozenset( "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_.-~" ) def percent_encode( string: Union[bytes, str], safe: str = "/", encoding: Optional[str] = None, errors: Optional[str] = None, ) -> str: """Percent-encodes a string so it can be used in an HTTP request target""" # Redefines 'urllib.parse.quote()' to always have the '~' character # within the 'ALWAYS_SAFE' list. The character was added in Python 3.7 safe = "".join(_QUOTE_ALWAYS_SAFE.union(set(safe))) return _quote(string, safe, encoding=encoding, errors=errors) # type: ignore[arg-type] def basic_auth_to_header(basic_auth: Tuple[str, str]) -> str: """Converts a 2-tuple into a 'Basic' HTTP Authorization header""" if ( not isinstance(basic_auth, tuple) or len(basic_auth) != 2 or any(not isinstance(item, (str, bytes)) for item in basic_auth) ): raise ValueError( "'basic_auth' must be a 2-tuple of str/bytes (username, password)" ) return ( f"Basic {base64.b64encode(b':'.join(to_bytes(x) for x in basic_auth)).decode()}" ) def url_to_node_config( url: str, use_default_ports_for_scheme: bool = False ) -> NodeConfig: """Constructs a :class:`elastic_transport.NodeConfig` instance from a URL. If a username/password are specified in the URL they are converted to an 'Authorization' header. Always fills in a default port for HTTPS. :param url: URL to transform into a NodeConfig. :param use_default_ports_for_scheme: If 'True' will resolve default ports for HTTP. """ try: parsed_url = parse_url(url) except LocationParseError: raise ValueError(f"Could not parse URL {url!r}") from None parsed_port: Optional[int] = parsed_url.port if parsed_url.port is None and parsed_url.scheme is not None: # Always fill in a default port for HTTPS if parsed_url.scheme == "https": parsed_port = 443 # Only fill HTTP default port when asked to explicitly elif parsed_url.scheme == "http" and use_default_ports_for_scheme: parsed_port = 80 if any( component in (None, "") for component in (parsed_url.scheme, parsed_url.host, parsed_port) ): raise ValueError( "URL must include a 'scheme', 'host', and 'port' component (ie 'https://localhost:9200')" ) assert parsed_url.scheme is not None assert parsed_url.host is not None assert parsed_port is not None headers = {} if parsed_url.auth: # `urllib3.util.url_parse` ensures `parsed_url` is correctly # percent-encoded but does not percent-decode userinfo, so we have to # do it ourselves to build the basic auth header correctly. encoded_username, _, encoded_password = parsed_url.auth.partition(":") username = urllib.parse.unquote(encoded_username) password = urllib.parse.unquote(encoded_password) headers["authorization"] = basic_auth_to_header((username, password)) host = parsed_url.host.strip("[]") if not parsed_url.path or parsed_url.path == "/": path_prefix = "" else: path_prefix = parsed_url.path return NodeConfig( scheme=parsed_url.scheme, host=host, port=parsed_port, path_prefix=path_prefix, headers=headers, ) fixup_module_metadata(__name__, globals()) del fixup_module_metadata	8.03 KB	2025-05-25 12:26
📄 py.typed	0 B	2025-05-25 12:26

Terminal