python elasticsearch 8.x通过署理发起请求方法
由于python elasticsearch v8 engine的源码包中并未开放对于请求添加proxies的支持，导致在某些环境下无法连通外网的es服务。目前网上暂无相关的修改内容，我这边提供下自己修改的动态运行时更换elasticsearch包的源码方法demoimport gzip
import ssl
import time
import requests
from elastic_transport._node._http_requests import RequestsHttpNode
from typing import Any, Optional, Union
from elastic_transport._compat import warn_stacklevel
from elastic_transport._exceptions import ConnectionError, ConnectionTimeout, SecurityWarning, TlsError
from elastic_transport._models import ApiResponseMeta, HttpHeaders, NodeConfig
from elastic_transport.client_utils import DEFAULT, DefaultType, client_meta_version
from elastic_transport._node._base import (
BUILTIN_EXCEPTIONS,
RERAISE_EXCEPTIONS,
BaseNode,
NodeApiResponse,
ssl_context_from_node_config,
)
def custom_perform_request(
self,
method: str,
target: str,
body: Optional = None,
headers: Optional = None,
request_timeout: Union] = DEFAULT,
) -> NodeApiResponse:
url = self.base_url + target
headers = HttpHeaders(headers or ())
request_headers = self._headers.copy()
if headers:
request_headers.update(headers)
body_to_send: Optional
if body:
if self._http_compress:
body_to_send = gzip.compress(body)
request_headers["content-encoding"] = "gzip"
else:
body_to_send = body
else:
body_to_send = None
start = time.time()
proxies_dict = {
"http": "http://xx.xx.xx.xx:xx",
"http": "http://xx.xx.xx.xx:xx",
}
request = requests.Request(
method=method, headers=request_headers, url=url, data=body_to_send
)
prepared_request = self.session.prepare_request(request)
send_kwargs = {
"timeout": (
request_timeout
if request_timeout is not DEFAULT
else self.config.request_timeout
)
}
send_kwargs.update(
self.session.merge_environment_settings(# type: ignore
prepared_request.url, {}, None, None, None
)
)
send_kwargs.pop('proxies')
try:
response = self.session.send(prepared_request, proxies=proxies_dict,**send_kwargs)# type: ignore
data = response.content
duration = time.time() - start
response_headers = HttpHeaders(response.headers)
except RERAISE_EXCEPTIONS:
raise
except Exception as e:
err: Exception
if isinstance(e, requests.Timeout):
err = ConnectionTimeout(
"Connection timed out during request", errors=(e,)
)
elif isinstance(e, (ssl.SSLError, requests.exceptions.SSLError)):
err = TlsError(str(e), errors=(e,))
elif isinstance(e, BUILTIN_EXCEPTIONS):
raise
else:
err = ConnectionError(str(e), errors=(e,))
self._log_request(
method=method,
target=target,
headers=request_headers,
body=body,
exception=err,
)
raise err from None
meta = ApiResponseMeta(
node=self.config,
duration=duration,
http_version="1.1",
status=response.status_code,
headers=response_headers,
)
self._log_request(
method=method,
target=target,
headers=request_headers,
body=body,
meta=meta,
response=data,
)
return NodeApiResponse(
meta,
data,
)
RequestsHttpNode.perform_request = custom_perform_request
from elasticsearch import Elasticsearch
es = Elasticsearch(hosts=["http://xx.xx.xxx.xxx:9200"], basic_auth=("elastic", "xxxxxxxxxxxx"), node_class=RequestsHttpNode)
query = {"query": {"match_all": {}}}
response = es.search(index="xxxxxx_prod", body= query)
print(response)
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]