update to ES 7 (#911)

* update to ES 7

* more updates to make tests run

* update the test matrix

* remove python 3.3

* update xpack apis

* relative imports
This commit is contained in:
Nick Lang
2019-03-29 09:25:23 -06:00
committed by GitHub
parent 3c2509d29d
commit 1afceba69b
30 changed files with 3560 additions and 1509 deletions
+2 -16
View File
@@ -1,28 +1,14 @@
---
ELASTICSEARCH_VERSION:
- 6.4.2
- 6.3.2
- 6.1.4
- 6.0.1
- 7.0.0-beta1
- 6.6.2
PYTHON_VERSION:
- 2.6
- 2.7
- 3.3
- 3.4
- 3.5
- 3.6
- 3.7
exclude:
# since a base image for python2.6 doens't exist i need to create one.
- PYTHON_VERSION: 2.6
# Something is wrong with the way 3.7 and the es-client handle multiprocessing.
# https://github.com/elastic/elasticsearch-py/issues/865
- PYTHON_VERSION: 3.7
# new feature in 6.4.2 which is causing a test failure.
# https://github.com/elastic/elasticsearch-py/issues/866
- ELASTICSEARCH_VERSION: 6.4.2
+2
View File
@@ -28,6 +28,8 @@ services:
- "repositories.url.allowed_urls=http://*"
- node.attr.testattr=test
- bootstrap.memory_lock=false
- "node.name=test"
- "cluster.initial_master_nodes=test"
- "discovery.zen.ping.unicast.hosts=elasticsearch"
- "http.max_content_length=5mb"
networks:
File diff suppressed because it is too large Load Diff
+297 -136
View File
@@ -1,7 +1,8 @@
from .utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class IndicesClient(NamespacedClient):
@query_params('format', 'prefer_local')
@query_params("format", "prefer_local")
def analyze(self, index=None, body=None, params=None):
"""
Perform the analysis process on a text and return the tokens breakdown of the text.
@@ -15,10 +16,11 @@ class IndicesClient(NamespacedClient):
:arg prefer_local: With `true`, specify that a local shard should be
used if available, with `false`, use a random shard (default: true)
"""
return self.transport.perform_request('GET', _make_path(index,
'_analyze'), params=params, body=body)
return self.transport.perform_request(
"GET", _make_path(index, "_analyze"), params=params, body=body
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable')
@query_params("allow_no_indices", "expand_wildcards", "ignore_unavailable")
def refresh(self, index=None, params=None):
"""
Explicitly refresh one or more index, making all operations performed
@@ -36,11 +38,17 @@ class IndicesClient(NamespacedClient):
:arg ignore_unavailable: Whether specified concrete indices should be
ignored when unavailable (missing or closed)
"""
return self.transport.perform_request('POST', _make_path(index,
'_refresh'), params=params)
return self.transport.perform_request(
"POST", _make_path(index, "_refresh"), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'force',
'ignore_unavailable', 'wait_if_ongoing')
@query_params(
"allow_no_indices",
"expand_wildcards",
"force",
"ignore_unavailable",
"wait_if_ongoing",
)
def flush(self, index=None, params=None):
"""
Explicitly flush one or more indices.
@@ -66,10 +74,13 @@ class IndicesClient(NamespacedClient):
already executing. The default is true. If set to false the flush
will be skipped iff if another flush operation is already running.
"""
return self.transport.perform_request('POST', _make_path(index,
'_flush'), params=params)
return self.transport.perform_request(
"POST", _make_path(index, "_flush"), params=params
)
@query_params('master_timeout', 'timeout', 'wait_for_active_shards', 'include_type_name')
@query_params(
"master_timeout", "timeout", "wait_for_active_shards", "include_type_name"
)
def create(self, index, body=None, params=None):
"""
Create an index in Elasticsearch.
@@ -86,11 +97,19 @@ class IndicesClient(NamespacedClient):
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request('PUT', _make_path(index),
params=params, body=body)
return self.transport.perform_request(
"PUT", _make_path(index), params=params, body=body
)
@query_params('allow_no_indices', 'expand_wildcards', 'flat_settings',
'ignore_unavailable', 'include_defaults', 'local', 'include_type_name')
@query_params(
"allow_no_indices",
"expand_wildcards",
"flat_settings",
"ignore_unavailable",
"include_defaults",
"local",
"include_type_name",
)
def get(self, index, feature=None, params=None):
"""
The get index API allows to retrieve information about one or more indexes.
@@ -113,11 +132,17 @@ class IndicesClient(NamespacedClient):
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request('GET', _make_path(index,
feature), params=params)
return self.transport.perform_request(
"GET", _make_path(index, feature), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'master_timeout', 'timeout')
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"master_timeout",
"timeout",
)
def open(self, index, params=None):
"""
Open a closed index to make it available for search.
@@ -137,11 +162,17 @@ class IndicesClient(NamespacedClient):
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request('POST', _make_path(index,
'_open'), params=params)
return self.transport.perform_request(
"POST", _make_path(index, "_open"), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'master_timeout', 'timeout')
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"master_timeout",
"timeout",
)
def close(self, index, params=None):
"""
Close an index to remove it's overhead from the cluster. Closed index
@@ -162,11 +193,17 @@ class IndicesClient(NamespacedClient):
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request('POST', _make_path(index,
'_close'), params=params)
return self.transport.perform_request(
"POST", _make_path(index, "_close"), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'master_timeout', 'timeout')
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"master_timeout",
"timeout",
)
def delete(self, index, params=None):
"""
Delete an index in Elasticsearch
@@ -185,11 +222,18 @@ class IndicesClient(NamespacedClient):
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request('DELETE', _make_path(index),
params=params)
return self.transport.perform_request(
"DELETE", _make_path(index), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'flat_settings',
'ignore_unavailable', 'include_defaults', 'local')
@query_params(
"allow_no_indices",
"expand_wildcards",
"flat_settings",
"ignore_unavailable",
"include_defaults",
"local",
)
def exists(self, index, params=None):
"""
Return a boolean indicating whether given index exists.
@@ -210,11 +254,9 @@ class IndicesClient(NamespacedClient):
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request('HEAD', _make_path(index),
params=params)
return self.transport.perform_request("HEAD", _make_path(index), params=params)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'local')
@query_params("allow_no_indices", "expand_wildcards", "ignore_unavailable", "local")
def exists_type(self, index, doc_type, params=None):
"""
Check if a type/types exists in an index/indices.
@@ -237,12 +279,19 @@ class IndicesClient(NamespacedClient):
for param in (index, doc_type):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('HEAD', _make_path(index,
'_mapping', doc_type), params=params)
return self.transport.perform_request(
"HEAD", _make_path(index, "_mapping", doc_type), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'master_timeout', 'timeout','include_type_name')
def put_mapping(self, doc_type, body, index=None, params=None):
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"master_timeout",
"timeout",
"include_type_name",
)
def put_mapping(self, body, doc_type=None, index=None, params=None):
"""
Register specific mapping definition for a specific type.
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/indices-put-mapping.html>`_
@@ -265,14 +314,20 @@ class IndicesClient(NamespacedClient):
:arg include_type_name: Specify whether requests and responses should include a
type name (default: depends on Elasticsearch version).
"""
for param in (doc_type, body):
for param in (body,):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('PUT', _make_path(index,
'_mapping', doc_type), params=params, body=body)
return self.transport.perform_request(
"PUT", _make_path(index, "_mapping", doc_type), params=params, body=body
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'local','include_type_name')
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"local",
"include_type_name",
)
def get_mapping(self, index=None, doc_type=None, params=None):
"""
Retrieve mapping definition of index or index/type.
@@ -293,11 +348,18 @@ class IndicesClient(NamespacedClient):
:arg include_type_name: Specify whether requests and responses should include a
type name (default: depends on Elasticsearch version).
"""
return self.transport.perform_request('GET', _make_path(index,
'_mapping', doc_type), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_mapping", doc_type), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'include_defaults', 'local', 'include_type_name')
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"include_defaults",
"local",
"include_type_name",
)
def get_field_mapping(self, fields, index=None, doc_type=None, params=None):
"""
Retrieve mapping definition of a specific field.
@@ -323,10 +385,13 @@ class IndicesClient(NamespacedClient):
"""
if fields in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'fields'.")
return self.transport.perform_request('GET', _make_path(index,
'_mapping', doc_type, 'field', fields), params=params)
return self.transport.perform_request(
"GET",
_make_path(index, "_mapping", doc_type, "field", fields),
params=params,
)
@query_params('master_timeout', 'timeout')
@query_params("master_timeout", "timeout")
def put_alias(self, index, name, body=None, params=None):
"""
Create an alias for a specific index/indices.
@@ -343,11 +408,11 @@ class IndicesClient(NamespacedClient):
for param in (index, name):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('PUT', _make_path(index,
'_alias', name), params=params, body=body)
return self.transport.perform_request(
"PUT", _make_path(index, "_alias", name), params=params, body=body
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'local')
@query_params("allow_no_indices", "expand_wildcards", "ignore_unavailable", "local")
def exists_alias(self, index=None, name=None, params=None):
"""
Return a boolean indicating whether given alias exists.
@@ -366,11 +431,11 @@ class IndicesClient(NamespacedClient):
:arg local: Return local information, do not retrieve the state from
master node (default: false)
"""
return self.transport.perform_request('HEAD', _make_path(index, '_alias',
name), params=params)
return self.transport.perform_request(
"HEAD", _make_path(index, "_alias", name), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'local')
@query_params("allow_no_indices", "expand_wildcards", "ignore_unavailable", "local")
def get_alias(self, index=None, name=None, params=None):
"""
Retrieve a specified alias.
@@ -389,10 +454,11 @@ class IndicesClient(NamespacedClient):
:arg local: Return local information, do not retrieve the state from
master node (default: false)
"""
return self.transport.perform_request('GET', _make_path(index,
'_alias', name), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_alias", name), params=params
)
@query_params('master_timeout', 'timeout')
@query_params("master_timeout", "timeout")
def update_aliases(self, body, params=None):
"""
Update specified aliases.
@@ -404,10 +470,11 @@ class IndicesClient(NamespacedClient):
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request('POST', '/_aliases',
params=params, body=body)
return self.transport.perform_request(
"POST", "/_aliases", params=params, body=body
)
@query_params('master_timeout', 'timeout')
@query_params("master_timeout", "timeout")
def delete_alias(self, index, name, params=None):
"""
Delete specific alias.
@@ -424,11 +491,18 @@ class IndicesClient(NamespacedClient):
for param in (index, name):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('DELETE', _make_path(index,
'_alias', name), params=params)
return self.transport.perform_request(
"DELETE", _make_path(index, "_alias", name), params=params
)
@query_params('create', 'flat_settings', 'master_timeout', 'order',
'timeout', 'include_type_name')
@query_params(
"create",
"flat_settings",
"master_timeout",
"order",
"timeout",
"include_type_name",
)
def put_template(self, name, body, params=None):
"""
Create an index template that will automatically be applied to new
@@ -450,10 +524,11 @@ class IndicesClient(NamespacedClient):
for param in (name, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('PUT', _make_path('_template',
name), params=params, body=body)
return self.transport.perform_request(
"PUT", _make_path("_template", name), params=params, body=body
)
@query_params('flat_settings', 'local', 'master_timeout')
@query_params("flat_settings", "local", "master_timeout")
def exists_template(self, name, params=None):
"""
Return a boolean indicating whether given template exists.
@@ -468,10 +543,11 @@ class IndicesClient(NamespacedClient):
"""
if name in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'name'.")
return self.transport.perform_request('HEAD', _make_path('_template',
name), params=params)
return self.transport.perform_request(
"HEAD", _make_path("_template", name), params=params
)
@query_params('flat_settings', 'local', 'master_timeout', 'include_type_name')
@query_params("flat_settings", "local", "master_timeout", "include_type_name")
def get_template(self, name=None, params=None):
"""
Retrieve an index template by its name.
@@ -486,10 +562,11 @@ class IndicesClient(NamespacedClient):
:arg include_type_name: Specify whether requests and responses should include a
type name (default: depends on Elasticsearch version).
"""
return self.transport.perform_request('GET', _make_path('_template',
name), params=params)
return self.transport.perform_request(
"GET", _make_path("_template", name), params=params
)
@query_params('master_timeout', 'timeout')
@query_params("master_timeout", "timeout")
def delete_template(self, name, params=None):
"""
Delete an index template by its name.
@@ -501,11 +578,18 @@ class IndicesClient(NamespacedClient):
"""
if name in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'name'.")
return self.transport.perform_request('DELETE',
_make_path('_template', name), params=params)
return self.transport.perform_request(
"DELETE", _make_path("_template", name), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'flat_settings',
'ignore_unavailable', 'include_defaults', 'local')
@query_params(
"allow_no_indices",
"expand_wildcards",
"flat_settings",
"ignore_unavailable",
"include_defaults",
"local",
)
def get_settings(self, index=None, name=None, params=None):
"""
Retrieve settings for one or more (or all) indices.
@@ -528,11 +612,18 @@ class IndicesClient(NamespacedClient):
:arg local: Return local information, do not retrieve the state from
master node (default: false)
"""
return self.transport.perform_request('GET', _make_path(index,
'_settings', name), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_settings", name), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'flat_settings',
'ignore_unavailable', 'master_timeout', 'preserve_existing')
@query_params(
"allow_no_indices",
"expand_wildcards",
"flat_settings",
"ignore_unavailable",
"master_timeout",
"preserve_existing",
)
def put_settings(self, body, index=None, params=None):
"""
Change specific index level settings in real time.
@@ -557,11 +648,19 @@ class IndicesClient(NamespacedClient):
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request('PUT', _make_path(index,
'_settings'), params=params, body=body)
return self.transport.perform_request(
"PUT", _make_path(index, "_settings"), params=params, body=body
)
@query_params('completion_fields', 'fielddata_fields', 'fields', 'groups',
'include_segment_file_sizes', 'level', 'types')
@query_params(
"completion_fields",
"fielddata_fields",
"fields",
"groups",
"include_segment_file_sizes",
"level",
"types",
)
def stats(self, index=None, metric=None, params=None):
"""
Retrieve statistics on different operations happening on an index.
@@ -586,11 +685,17 @@ class IndicesClient(NamespacedClient):
:arg types: A comma-separated list of document types for the `indexing`
index metric
"""
return self.transport.perform_request('GET', _make_path(index,
'_stats', metric), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_stats", metric), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'operation_threading', 'verbose')
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"operation_threading",
"verbose",
)
def segments(self, index=None, params=None):
"""
Provide low level segments information that a Lucene index (shard level) is built with.
@@ -609,12 +714,25 @@ class IndicesClient(NamespacedClient):
:arg operation_threading: TODO: ?
:arg verbose: Includes detailed memory usage by Lucene., default False
"""
return self.transport.perform_request('GET', _make_path(index,
'_segments'), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_segments"), params=params
)
@query_params('all_shards', 'allow_no_indices', 'analyze_wildcard',
'analyzer', 'default_operator', 'df', 'expand_wildcards', 'explain',
'ignore_unavailable', 'lenient', 'operation_threading', 'q', 'rewrite')
@query_params(
"all_shards",
"allow_no_indices",
"analyze_wildcard",
"analyzer",
"default_operator",
"df",
"expand_wildcards",
"explain",
"ignore_unavailable",
"lenient",
"operation_threading",
"q",
"rewrite",
)
def validate_query(self, index=None, doc_type=None, body=None, params=None):
"""
Validate a potentially expensive query without executing it.
@@ -651,12 +769,24 @@ class IndicesClient(NamespacedClient):
:arg rewrite: Provide a more detailed explanation showing the actual
Lucene query that will be executed.
"""
return self.transport.perform_request('GET', _make_path(index,
doc_type, '_validate', 'query'), params=params, body=body)
return self.transport.perform_request(
"GET",
_make_path(index, doc_type, "_validate", "query"),
params=params,
body=body,
)
@query_params('allow_no_indices', 'expand_wildcards', 'field_data',
'fielddata', 'fields', 'ignore_unavailable', 'query', 'recycler',
'request')
@query_params(
"allow_no_indices",
"expand_wildcards",
"field_data",
"fielddata",
"fields",
"ignore_unavailable",
"query",
"recycler",
"request",
)
def clear_cache(self, index=None, params=None):
"""
Clear either all caches or specific cached associated with one ore more indices.
@@ -680,10 +810,11 @@ class IndicesClient(NamespacedClient):
:arg request: Clear request cache
:arg request_cache: Clear request cache
"""
return self.transport.perform_request('POST', _make_path(index,
'_cache', 'clear'), params=params)
return self.transport.perform_request(
"POST", _make_path(index, "_cache", "clear"), params=params
)
@query_params('active_only', 'detailed')
@query_params("active_only", "detailed")
def recovery(self, index=None, params=None):
"""
The indices recovery API provides insight into on-going shard
@@ -698,11 +829,17 @@ class IndicesClient(NamespacedClient):
:arg detailed: Whether to display detailed information about shard
recovery, default False
"""
return self.transport.perform_request('GET', _make_path(index,
'_recovery'), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_recovery"), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'only_ancient_segments', 'wait_for_completion')
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"only_ancient_segments",
"wait_for_completion",
)
def upgrade(self, index=None, params=None):
"""
Upgrade one or more indices to the latest format through an API.
@@ -723,10 +860,11 @@ class IndicesClient(NamespacedClient):
:arg wait_for_completion: Specify whether the request should block until
the all segments are upgraded (default: false)
"""
return self.transport.perform_request('POST', _make_path(index,
'_upgrade'), params=params)
return self.transport.perform_request(
"POST", _make_path(index, "_upgrade"), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable')
@query_params("allow_no_indices", "expand_wildcards", "ignore_unavailable")
def get_upgrade(self, index=None, params=None):
"""
Monitor how much of one or more index is upgraded.
@@ -743,10 +881,11 @@ class IndicesClient(NamespacedClient):
:arg ignore_unavailable: Whether specified concrete indices should be
ignored when unavailable (missing or closed)
"""
return self.transport.perform_request('GET', _make_path(index,
'_upgrade'), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_upgrade"), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable')
@query_params("allow_no_indices", "expand_wildcards", "ignore_unavailable")
def flush_synced(self, index=None, params=None):
"""
Perform a normal flush, then add a generated unique marker (sync_id) to all shards.
@@ -763,11 +902,17 @@ class IndicesClient(NamespacedClient):
:arg ignore_unavailable: Whether specified concrete indices should be
ignored when unavailable (missing or closed)
"""
return self.transport.perform_request('POST', _make_path(index,
'_flush', 'synced'), params=params)
return self.transport.perform_request(
"POST", _make_path(index, "_flush", "synced"), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable',
'operation_threading', 'status')
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"operation_threading",
"status",
)
def shard_stores(self, index=None, params=None):
"""
Provides store information for shard copies of indices. Store
@@ -791,12 +936,20 @@ class IndicesClient(NamespacedClient):
to get store information for, valid choices are: 'green', 'yellow',
'red', 'all'
"""
return self.transport.perform_request('GET', _make_path(index,
'_shard_stores'), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_shard_stores"), params=params
)
@query_params('allow_no_indices', 'expand_wildcards', 'flush',
'ignore_unavailable', 'max_num_segments', 'only_expunge_deletes',
'operation_threading', 'wait_for_merge')
@query_params(
"allow_no_indices",
"expand_wildcards",
"flush",
"ignore_unavailable",
"max_num_segments",
"only_expunge_deletes",
"operation_threading",
"wait_for_merge",
)
def forcemerge(self, index=None, params=None):
"""
The force merge API allows to force merging of one or more indices
@@ -827,10 +980,11 @@ class IndicesClient(NamespacedClient):
expunge deleted documents
:arg operation_threading: TODO: ?
"""
return self.transport.perform_request('POST', _make_path(index,
'_forcemerge'), params=params)
return self.transport.perform_request(
"POST", _make_path(index, "_forcemerge"), params=params
)
@query_params('master_timeout', 'timeout', 'wait_for_active_shards')
@query_params("master_timeout", "timeout", "wait_for_active_shards")
def shrink(self, index, target, body=None, params=None):
"""
The shrink index API allows you to shrink an existing index into a new
@@ -856,11 +1010,17 @@ class IndicesClient(NamespacedClient):
for param in (index, target):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('PUT', _make_path(index,
'_shrink', target), params=params, body=body)
return self.transport.perform_request(
"PUT", _make_path(index, "_shrink", target), params=params, body=body
)
@query_params('dry_run', 'master_timeout', 'timeout',
'wait_for_active_shards', 'include_type_name')
@query_params(
"dry_run",
"master_timeout",
"timeout",
"wait_for_active_shards",
"include_type_name",
)
def rollover(self, alias, new_index=None, body=None, params=None):
"""
The rollover index API rolls an alias over to a new index when the
@@ -887,5 +1047,6 @@ class IndicesClient(NamespacedClient):
"""
if alias in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'alias'.")
return self.transport.perform_request('POST', _make_path(alias,
'_rollover', new_index), params=params, body=body)
return self.transport.perform_request(
"POST", _make_path(alias, "_rollover", new_index), params=params, body=body
)
+43 -16
View File
@@ -1,7 +1,19 @@
from .utils import NamespacedClient, query_params, _make_path
class NodesClient(NamespacedClient):
@query_params('flat_settings', 'timeout')
@query_params()
def reload_secure_settings(self, params=None):
"""
Reload any settings that have been marked as "reloadable"
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/secure-settings.html#reloadable-secure-settings>`_
"""
return self.transport.perform_request(
"POST", _make_path("_nodes", "reload_secure_settings"), params=params
)
@query_params("flat_settings", "timeout")
def info(self, node_id=None, metric=None, params=None):
"""
The cluster nodes info API allows to retrieve one or more (or all) of
@@ -17,11 +29,20 @@ class NodesClient(NamespacedClient):
:arg flat_settings: Return settings in flat format (default: false)
:arg timeout: Explicit operation timeout
"""
return self.transport.perform_request('GET', _make_path('_nodes',
node_id, metric), params=params)
return self.transport.perform_request(
"GET", _make_path("_nodes", node_id, metric), params=params
)
@query_params('completion_fields', 'fielddata_fields', 'fields', 'groups',
'include_segment_file_sizes', 'level', 'timeout', 'types')
@query_params(
"completion_fields",
"fielddata_fields",
"fields",
"groups",
"include_segment_file_sizes",
"level",
"timeout",
"types",
)
def stats(self, node_id=None, metric=None, index_metric=None, params=None):
"""
The cluster nodes stats API allows to retrieve one or more (or all) of
@@ -54,11 +75,15 @@ class NodesClient(NamespacedClient):
:arg types: A comma-separated list of document types for the `indexing`
index metric
"""
return self.transport.perform_request('GET', _make_path('_nodes',
node_id, 'stats', metric, index_metric), params=params)
return self.transport.perform_request(
"GET",
_make_path("_nodes", node_id, "stats", metric, index_metric),
params=params,
)
@query_params('type', 'ignore_idle_threads', 'interval', 'snapshots',
'threads', 'timeout')
@query_params(
"type", "ignore_idle_threads", "interval", "snapshots", "threads", "timeout"
)
def hot_threads(self, node_id=None, params=None):
"""
An API allowing to get the current hot threads on each node in the cluster.
@@ -80,12 +105,13 @@ class NodesClient(NamespacedClient):
:arg timeout: Explicit operation timeout
"""
# avoid python reserved words
if params and 'type_' in params:
params['type'] = params.pop('type_')
return self.transport.perform_request('GET', _make_path('_cluster',
'nodes', node_id, 'hotthreads'), params=params)
if params and "type_" in params:
params["type"] = params.pop("type_")
return self.transport.perform_request(
"GET", _make_path("_cluster", "nodes", node_id, "hotthreads"), params=params
)
@query_params('human', 'timeout')
@query_params("human", "timeout")
def usage(self, node_id=None, metric=None, params=None):
"""
The cluster nodes usage API allows to retrieve information on the usage
@@ -101,5 +127,6 @@ class NodesClient(NamespacedClient):
format., default False
:arg timeout: Explicit operation timeout
"""
return self.transport.perform_request('GET', _make_path('_nodes',
node_id, 'usage', metric), params=params)
return self.transport.perform_request(
"GET", _make_path("_nodes", node_id, "usage", metric), params=params
)
+22 -11
View File
@@ -6,7 +6,8 @@ from functools import wraps
from ..compat import string_types, quote_plus, PY2
# parts of URL to be omitted
SKIP_IN_PATH = (None, '', b'', [], ())
SKIP_IN_PATH = (None, "", b"", [], ())
def _escape(value):
"""
@@ -16,7 +17,7 @@ def _escape(value):
# make sequences into comma-separated stings
if isinstance(value, (list, tuple)):
value = ','.join(value)
value = ",".join(value)
# dates and datetimes into isoformat
elif isinstance(value, (date, datetime)):
@@ -33,36 +34,43 @@ def _escape(value):
# encode strings to utf-8
if isinstance(value, string_types):
if PY2 and isinstance(value, unicode):
return value.encode('utf-8')
return value.encode("utf-8")
if not PY2 and isinstance(value, str):
return value.encode('utf-8')
return value.encode("utf-8")
return str(value)
def _make_path(*parts):
"""
Create a URL string from parts, omit all `None` values and empty strings.
Convert lists nad tuples to comma separated values.
"""
#TODO: maybe only allow some parts to be lists/tuples ?
return '/' + '/'.join(
# TODO: maybe only allow some parts to be lists/tuples ?
return "/" + "/".join(
# preserve ',' and '*' in url for nicer URLs in logs
quote_plus(_escape(p), b',*') for p in parts if p not in SKIP_IN_PATH)
quote_plus(_escape(p), b",*")
for p in parts
if p not in SKIP_IN_PATH
)
# parameters that apply to all methods
GLOBAL_PARAMS = ('pretty', 'human', 'error_trace', 'format', 'filter_path')
GLOBAL_PARAMS = ("pretty", "human", "error_trace", "format", "filter_path")
def query_params(*es_query_params):
"""
Decorator that pops all accepted parameters from method's kwargs and puts
them in the params argument.
"""
def _wrapper(func):
@wraps(func)
def _wrapped(*args, **kwargs):
params = {}
if 'params' in kwargs:
params = kwargs.pop('params').copy()
if "params" in kwargs:
params = kwargs.pop("params").copy()
for p in es_query_params + GLOBAL_PARAMS:
if p in kwargs:
v = kwargs.pop(p)
@@ -70,11 +78,13 @@ def query_params(*es_query_params):
params[p] = _escape(v)
# don't treat ignore and request_timeout as other params to avoid escaping
for p in ('ignore', 'request_timeout'):
for p in ("ignore", "request_timeout"):
if p in kwargs:
params[p] = kwargs.pop(p)
return func(*args, params=params, **kwargs)
return _wrapped
return _wrapper
@@ -86,6 +96,7 @@ class NamespacedClient(object):
def transport(self):
return self.client.transport
class AddonClient(NamespacedClient):
@classmethod
def infect_client(cls, client):
+6 -6
View File
@@ -9,8 +9,9 @@ from .ml import MlClient
from .migration import MigrationClient
from .deprecation import DeprecationClient
class XPackClient(NamespacedClient):
namespace = 'xpack'
namespace = "xpack"
def __init__(self, *args, **kwargs):
super(XPackClient, self).__init__(*args, **kwargs)
@@ -23,7 +24,7 @@ class XPackClient(NamespacedClient):
self.migration = MigrationClient(self.client)
self.deprecation = DeprecationClient(self.client)
@query_params('categories', 'human')
@query_params("categories", "human")
def info(self, params=None):
"""
Retrieve information about xpack, including build number/timestamp and license status
@@ -34,14 +35,13 @@ class XPackClient(NamespacedClient):
:arg human: Presents additional info for humans (feature descriptions
and X-Pack tagline)
"""
return self.transport.perform_request('GET', '/_xpack', params=params)
return self.transport.perform_request("GET", "/_xpack", params=params)
@query_params('master_timeout')
@query_params("master_timeout")
def usage(self, params=None):
"""
Retrieve information about xpack features usage
:arg master_timeout: Specify timeout for watch write operation
"""
return self.transport.perform_request('GET', '/_xpack/usage',
params=params)
return self.transport.perform_request("GET", "/_xpack/usage", params=params)
+161
View File
@@ -0,0 +1,161 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class CcrClient(NamespacedClient):
@query_params()
def delete_auto_follow_pattern(self, name, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-delete-auto-follow-pattern.html>`_
:arg name: The name of the auto follow pattern.
"""
if name in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'name'.")
return self.transport.perform_request(
"DELETE", _make_path("_ccr", "auto_follow", name), params=params
)
@query_params("wait_for_active_shards")
def follow(self, index, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-put-follow.html>`_
:arg index: The name of the follower index
:arg body: The name of the leader index and other optional ccr related
parameters
:arg wait_for_active_shards: Sets the number of shard copies that must
be active before returning. Defaults to 0. Set to `all` for all
shard copies, otherwise set to any non-negative value less than or
equal to the total number of copies for the shard (number of
replicas + 1), default '0'
"""
for param in (index, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"PUT", _make_path(index, "_ccr", "follow"), params=params, body=body
)
@query_params()
def follow_info(self, index=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-get-follow-info.html>`_
:arg index: A comma-separated list of index patterns; use `_all` to
perform the operation on all indices
"""
return self.transport.perform_request(
"GET", _make_path(index, "_ccr", "info"), params=params
)
@query_params()
def follow_stats(self, index, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-get-follow-stats.html>`_
:arg index: A comma-separated list of index patterns; use `_all` to
perform the operation on all indices
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"GET", _make_path(index, "_ccr", "stats"), params=params
)
@query_params()
def forget_follower(self, index, body, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current>`_
:arg index: the name of the leader index for which specified follower
retention leases should be removed
:arg body: the name and UUID of the follower index, the name of the
cluster containing the follower index, and the alias from the
perspective of that cluster for the remote cluster containing the
leader index
"""
for param in (index, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"POST",
_make_path(index, "_ccr", "forget_follower"),
params=params,
body=body,
)
@query_params()
def get_auto_follow_pattern(self, name=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-get-auto-follow-pattern.html>`_
:arg name: The name of the auto follow pattern.
"""
return self.transport.perform_request(
"GET", _make_path("_ccr", "auto_follow", name), params=params
)
@query_params()
def pause_follow(self, index, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-post-pause-follow.html>`_
:arg index: The name of the follower index that should pause following
its leader index.
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST", _make_path(index, "_ccr", "pause_follow"), params=params
)
@query_params()
def put_auto_follow_pattern(self, name, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-put-auto-follow-pattern.html>`_
:arg name: The name of the auto follow pattern.
:arg body: The specification of the auto follow pattern
"""
for param in (name, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"PUT", _make_path("_ccr", "auto_follow", name), params=params, body=body
)
@query_params()
def resume_follow(self, index, body=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-post-resume-follow.html>`_
:arg index: The name of the follow index to resume following.
:arg body: The name of the leader index and other optional ccr related
parameters
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST", _make_path(index, "_ccr", "resume_follow"), params=params, body=body
)
@query_params()
def stats(self, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-get-stats.html>`_
"""
return self.transport.perform_request("GET", "/_ccr/stats", params=params)
@query_params()
def unfollow(self, index, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current>`_
:arg index: The name of the follower index that should be turned into a
regular index.
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST", _make_path(index, "_ccr", "unfollow"), params=params
)
+118
View File
@@ -0,0 +1,118 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class Data_FrameClient(NamespacedClient):
@query_params()
def delete_data_frame_transform(self, transform_id, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/delete-data-frame-transform.html>`_
:arg transform_id: The id of the transform to delete
"""
if transform_id in SKIP_IN_PATH:
raise ValueError(
"Empty value passed for a required argument 'transform_id'."
)
return self.transport.perform_request(
"DELETE",
_make_path("_data_frame", "transforms", transform_id),
params=params,
)
@query_params("from_", "size")
def get_data_frame_transform(self, transform_id=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/get-data-frame-transform.html>`_
:arg transform_id: The id or comma delimited list of id expressions of
the transforms to get, '_all' or '*' implies get all transforms
:arg from_: skips a number of transform configs, defaults to 0
:arg size: specifies a max number of transforms to get, defaults to 100
"""
return self.transport.perform_request(
"GET", _make_path("_data_frame", "transforms", transform_id), params=params
)
@query_params()
def get_data_frame_transform_stats(self, transform_id=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/get-data-frame-transform-stats.html>`_
:arg transform_id: The id of the transform for which to get stats.
'_all' or '*' implies all transforms
"""
return self.transport.perform_request(
"GET",
_make_path("_data_frame", "transforms", transform_id, "_stats"),
params=params,
)
@query_params()
def preview_data_frame_transform(self, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/preview-data-frame-transform.html>`_
:arg body: The definition for the data_frame transform to preview
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request(
"POST", "/_data_frame/transforms/_preview", params=params, body=body
)
@query_params()
def put_data_frame_transform(self, transform_id, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/put-data-frame-transform.html>`_
:arg transform_id: The id of the new transform.
:arg body: The data frame transform definition
"""
for param in (transform_id, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"PUT",
_make_path("_data_frame", "transforms", transform_id),
params=params,
body=body,
)
@query_params("timeout")
def start_data_frame_transform(self, transform_id, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/start-data-frame-transform.html>`_
:arg transform_id: The id of the transform to start
:arg timeout: Controls the time to wait for the transform to start
"""
if transform_id in SKIP_IN_PATH:
raise ValueError(
"Empty value passed for a required argument 'transform_id'."
)
return self.transport.perform_request(
"POST",
_make_path("_data_frame", "transforms", transform_id, "_start"),
params=params,
)
@query_params("timeout", "wait_for_completion")
def stop_data_frame_transform(self, transform_id, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/stop-data-frame-transform.html>`_
:arg transform_id: The id of the transform to stop
:arg timeout: Controls the time to wait until the transform has stopped.
Default to 30 seconds
:arg wait_for_completion: Whether to wait for the transform to fully
stop before returning or not. Default to false
"""
if transform_id in SKIP_IN_PATH:
raise ValueError(
"Empty value passed for a required argument 'transform_id'."
)
return self.transport.perform_request(
"POST",
_make_path("_data_frame", "transforms", transform_id, "_stop"),
params=params,
)
+6 -3
View File
@@ -1,5 +1,6 @@
from ..utils import NamespacedClient, query_params, _make_path
class DeprecationClient(NamespacedClient):
@query_params()
def info(self, index=None, params=None):
@@ -8,6 +9,8 @@ class DeprecationClient(NamespacedClient):
:arg index: Index pattern
"""
return self.transport.perform_request('GET', _make_path(index, '_xpack',
'migration', 'deprecations'), params=params)
return self.transport.perform_request(
"GET",
_make_path(index, "_xpack", "migration", "deprecations"),
params=params,
)
+9 -4
View File
@@ -1,7 +1,8 @@
from ..utils import NamespacedClient, query_params, _make_path
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class GraphClient(NamespacedClient):
@query_params('routing', 'timeout')
@query_params("routing", "timeout")
def explore(self, index=None, doc_type=None, body=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/graph-explore-api.html>`_
@@ -14,5 +15,9 @@ class GraphClient(NamespacedClient):
:arg routing: Specific routing value
:arg timeout: Explicit operation timeout
"""
return self.transport.perform_request('GET', _make_path(index, doc_type,
'_xpack', 'graph', '_explore'), params=params, body=body)
return self.transport.perform_request(
"GET",
_make_path(index, doc_type, "_graph", "explore"),
params=params,
body=body,
)
+104
View File
@@ -0,0 +1,104 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class IlmClient(NamespacedClient):
@query_params()
def delete_lifecycle(self, policy=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-delete-lifecycle.html>`_
:arg policy: The name of the index lifecycle policy
"""
return self.transport.perform_request(
"DELETE", _make_path("_ilm", "policy", policy), params=params
)
@query_params()
def explain_lifecycle(self, index=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-explain-lifecycle.html>`_
:arg index: The name of the index to explain
"""
return self.transport.perform_request(
"GET", _make_path(index, "_ilm", "explain"), params=params
)
@query_params()
def get_lifecycle(self, policy=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-get-lifecycle.html>`_
:arg policy: The name of the index lifecycle policy
"""
return self.transport.perform_request(
"GET", _make_path("_ilm", "policy", policy), params=params
)
@query_params()
def get_status(self, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-get-status.html>`_
"""
return self.transport.perform_request("GET", "/_ilm/status", params=params)
@query_params()
def move_to_step(self, index=None, body=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-move-to-step.html>`_
:arg index: The name of the index whose lifecycle step is to change
:arg body: The new lifecycle step to move to
"""
return self.transport.perform_request(
"POST", _make_path("_ilm", "move", index), params=params, body=body
)
@query_params()
def put_lifecycle(self, policy=None, body=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-put-lifecycle.html>`_
:arg policy: The name of the index lifecycle policy
:arg body: The lifecycle policy definition to register
"""
return self.transport.perform_request(
"PUT", _make_path("_ilm", "policy", policy), params=params, body=body
)
@query_params()
def remove_policy(self, index=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-remove-policy.html>`_
:arg index: The name of the index to remove policy on
"""
return self.transport.perform_request(
"POST", _make_path(index, "_ilm", "remove"), params=params
)
@query_params()
def retry(self, index=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-retry-policy.html>`_
:arg index: The name of the indices (comma-separated) whose failed
lifecycle step is to be retry
"""
return self.transport.perform_request(
"POST", _make_path(index, "_ilm", "retry"), params=params
)
@query_params()
def start(self, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-start.html>`_
"""
return self.transport.perform_request("POST", "/_ilm/start", params=params)
@query_params()
def stop(self, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/ilm-stop.html>`_
"""
return self.transport.perform_request("POST", "/_ilm/stop", params=params)
+67
View File
@@ -0,0 +1,67 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class IndicesClient(NamespacedClient):
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"master_timeout",
"timeout",
"wait_for_active_shards",
)
def freeze(self, index, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/frozen.html>`_
:arg index: The name of the index to freeze
:arg allow_no_indices: Whether to ignore if a wildcard indices
expression resolves into no concrete indices. (This includes `_all`
string or when no indices have been specified)
:arg expand_wildcards: Whether to expand wildcard expression to concrete
indices that are open, closed or both., default 'closed', valid
choices are: 'open', 'closed', 'none', 'all'
:arg ignore_unavailable: Whether specified concrete indices should be
ignored when unavailable (missing or closed)
:arg master_timeout: Specify timeout for connection to master
:arg timeout: Explicit operation timeout
:arg wait_for_active_shards: Sets the number of active shards to wait
for before the operation returns.
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST", _make_path(index, "_freeze"), params=params
)
@query_params(
"allow_no_indices",
"expand_wildcards",
"ignore_unavailable",
"master_timeout",
"timeout",
"wait_for_active_shards",
)
def unfreeze(self, index, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/frozen.html>`_
:arg index: The name of the index to unfreeze
:arg allow_no_indices: Whether to ignore if a wildcard indices
expression resolves into no concrete indices. (This includes `_all`
string or when no indices have been specified)
:arg expand_wildcards: Whether to expand wildcard expression to concrete
indices that are open, closed or both., default 'closed', valid
choices are: 'open', 'closed', 'none', 'all'
:arg ignore_unavailable: Whether specified concrete indices should be
ignored when unavailable (missing or closed)
:arg master_timeout: Specify timeout for connection to master
:arg timeout: Explicit operation timeout
:arg wait_for_active_shards: Sets the number of active shards to wait
for before the operation returns.
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST", _make_path(index, "_unfreeze"), params=params
)
+50 -11
View File
@@ -1,37 +1,76 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class LicenseClient(NamespacedClient):
@query_params()
def delete(self, params=None):
"""
`<https://www.elastic.co/guide/en/x-pack/current/license-management.html>`_
"""
return self.transport.perform_request('DELETE', '/_xpack/license',
params=params)
return self.transport.perform_request("DELETE", "/_license", params=params)
@query_params('local')
@query_params("local")
def get(self, params=None):
"""
`<https://www.elastic.co/guide/en/x-pack/current/license-management.html>`_
:arg local: Return local information, do not retrieve the state from
master node (default: false)
"""
return self.transport.perform_request('GET', '/_xpack/license',
params=params)
return self.transport.perform_request("GET", "/_license", params=params)
@query_params('acknowledge')
@query_params()
def get_basic_status(self, params=None):
"""
`<https://www.elastic.co/guide/en/x-pack/current/license-management.html>`_
"""
return self.transport.perform_request(
"GET", "/_license/basic_status", params=params
)
@query_params()
def get_trial_status(self, params=None):
"""
`<https://www.elastic.co/guide/en/x-pack/current/license-management.html>`_
"""
return self.transport.perform_request(
"GET", "/_license/trial_status", params=params
)
@query_params("acknowledge")
def post(self, body=None, params=None):
"""
`<https://www.elastic.co/guide/en/x-pack/current/license-management.html>`_
:arg body: licenses to be installed
:arg acknowledge: whether the user has acknowledged acknowledge messages
(default: false)
"""
return self.transport.perform_request('PUT', '/_xpack/license',
params=params, body=body)
return self.transport.perform_request(
"PUT", "/_license", params=params, body=body
)
@query_params("acknowledge")
def post_start_basic(self, params=None):
"""
`<https://www.elastic.co/guide/en/x-pack/current/license-management.html>`_
:arg acknowledge: whether the user has acknowledged acknowledge messages
(default: false)
"""
return self.transport.perform_request(
"POST", "/_license/start_basic", params=params
)
@query_params("acknowledge", "doc_type")
def post_start_trial(self, params=None):
"""
`<https://www.elastic.co/guide/en/x-pack/current/license-management.html>`_
:arg acknowledge: whether the user has acknowledged acknowledge messages
(default: false)
:arg doc_type: The type of trial license to generate (default: "trial")
"""
return self.transport.perform_request(
"POST", "/_license/start_trial", params=params
)
+8 -30
View File
@@ -1,36 +1,14 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class MigrationClient(NamespacedClient):
@query_params('allow_no_indices', 'expand_wildcards', 'ignore_unavailable')
def get_assistance(self, index=None, params=None):
@query_params()
def deprecations(self, index=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/migration-api-assistance.html>`_
`<http://www.elastic.co/guide/en/migration/current/migration-api-deprecation.html>`_
:arg index: A comma-separated list of index names; use `_all` or empty
string to perform the operation on all indices
:arg allow_no_indices: Whether to ignore if a wildcard indices
expression resolves into no concrete indices. (This includes `_all`
string or when no indices have been specified)
:arg expand_wildcards: Whether to expand wildcard expression to concrete
indices that are open, closed or both., default 'open', valid
choices are: 'open', 'closed', 'none', 'all'
:arg ignore_unavailable: Whether specified concrete indices should be
ignored when unavailable (missing or closed)
:arg index: Index pattern
"""
return self.transport.perform_request('GET', _make_path('_xpack',
'migration', 'assistance', index), params=params)
@query_params('wait_for_completion')
def upgrade(self, index, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/migration-api-upgrade.html>`_
:arg index: The name of the index
:arg wait_for_completion: Should the request block until the upgrade
operation is completed, default True
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request('POST', _make_path('_xpack',
'migration', 'upgrade', index), params=params)
return self.transport.perform_request(
"GET", _make_path(index, "_migration", "deprecations"), params=params
)
File diff suppressed because it is too large Load Diff
+8 -4
View File
@@ -1,7 +1,8 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class MonitoringClient(NamespacedClient):
@query_params('interval', 'system_api_version', 'system_id')
@query_params("interval", "system_api_version", "system_id")
def bulk(self, body, doc_type=None, params=None):
"""
`<http://www.elastic.co/guide/en/monitoring/current/appendix-api-bulk.html>`_
@@ -16,6 +17,9 @@ class MonitoringClient(NamespacedClient):
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request('POST', _make_path('_xpack',
'monitoring', doc_type, '_bulk'), params=params,
body=self.client._bulk_body(body))
return self.transport.perform_request(
"POST",
_make_path("_monitoring", doc_type, "bulk"),
params=params,
body=self._bulk_body(body),
)
+124
View File
@@ -0,0 +1,124 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class RollupClient(NamespacedClient):
@query_params()
def delete_job(self, id, params=None):
"""
`<>`_
:arg id: The ID of the job to delete
"""
if id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'id'.")
return self.transport.perform_request(
"DELETE", _make_path("_rollup", "job", id), params=params
)
@query_params()
def get_jobs(self, id=None, params=None):
"""
`<>`_
:arg id: The ID of the job(s) to fetch. Accepts glob patterns, or left
blank for all jobs
"""
return self.transport.perform_request(
"GET", _make_path("_rollup", "job", id), params=params
)
@query_params()
def get_rollup_caps(self, id=None, params=None):
"""
`<>`_
:arg id: The ID of the index to check rollup capabilities on, or left
blank for all jobs
"""
return self.transport.perform_request(
"GET", _make_path("_rollup", "data", id), params=params
)
@query_params()
def get_rollup_index_caps(self, index, params=None):
"""
`<>`_
:arg index: The rollup index or index pattern to obtain rollup
capabilities from.
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"GET", _make_path(index, "_rollup", "data"), params=params
)
@query_params()
def put_job(self, id, body, params=None):
"""
`<>`_
:arg id: The ID of the job to create
:arg body: The job configuration
"""
for param in (id, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"PUT", _make_path("_rollup", "job", id), params=params, body=body
)
@query_params("rest_total_hits_as_int", "typed_keys")
def rollup_search(self, index, body, doc_type=None, params=None):
"""
`<>`_
:arg index: The indices or index-pattern(s) (containing rollup or
regular data) that should be searched
:arg body: The search request body
:arg doc_type: The doc type inside the index
:arg rest_total_hits_as_int: Indicates whether hits.total should be
rendered as an integer or an object in the rest search response,
default False
:arg typed_keys: Specify whether aggregation and suggester names should
be prefixed by their respective types in the response
"""
for param in (index, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"GET",
_make_path(index, doc_type, "_rollup_search"),
params=params,
body=body,
)
@query_params()
def start_job(self, id, params=None):
"""
`<>`_
:arg id: The ID of the job to start
"""
if id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'id'.")
return self.transport.perform_request(
"POST", _make_path("_rollup", "job", id, "_start"), params=params
)
@query_params("timeout", "wait_for_completion")
def stop_job(self, id, params=None):
"""
`<>`_
:arg id: The ID of the job to stop
:arg timeout: Block for (at maximum) the specified duration while
waiting for the job to stop. Defaults to 30s.
:arg wait_for_completion: True if the API should block until the job has
fully stopped, false if should be executed async. Defaults to false.
"""
if id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'id'.")
return self.transport.perform_request(
"POST", _make_path("_rollup", "job", id, "_stop"), params=params
)
+280 -135
View File
@@ -1,103 +1,19 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class SecurityClient(NamespacedClient):
@query_params('refresh')
def delete_user(self, username, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-users.html#security-api-delete-user>`_
:arg username: username
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
if username in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'username'.")
return self.transport.perform_request('DELETE', _make_path('_xpack',
'security', 'user', username), params=params)
@query_params()
def get_user(self, username=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-users.html#security-api-get-user>`_
:arg username: A comma-separated list of usernames
"""
return self.transport.perform_request('GET', _make_path('_xpack',
'security', 'user', username), params=params)
@query_params('refresh')
def put_role(self, name, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-roles.html#security-api-put-role>`_
:arg name: Role name
:arg body: The role to add
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
for param in (name, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('PUT', _make_path('_xpack',
'security', 'role', name), params=params, body=body)
@query_params()
def authenticate(self, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-authenticate.html>`_
"""
return self.transport.perform_request('GET',
'/_xpack/security/_authenticate', params=params)
return self.transport.perform_request(
"GET", "/_security/_authenticate", params=params
)
@query_params('refresh')
def put_user(self, username, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-users.html#security-api-put-user>`_
:arg username: The username of the User
:arg body: The user to add
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
for param in (username, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('PUT', _make_path('_xpack',
'security', 'user', username), params=params, body=body)
@query_params('usernames')
def clear_cached_realms(self, realms, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-clear-cache.html>`_
:arg realms: Comma-separated list of realms to clear
:arg usernames: Comma-separated list of usernames to clear from the
cache
"""
if realms in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'realms'.")
return self.transport.perform_request('POST', _make_path('_xpack',
'security', 'realm', realms, '_clear_cache'), params=params)
@query_params('refresh')
@query_params("refresh")
def change_password(self, body, username=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-change-password.html>`_
:arg body: the new password for the user
@@ -110,39 +26,87 @@ class SecurityClient(NamespacedClient):
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request('PUT', _make_path('_xpack',
'security', 'user', username, '_password'), params=params,
body=body)
return self.transport.perform_request(
"PUT",
_make_path("_security", "user", username, "_password"),
params=params,
body=body,
)
@query_params()
def get_role(self, name=None, params=None):
@query_params("usernames")
def clear_cached_realms(self, realms, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-clear-cache.html>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-roles.html#security-api-get-role>`_
:arg name: Role name
:arg realms: Comma-separated list of realms to clear
:arg usernames: Comma-separated list of usernames to clear from the
cache
"""
return self.transport.perform_request('GET', _make_path('_xpack',
'security', 'role', name), params=params)
if realms in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'realms'.")
return self.transport.perform_request(
"POST",
_make_path("_security", "realm", realms, "_clear_cache"),
params=params,
)
@query_params()
def clear_cached_roles(self, name, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-roles.html#security-api-clear-role-cache>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-clear-role-cache.html>`_
:arg name: Role name
"""
if name in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'name'.")
return self.transport.perform_request('POST', _make_path('_xpack',
'security', 'role', name, '_clear_cache'), params=params)
return self.transport.perform_request(
"POST", _make_path("_security", "role", name, "_clear_cache"), params=params
)
@query_params('refresh')
@query_params("refresh")
def create_api_key(self, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html>`_
:arg body: The api key request to create an API key
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request(
"PUT", "/_security/api_key", params=params, body=body
)
@query_params("refresh")
def delete_privileges(self, application, name, params=None):
"""
`<TODO>`_
:arg application: Application name
:arg name: Privilege name
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
for param in (application, name):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"DELETE",
_make_path("_security", "privilege", application, name),
params=params,
)
@query_params("refresh")
def delete_role(self, name, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-roles.html#security-api-delete-role>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-delete-role.html>`_
:arg name: Role name
:arg refresh: If `true` (the default) then refresh the affected shards
@@ -153,13 +117,14 @@ class SecurityClient(NamespacedClient):
"""
if name in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'name'.")
return self.transport.perform_request('DELETE', _make_path('_xpack',
'security', 'role', name), params=params)
return self.transport.perform_request(
"DELETE", _make_path("_security", "role", name), params=params
)
@query_params('refresh')
@query_params("refresh")
def delete_role_mapping(self, name, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-role-mapping.html#security-api-delete-role-mapping>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-delete-role-mapping.html>`_
:arg name: Role-mapping name
:arg refresh: If `true` (the default) then refresh the affected shards
@@ -170,13 +135,32 @@ class SecurityClient(NamespacedClient):
"""
if name in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'name'.")
return self.transport.perform_request('DELETE', _make_path('_xpack',
'security', 'role_mapping', name), params=params)
return self.transport.perform_request(
"DELETE", _make_path("_security", "role_mapping", name), params=params
)
@query_params('refresh')
def disable_user(self, username=None, params=None):
@query_params("refresh")
def delete_user(self, username, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-users.html#security-api-disable-user>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-delete-user.html>`_
:arg username: username
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
if username in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'username'.")
return self.transport.perform_request(
"DELETE", _make_path("_security", "user", username), params=params
)
@query_params("refresh")
def disable_user(self, username, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-disable-user.html>`_
:arg username: The username of the user to disable
:arg refresh: If `true` (the default) then refresh the affected shards
@@ -185,13 +169,16 @@ class SecurityClient(NamespacedClient):
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
return self.transport.perform_request('PUT', _make_path('_xpack',
'security', 'user', username, '_disable'), params=params)
if username in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'username'.")
return self.transport.perform_request(
"PUT", _make_path("_security", "user", username, "_disable"), params=params
)
@query_params('refresh')
def enable_user(self, username=None, params=None):
@query_params("refresh")
def enable_user(self, username, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-users.html#security-api-enable-user>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-enable-user.html>`_
:arg username: The username of the user to enable
:arg refresh: If `true` (the default) then refresh the affected shards
@@ -200,47 +187,182 @@ class SecurityClient(NamespacedClient):
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
return self.transport.perform_request('PUT', _make_path('_xpack',
'security', 'user', username, '_enable'), params=params)
if username in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'username'.")
return self.transport.perform_request(
"PUT", _make_path("_security", "user", username, "_enable"), params=params
)
@query_params("id", "name", "realm_name", "username")
def get_api_key(self, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-get-api-key.html>`_
:arg id: API key id of the API key to be retrieved
:arg name: API key name of the API key to be retrieved
:arg realm_name: realm name of the user who created this API key to be
retrieved
:arg username: user name of the user who created this API key to be
retrieved
"""
return self.transport.perform_request(
"GET", "/_security/api_key", params=params
)
@query_params()
def get_privileges(self, application=None, name=None, params=None):
"""
`<TODO>`_
:arg application: Application name
:arg name: Privilege name
"""
return self.transport.perform_request(
"GET",
_make_path("_security", "privilege", application, name),
params=params,
)
@query_params()
def get_role(self, name=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-get-role.html>`_
:arg name: Role name
"""
return self.transport.perform_request(
"GET", _make_path("_security", "role", name), params=params
)
@query_params()
def get_role_mapping(self, name=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-role-mapping.html#security-api-get-role-mapping>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-get-role-mapping.html>`_
:arg name: Role-Mapping name
"""
return self.transport.perform_request('GET', _make_path('_xpack',
'security', 'role_mapping', name), params=params)
return self.transport.perform_request(
"GET", _make_path("_security", "role_mapping", name), params=params
)
@query_params()
def get_token(self, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-tokens.html#security-api-get-token>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-get-token.html>`_
:arg body: The token request to get
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request('POST',
'/_xpack/security/oauth2/token', params=params, body=body)
return self.transport.perform_request(
"POST", "/_security/oauth2/token", params=params, body=body
)
@query_params()
def get_user(self, username=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-get-user.html>`_
:arg username: A comma-separated list of usernames
"""
return self.transport.perform_request(
"GET", _make_path("_security", "user", username), params=params
)
@query_params()
def get_user_privileges(self, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-get-user-privileges.html>`_
"""
return self.transport.perform_request(
"GET", "/_security/user/_privileges", params=params
)
@query_params()
def has_privileges(self, body, user=None, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-has-privileges.html>`_
:arg body: The privileges to test
:arg user: Username
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request(
"GET",
_make_path("_security", "user", user, "_has_privileges"),
params=params,
body=body,
)
@query_params()
def invalidate_api_key(self, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-invalidate-api-key.html>`_
:arg body: The api key request to invalidate API key(s)
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request(
"DELETE", "/_security/api_key", params=params, body=body
)
@query_params()
def invalidate_token(self, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-tokens.html#security-api-invalidate-token>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-invalidate-token.html>`_
:arg body: The token to invalidate
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request('DELETE',
'/_xpack/security/oauth2/token', params=params, body=body)
return self.transport.perform_request(
"DELETE", "/_security/oauth2/token", params=params, body=body
)
@query_params('refresh')
@query_params("refresh")
def put_privileges(self, body, params=None):
"""
`<TODO>`_
:arg body: The privilege(s) to add
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request(
"PUT", "/_security/privilege/", params=params, body=body
)
@query_params("refresh")
def put_role(self, name, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-put-role.html>`_
:arg name: Role name
:arg body: The role to add
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
for param in (name, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"PUT", _make_path("_security", "role", name), params=params, body=body
)
@query_params("refresh")
def put_role_mapping(self, name, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-role-mapping.html#security-api-put-role-mapping>`_
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-put-role-mapping.html>`_
:arg name: Role-mapping name
:arg body: The role to add
@@ -253,6 +375,29 @@ class SecurityClient(NamespacedClient):
for param in (name, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('PUT', _make_path('_xpack',
'security', 'role_mapping', name), params=params, body=body)
return self.transport.perform_request(
"PUT",
_make_path("_security", "role_mapping", name),
params=params,
body=body,
)
@query_params("refresh")
def put_user(self, username, body, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-put-user.html>`_
:arg username: The username of the User
:arg body: The user to add
:arg refresh: If `true` (the default) then refresh the affected shards
to make this operation visible to search, if `wait_for` then wait
for a refresh to make this operation visible to search, if `false`
then do nothing with refreshes., valid choices are: 'true', 'false',
'wait_for'
"""
for param in (username, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request(
"PUT", _make_path("_security", "user", username), params=params, body=body
)
+43
View File
@@ -0,0 +1,43 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class SqlClient(NamespacedClient):
@query_params()
def clear_cursor(self, body, params=None):
"""
`<Clear SQL cursor>`_
:arg body: Specify the cursor value in the `cursor` element to clean the
cursor.
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request(
"POST", "/_sql/close", params=params, body=body
)
@query_params("format")
def query(self, body, params=None):
"""
`<Execute SQL>`_
:arg body: Use the `query` element to start a query. Use the `cursor`
element to continue a query.
:arg format: a short version of the Accept header, e.g. json, yaml
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request("POST", "/_sql", params=params, body=body)
@query_params()
def translate(self, body, params=None):
"""
`<Translate SQL into Elasticsearch queries>`_
:arg body: Specify the query in the `query` element.
"""
if body in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'body'.")
return self.transport.perform_request(
"POST", "/_sql/translate", params=params, body=body
)
+12
View File
@@ -0,0 +1,12 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class SslClient(NamespacedClient):
@query_params()
def certificates(self, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-ssl.html>`_
"""
return self.transport.perform_request(
"GET", "/_ssl/certificates", params=params
)
+91 -103
View File
@@ -1,148 +1,136 @@
from ..utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
class WatcherClient(NamespacedClient):
@query_params()
def stop(self, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-stop.html>`_
"""
return self.transport.perform_request('POST', '/_xpack/watcher/_stop',
params=params)
@query_params('master_timeout')
def ack_watch(self, watch_id, action_id=None, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-ack-watch.html>`_
:arg watch_id: Watch ID
:arg action_id: A comma-separated list of the action ids to be acked
:arg master_timeout: Explicit operation timeout for connection to master
node
"""
if watch_id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'watch_id'.")
return self.transport.perform_request('PUT', _make_path('_xpack',
'watcher', 'watch', watch_id, '_ack', action_id), params=params)
return self.transport.perform_request(
"PUT",
_make_path("_watcher", "watch", watch_id, "_ack", action_id),
params=params,
)
@query_params('debug')
@query_params()
def activate_watch(self, watch_id, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-activate-watch.html>`_
:arg watch_id: Watch ID
"""
if watch_id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'watch_id'.")
return self.transport.perform_request(
"PUT", _make_path("_watcher", "watch", watch_id, "_activate"), params=params
)
@query_params()
def deactivate_watch(self, watch_id, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-deactivate-watch.html>`_
:arg watch_id: Watch ID
"""
if watch_id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'watch_id'.")
return self.transport.perform_request(
"PUT",
_make_path("_watcher", "watch", watch_id, "_deactivate"),
params=params,
)
@query_params()
def delete_watch(self, id, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-delete-watch.html>`_
:arg id: Watch ID
"""
if id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'id'.")
return self.transport.perform_request(
"DELETE", _make_path("_watcher", "watch", id), params=params
)
@query_params("debug")
def execute_watch(self, id=None, body=None, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-execute-watch.html>`_
:arg id: Watch ID
:arg body: Execution control
:arg debug: indicates whether the watch should execute in debug mode
"""
return self.transport.perform_request('PUT', _make_path('_xpack',
'watcher', 'watch', id, '_execute'), params=params, body=body)
@query_params()
def start(self, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-start.html>`_
"""
return self.transport.perform_request('POST', '/_xpack/watcher/_start',
params=params)
@query_params('master_timeout')
def activate_watch(self, watch_id, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-activate-watch.html>`_
:arg watch_id: Watch ID
:arg master_timeout: Explicit operation timeout for connection to master
node
"""
if watch_id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'watch_id'.")
return self.transport.perform_request('PUT', _make_path('_xpack',
'watcher', 'watch', watch_id, '_activate'), params=params)
@query_params('master_timeout')
def deactivate_watch(self, watch_id, params=None):
"""
`<https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-deactivate-watch.html>`_
:arg watch_id: Watch ID
:arg master_timeout: Explicit operation timeout for connection to master
node
"""
if watch_id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'watch_id'.")
return self.transport.perform_request('PUT', _make_path('_xpack',
'watcher', 'watch', watch_id, '_deactivate'), params=params)
@query_params('active', 'master_timeout')
def put_watch(self, id, body, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-put-watch.html>`_
:arg id: Watch ID
:arg body: The watch
:arg active: Specify whether the watch is in/active by default
:arg master_timeout: Explicit operation timeout for connection to master
node
"""
for param in (id, body):
if param in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument.")
return self.transport.perform_request('PUT', _make_path('_xpack',
'watcher', 'watch', id), params=params, body=body)
@query_params('master_timeout')
def delete_watch(self, id, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-delete-watch.html>`_
:arg id: Watch ID
:arg master_timeout: Explicit operation timeout for connection to master
node
"""
if id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'id'.")
return self.transport.perform_request('DELETE', _make_path('_xpack',
'watcher', 'watch', id), params=params)
return self.transport.perform_request(
"PUT",
_make_path("_watcher", "watch", id, "_execute"),
params=params,
body=body,
)
@query_params()
def get_watch(self, id, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-get-watch.html>`_
:arg id: Watch ID
"""
if id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'id'.")
return self.transport.perform_request('GET', _make_path('_xpack',
'watcher', 'watch', id), params=params)
return self.transport.perform_request(
"GET", _make_path("_watcher", "watch", id), params=params
)
@query_params('emit_stacktraces')
@query_params("active", "if_primary_term", "if_seq_no", "version")
def put_watch(self, id, body=None, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-put-watch.html>`_
:arg id: Watch ID
:arg body: The watch
:arg active: Specify whether the watch is in/active by default
:arg if_primary_term: only update the watch if the last operation that
has changed the watch has the specified primary term
:arg if_seq_no: only update the watch if the last operation that has
changed the watch has the specified sequence number
:arg version: Explicit version number for concurrency control
"""
if id in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'id'.")
return self.transport.perform_request(
"PUT", _make_path("_watcher", "watch", id), params=params, body=body
)
@query_params()
def start(self, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-start.html>`_
"""
return self.transport.perform_request("POST", "/_watcher/_start", params=params)
@query_params("emit_stacktraces")
def stats(self, metric=None, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-stats.html>`_
:arg metric: Controls what additional stat metrics should be include in
the response
:arg emit_stacktraces: Emits stack traces of currently running watches
"""
return self.transport.perform_request('GET', _make_path('_xpack',
'watcher', 'stats', metric), params=params)
return self.transport.perform_request(
"GET", _make_path("_watcher", "stats", metric), params=params
)
@query_params()
def restart(self, params=None):
def stop(self, params=None):
"""
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-restart.html>`_
`<http://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-stop.html>`_
"""
return self.transport.perform_request('POST',
'/_xpack/watcher/_restart', params=params)
return self.transport.perform_request("POST", "/_watcher/_stop", params=params)
+98 -46
View File
@@ -15,12 +15,18 @@ CA_CERTS = None
try:
import certifi
CA_CERTS = certifi.where()
except ImportError:
pass
from .base import Connection
from ..exceptions import ConnectionError, ImproperlyConfigured, ConnectionTimeout, SSLError
from ..exceptions import (
ConnectionError,
ImproperlyConfigured,
ConnectionTimeout,
SSLError,
)
from ..compat import urlencode
@@ -67,17 +73,36 @@ class Urllib3HttpConnection(Connection):
:arg headers: any custom http headers to be add to requests
:arg http_compress: Use gzip compression
"""
def __init__(self, host='localhost', port=9200, http_auth=None,
use_ssl=False, verify_certs=VERIFY_CERTS_DEFAULT, ssl_show_warn=True, ca_certs=None, client_cert=None,
client_key=None, ssl_version=None, ssl_assert_hostname=None,
ssl_assert_fingerprint=None, maxsize=10, headers=None, ssl_context=None, http_compress=False, **kwargs):
super(Urllib3HttpConnection, self).__init__(host=host, port=port, use_ssl=use_ssl, **kwargs)
def __init__(
self,
host="localhost",
port=9200,
http_auth=None,
use_ssl=False,
verify_certs=VERIFY_CERTS_DEFAULT,
ssl_show_warn=True,
ca_certs=None,
client_cert=None,
client_key=None,
ssl_version=None,
ssl_assert_hostname=None,
ssl_assert_fingerprint=None,
maxsize=10,
headers=None,
ssl_context=None,
http_compress=False,
**kwargs
):
super(Urllib3HttpConnection, self).__init__(
host=host, port=port, use_ssl=use_ssl, **kwargs
)
self.http_compress = http_compress
self.headers = urllib3.make_headers(keep_alive=True)
if http_auth is not None:
if isinstance(http_auth, (tuple, list)):
http_auth = ':'.join(http_auth)
http_auth = ":".join(http_auth)
self.headers.update(urllib3.make_headers(basic_auth=http_auth))
# update headers in lowercase to allow overriding of auth headers
@@ -87,32 +112,43 @@ class Urllib3HttpConnection(Connection):
if self.http_compress == True:
self.headers.update(urllib3.make_headers(accept_encoding=True))
self.headers.update({'content-encoding': 'gzip'})
self.headers.update({"content-encoding": "gzip"})
self.headers.setdefault('content-type', 'application/json')
self.headers.setdefault("content-type", "application/json")
pool_class = urllib3.HTTPConnectionPool
kw = {}
# if providing an SSL context, raise error if any other SSL related flag is used
if ssl_context and ( (verify_certs is not VERIFY_CERTS_DEFAULT) or ca_certs
or client_cert or client_key or ssl_version):
warnings.warn("When using `ssl_context`, all other SSL related kwargs are ignored")
if ssl_context and (
(verify_certs is not VERIFY_CERTS_DEFAULT)
or ca_certs
or client_cert
or client_key
or ssl_version
):
warnings.warn(
"When using `ssl_context`, all other SSL related kwargs are ignored"
)
# if ssl_context provided use SSL by default
if ssl_context and self.use_ssl:
pool_class = urllib3.HTTPSConnectionPool
kw.update({
'assert_fingerprint': ssl_assert_fingerprint,
'ssl_context': ssl_context,
})
kw.update(
{
"assert_fingerprint": ssl_assert_fingerprint,
"ssl_context": ssl_context,
}
)
elif self.use_ssl:
pool_class = urllib3.HTTPSConnectionPool
kw.update({
'ssl_version': ssl_version,
'assert_hostname': ssl_assert_hostname,
'assert_fingerprint': ssl_assert_fingerprint,
})
kw.update(
{
"ssl_version": ssl_version,
"assert_hostname": ssl_assert_hostname,
"assert_fingerprint": ssl_assert_fingerprint,
}
)
# If `verify_certs` is sentinal value, default `verify_certs` to `True`
if verify_certs is VERIFY_CERTS_DEFAULT:
@@ -121,43 +157,52 @@ class Urllib3HttpConnection(Connection):
ca_certs = CA_CERTS if ca_certs is None else ca_certs
if verify_certs:
if not ca_certs:
raise ImproperlyConfigured("Root certificates are missing for certificate "
raise ImproperlyConfigured(
"Root certificates are missing for certificate "
"validation. Either pass them in using the ca_certs parameter or "
"install certifi to use it automatically.")
"install certifi to use it automatically."
)
kw.update({
'cert_reqs': 'CERT_REQUIRED',
'ca_certs': ca_certs,
'cert_file': client_cert,
'key_file': client_key,
})
kw.update(
{
"cert_reqs": "CERT_REQUIRED",
"ca_certs": ca_certs,
"cert_file": client_cert,
"key_file": client_key,
}
)
else:
if ssl_show_warn:
warnings.warn(
'Connecting to %s using SSL with verify_certs=False is insecure.' % host)
"Connecting to %s using SSL with verify_certs=False is insecure."
% host
)
self.pool = pool_class(host, port=port, timeout=self.timeout, maxsize=maxsize, **kw)
self.pool = pool_class(
host, port=port, timeout=self.timeout, maxsize=maxsize, **kw
)
def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=(), headers=None):
def perform_request(
self, method, url, params=None, body=None, timeout=None, ignore=(), headers=None
):
url = self.url_prefix + url
if params:
url = '%s?%s' % (url, urlencode(params))
url = "%s?%s" % (url, urlencode(params))
full_url = self.host + url
start = time.time()
try:
kw = {}
if timeout:
kw['timeout'] = timeout
kw["timeout"] = timeout
# in python2 we need to make sure the url and method are not
# unicode. Otherwise the body will be decoded into unicode too and
# that will fail (#133, #201).
if not isinstance(url, str):
url = url.encode('utf-8')
url = url.encode("utf-8")
if not isinstance(method, str):
method = method.encode('utf-8')
method = method.encode("utf-8")
request_headers = self.headers
if headers:
@@ -171,24 +216,31 @@ class Urllib3HttpConnection(Connection):
# again
body = gzip.zlib.compress(body)
response = self.pool.urlopen(method, url, body, retries=Retry(False), headers=request_headers, **kw)
response = self.pool.urlopen(
method, url, body, retries=Retry(False), headers=request_headers, **kw
)
duration = time.time() - start
raw_data = response.data.decode('utf-8')
raw_data = response.data.decode("utf-8")
except Exception as e:
self.log_request_fail(method, full_url, url, body, time.time() - start, exception=e)
self.log_request_fail(
method, full_url, url, body, time.time() - start, exception=e
)
if isinstance(e, UrllibSSLError):
raise SSLError('N/A', str(e), e)
raise SSLError("N/A", str(e), e)
if isinstance(e, ReadTimeoutError):
raise ConnectionTimeout('TIMEOUT', str(e), e)
raise ConnectionError('N/A', str(e), e)
raise ConnectionTimeout("TIMEOUT", str(e), e)
raise ConnectionError("N/A", str(e), e)
# raise errors based on http status codes, let the client handle those if needed
if not (200 <= response.status < 300) and response.status not in ignore:
self.log_request_fail(method, full_url, url, body, duration, response.status, raw_data)
self.log_request_fail(
method, full_url, url, body, duration, response.status, raw_data
)
self._raise_error(response.status, raw_data)
self.log_request_success(method, full_url, url, body, response.status,
raw_data, duration)
self.log_request_success(
method, full_url, url, body, response.status, raw_data, duration
)
return response.status, response.getheaders(), raw_data
+161 -83
View File
@@ -1,5 +1,3 @@
from operator import methodcaller
import time
@@ -11,7 +9,7 @@ from .errors import ScanError, BulkIndexError
import logging
logger = logging.getLogger('elasticsearch.helpers')
logger = logging.getLogger("elasticsearch.helpers")
def expand_action(data):
@@ -26,19 +24,33 @@ def expand_action(data):
# make sure we don't alter the action
data = data.copy()
op_type = data.pop('_op_type', 'index')
op_type = data.pop("_op_type", "index")
action = {op_type: {}}
for key in ('_index', '_parent', '_percolate', '_routing', '_timestamp', 'routing',
'_type', '_version', '_version_type', '_id',
'retry_on_conflict', 'pipeline'):
for key in (
"_index",
"_parent",
"_percolate",
"_routing",
"_timestamp",
"routing",
"_type",
"_version",
"_version_type",
"_id",
"retry_on_conflict",
"pipeline",
):
if key in data:
action[op_type][key] = data.pop(key)
if key == "_routing":
action[op_type]["routing"] = data.pop(key)
else:
action[op_type][key] = data.pop(key)
# no data payload for delete
if op_type == 'delete':
if op_type == "delete":
return action, None
return action, data.get('_source', data)
return action, data.get("_source", data)
def _chunk_actions(actions, chunk_size, max_chunk_bytes, serializer):
@@ -52,14 +64,16 @@ def _chunk_actions(actions, chunk_size, max_chunk_bytes, serializer):
raw_data, raw_action = data, action
action = serializer.dumps(action)
# +1 to account for the trailing new line character
cur_size = len(action.encode('utf-8')) + 1
cur_size = len(action.encode("utf-8")) + 1
if data is not None:
data = serializer.dumps(data)
cur_size += len(data.encode('utf-8')) + 1
cur_size += len(data.encode("utf-8")) + 1
# full chunk, send it and start a new one
if bulk_actions and (size + cur_size > max_chunk_bytes or action_count == chunk_size):
if bulk_actions and (
size + cur_size > max_chunk_bytes or action_count == chunk_size
):
yield bulk_data, bulk_actions
bulk_actions, bulk_data = [], []
size, action_count = 0, 0
@@ -69,7 +83,7 @@ def _chunk_actions(actions, chunk_size, max_chunk_bytes, serializer):
bulk_actions.append(data)
bulk_data.append((raw_action, raw_data))
else:
bulk_data.append((raw_action, ))
bulk_data.append((raw_action,))
size += cur_size
action_count += 1
@@ -78,7 +92,15 @@ def _chunk_actions(actions, chunk_size, max_chunk_bytes, serializer):
yield bulk_data, bulk_actions
def _process_bulk_chunk(client, bulk_actions, bulk_data, raise_on_exception=True, raise_on_error=True, *args, **kwargs):
def _process_bulk_chunk(
client,
bulk_actions,
bulk_data,
raise_on_exception=True,
raise_on_error=True,
*args,
**kwargs
):
"""
Send a bulk request to elasticsearch and process the output.
"""
@@ -87,7 +109,7 @@ def _process_bulk_chunk(client, bulk_actions, bulk_data, raise_on_exception=True
try:
# send the actual request
resp = client.bulk('\n'.join(bulk_actions) + '\n', *args, **kwargs)
resp = client.bulk("\n".join(bulk_actions) + "\n", *args, **kwargs)
except TransportError as e:
# default behavior - just propagate exception
if raise_on_exception:
@@ -101,26 +123,30 @@ def _process_bulk_chunk(client, bulk_actions, bulk_data, raise_on_exception=True
# collect all the information about failed actions
op_type, action = data[0].copy().popitem()
info = {"error": err_message, "status": e.status_code, "exception": e}
if op_type != 'delete':
info['data'] = data[1]
if op_type != "delete":
info["data"] = data[1]
info.update(action)
exc_errors.append({op_type: info})
# emulate standard behavior for failed actions
if raise_on_error:
raise BulkIndexError('%i document(s) failed to index.' % len(exc_errors), exc_errors)
raise BulkIndexError(
"%i document(s) failed to index." % len(exc_errors), exc_errors
)
else:
for err in exc_errors:
yield False, err
return
# go through request-response pairs and detect failures
for data, (op_type, item) in zip(bulk_data, map(methodcaller('popitem'), resp['items'])):
ok = 200 <= item.get('status', 500) < 300
for data, (op_type, item) in zip(
bulk_data, map(methodcaller("popitem"), resp["items"])
):
ok = 200 <= item.get("status", 500) < 300
if not ok and raise_on_error:
# include original document source
if len(data) > 1:
item['data'] = data[1]
item["data"] = data[1]
errors.append({op_type: item})
if ok or not errors:
@@ -129,13 +155,24 @@ def _process_bulk_chunk(client, bulk_actions, bulk_data, raise_on_exception=True
yield ok, {op_type: item}
if errors:
raise BulkIndexError('%i document(s) failed to index.' % len(errors), errors)
raise BulkIndexError("%i document(s) failed to index." % len(errors), errors)
def streaming_bulk(client, actions, chunk_size=500, max_chunk_bytes=100 * 1024 * 1024,
raise_on_error=True, expand_action_callback=expand_action,
raise_on_exception=True, max_retries=0, initial_backoff=2,
max_backoff=600, yield_ok=True, *args, **kwargs):
def streaming_bulk(
client,
actions,
chunk_size=500,
max_chunk_bytes=100 * 1024 * 1024,
raise_on_error=True,
expand_action_callback=expand_action,
raise_on_exception=True,
max_retries=0,
initial_backoff=2,
max_backoff=600,
yield_ok=True,
*args,
**kwargs
):
"""
Streaming bulk consumes actions from the iterable passed in and yields
@@ -171,33 +208,43 @@ def streaming_bulk(client, actions, chunk_size=500, max_chunk_bytes=100 * 1024 *
"""
actions = map(expand_action_callback, actions)
for bulk_data, bulk_actions in _chunk_actions(actions, chunk_size,
max_chunk_bytes,
client.transport.serializer):
for bulk_data, bulk_actions in _chunk_actions(
actions, chunk_size, max_chunk_bytes, client.transport.serializer
):
for attempt in range(max_retries + 1):
to_retry, to_retry_data = [], []
if attempt:
time.sleep(min(max_backoff, initial_backoff * 2**(attempt-1)))
time.sleep(min(max_backoff, initial_backoff * 2 ** (attempt - 1)))
try:
for data, (ok, info) in zip(
bulk_data,
_process_bulk_chunk(client, bulk_actions, bulk_data,
raise_on_exception,
raise_on_error, *args, **kwargs)
):
bulk_data,
_process_bulk_chunk(
client,
bulk_actions,
bulk_data,
raise_on_exception,
raise_on_error,
*args,
**kwargs
),
):
if not ok:
action, info = info.popitem()
# retry if retries enabled, we get 429, and we are not
# in the last attempt
if max_retries \
and info['status'] == 429 \
and (attempt+1) <= max_retries:
if (
max_retries
and info["status"] == 429
and (attempt + 1) <= max_retries
):
# _process_bulk_chunk expects strings so we need to
# re-serialize the data
to_retry.extend(map(client.transport.serializer.dumps, data))
to_retry.extend(
map(client.transport.serializer.dumps, data)
)
to_retry_data.append(data)
else:
yield ok, {action: info}
@@ -249,7 +296,7 @@ def bulk(client, actions, stats_only=False, *args, **kwargs):
errors = []
# make streaming_bulk yield successful results so we can count them
kwargs['yield_ok'] = True
kwargs["yield_ok"] = True
for ok, item in streaming_bulk(client, actions, *args, **kwargs):
# go through request-response pairs and detect failures
if not ok:
@@ -262,9 +309,17 @@ def bulk(client, actions, stats_only=False, *args, **kwargs):
return success, failed if stats_only else errors
def parallel_bulk(client, actions, thread_count=4, chunk_size=500,
max_chunk_bytes=100 * 1024 * 1024, queue_size=4,
expand_action_callback=expand_action, *args, **kwargs):
def parallel_bulk(
client,
actions,
thread_count=4,
chunk_size=500,
max_chunk_bytes=100 * 1024 * 1024,
queue_size=4,
expand_action_callback=expand_action,
*args,
**kwargs
):
"""
Parallel version of the bulk helper run in multiple threads at once.
@@ -299,9 +354,15 @@ def parallel_bulk(client, actions, thread_count=4, chunk_size=500,
try:
for result in pool.imap(
lambda bulk_chunk: list(_process_bulk_chunk(client, bulk_chunk[1], bulk_chunk[0], *args, **kwargs)),
_chunk_actions(actions, chunk_size, max_chunk_bytes, client.transport.serializer)
):
lambda bulk_chunk: list(
_process_bulk_chunk(
client, bulk_chunk[1], bulk_chunk[0], *args, **kwargs
)
),
_chunk_actions(
actions, chunk_size, max_chunk_bytes, client.transport.serializer
),
):
for item in result:
yield item
@@ -310,9 +371,18 @@ def parallel_bulk(client, actions, thread_count=4, chunk_size=500,
pool.join()
def scan(client, query=None, scroll='5m', raise_on_error=True,
preserve_order=False, size=1000, request_timeout=None, clear_scroll=True,
scroll_kwargs=None, **kwargs):
def scan(
client,
query=None,
scroll="5m",
raise_on_error=True,
preserve_order=False,
size=1000,
request_timeout=None,
clear_scroll=True,
scroll_kwargs=None,
**kwargs
):
"""
Simple abstraction on top of the
:meth:`~elasticsearch.Elasticsearch.scroll` api - a simple iterator that
@@ -358,10 +428,11 @@ def scan(client, query=None, scroll='5m', raise_on_error=True,
query = query.copy() if query else {}
query["sort"] = "_doc"
# initial search
resp = client.search(body=query, scroll=scroll, size=size,
request_timeout=request_timeout, **kwargs)
resp = client.search(
body=query, scroll=scroll, size=size, request_timeout=request_timeout, **kwargs
)
scroll_id = resp.get('_scroll_id')
scroll_id = resp.get("_scroll_id")
if scroll_id is None:
return
@@ -372,37 +443,50 @@ def scan(client, query=None, scroll='5m', raise_on_error=True,
if first_run:
first_run = False
else:
resp = client.scroll(scroll_id, scroll=scroll,
request_timeout=request_timeout,
**scroll_kwargs)
resp = client.scroll(
scroll_id,
scroll=scroll,
request_timeout=request_timeout,
**scroll_kwargs
)
for hit in resp['hits']['hits']:
for hit in resp["hits"]["hits"]:
yield hit
# check if we have any errrors
if resp["_shards"]["successful"] < resp["_shards"]["total"]:
logger.warning(
'Scroll request has only succeeded on %d shards out of %d.',
resp['_shards']['successful'], resp['_shards']['total']
"Scroll request has only succeeded on %d shards out of %d.",
resp["_shards"]["successful"],
resp["_shards"]["total"],
)
if raise_on_error:
raise ScanError(
scroll_id,
'Scroll request has only succeeded on %d shards out of %d.' %
(resp['_shards']['successful'], resp['_shards']['total'])
"Scroll request has only succeeded on %d shards out of %d."
% (resp["_shards"]["successful"], resp["_shards"]["total"]),
)
scroll_id = resp.get('_scroll_id')
scroll_id = resp.get("_scroll_id")
# end of scroll
if scroll_id is None or not resp['hits']['hits']:
if scroll_id is None or not resp["hits"]["hits"]:
break
finally:
if scroll_id and clear_scroll:
client.clear_scroll(body={'scroll_id': [scroll_id]}, ignore=(404, ))
client.clear_scroll(body={"scroll_id": [scroll_id]}, ignore=(404,))
def reindex(client, source_index, target_index, query=None, target_client=None,
chunk_size=500, scroll='5m', scan_kwargs={}, bulk_kwargs={}):
def reindex(
client,
source_index,
target_index,
query=None,
target_client=None,
chunk_size=500,
scroll="5m",
scan_kwargs={},
bulk_kwargs={},
):
"""
Reindex all documents from one index that satisfy a given query
@@ -435,26 +519,20 @@ def reindex(client, source_index, target_index, query=None, target_client=None,
:func:`~elasticsearch.helpers.bulk`
"""
target_client = client if target_client is None else target_client
docs = scan(client,
query=query,
index=source_index,
scroll=scroll,
**scan_kwargs
)
docs = scan(client, query=query, index=source_index, scroll=scroll, **scan_kwargs)
def _change_doc_index(hits, index):
for h in hits:
h['_index'] = index
if 'fields' in h:
h.update(h.pop('fields'))
h["_index"] = index
if "fields" in h:
h.update(h.pop("fields"))
yield h
kwargs = {
'stats_only': True,
}
kwargs = {"stats_only": True}
kwargs.update(bulk_kwargs)
return bulk(target_client,
_change_doc_index(docs, target_index),
chunk_size=chunk_size,
**kwargs)
return bulk(
target_client,
_change_doc_index(docs, target_index),
chunk_size=chunk_size,
**kwargs
)
+35 -23
View File
@@ -8,28 +8,31 @@ import subprocess
import nose
def fetch_es_repo():
# user is manually setting YAML dir, don't tamper with it
if 'TEST_ES_YAML_DIR' in environ:
if "TEST_ES_YAML_DIR" in environ:
return
repo_path = environ.get(
'TEST_ES_REPO',
abspath(join(dirname(__file__), pardir, pardir, 'elasticsearch'))
"TEST_ES_REPO",
abspath(join(dirname(__file__), pardir, pardir, "elasticsearch")),
)
# no repo
if not exists(repo_path) or not exists(join(repo_path, '.git')):
print('No elasticsearch repo found...')
if not exists(repo_path) or not exists(join(repo_path, ".git")):
print("No elasticsearch repo found...")
# set YAML DIR to empty to skip yaml tests
environ['TEST_ES_YAML_DIR'] = ''
environ["TEST_ES_YAML_DIR"] = ""
return
# set YAML test dir
environ['TEST_ES_YAML_DIR'] = join(repo_path, 'rest-api-spec', 'src', 'main', 'resources', 'rest-api-spec', 'test')
environ["TEST_ES_YAML_DIR"] = join(
repo_path, "rest-api-spec", "src", "main", "resources", "rest-api-spec", "test"
)
# fetching of yaml tests disabled, we'll run with what's there
if environ.get('TEST_ES_NOFETCH', False):
if environ.get("TEST_ES_NOFETCH", False):
return
from test_elasticsearch.test_server import get_client
@@ -38,19 +41,25 @@ def fetch_es_repo():
# find out the sha of the running es
try:
es = get_client()
sha = es.info()['version']['build_hash']
sha = es.info()["version"]["build_hash"]
except (SkipTest, KeyError):
print('No running elasticsearch >1.X server...')
print("No running elasticsearch >1.X server...")
return
# fetch new commits to be sure...
print('Fetching elasticsearch repo...')
subprocess.check_call('cd %s && git fetch https://github.com/elasticsearch/elasticsearch.git' % repo_path, shell=True)
print("Fetching elasticsearch repo...")
subprocess.check_call(
"cd %s && git fetch https://github.com/elasticsearch/elasticsearch.git"
% repo_path,
shell=True,
)
# reset to the version fron info()
subprocess.check_call('cd %s && git reset --hard %s' % (repo_path, sha), shell=True)
subprocess.check_call("cd %s && git pull" % repo_path, shell=True)
subprocess.check_call("cd %s && git reset --hard %s" % (repo_path, sha), shell=True)
def run_all(argv=None):
sys.exitfunc = lambda: sys.stderr.write('Shutting down....\n')
sys.exitfunc = lambda: sys.stderr.write("Shutting down....\n")
# fetch yaml tests
fetch_es_repo()
@@ -58,16 +67,19 @@ def run_all(argv=None):
# always insert coverage when running tests
if argv is None:
argv = [
'nosetests', '--with-xunit',
'--with-xcoverage', '--cover-package=elasticsearch', '--cover-erase',
'--logging-filter=elasticsearch', '--logging-level=DEBUG',
'--verbose', '--with-id',
"nosetests",
"--with-xunit",
"--with-xcoverage",
"--cover-package=elasticsearch",
"--cover-erase",
"--logging-filter=elasticsearch",
"--logging-level=DEBUG",
"--verbose",
"--with-id",
]
nose.run_exit(
argv=argv,
defaultTest=abspath(dirname(__file__))
)
nose.run_exit(argv=argv, defaultTest=abspath(dirname(__file__)))
if __name__ == '__main__':
if __name__ == "__main__":
run_all(sys.argv)
+3 -1
View File
@@ -2,7 +2,7 @@
# Start elasticsearch in a docker container
ES_VERSION=${ES_VERSION:-"6.1.3"}
ES_VERSION=${ES_VERSION:-"7.0.0-beta1"}
ES_TEST_SERVER=${ES_TEST_SERVER:-"http://localhost:9200"}
SOURCE_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
@@ -11,7 +11,9 @@ exec docker run -d \
-e path.repo=/tmp \
-e "repositories.url.allowed_urls=http://*" \
-e node.attr.testattr=test \
-e node.name=test \
-e ES_HOST=$ES_TEST_SERVER \
-e cluster.initial_master_nodes=test \
-v $SOURCE_DIR/../elasticsearch:/code/elasticsearch \
-v /tmp:/tmp \
-p "9200:9200" \
+36 -36
View File
@@ -4,6 +4,7 @@ from elasticsearch.client import _normalize_hosts, Elasticsearch
from ..test_cases import TestCase, ElasticsearchTestCase
class TestNormalizeHosts(TestCase):
def test_none_uses_defaults(self):
self.assertEquals([{}], _normalize_hosts(None))
@@ -13,87 +14,86 @@ class TestNormalizeHosts(TestCase):
def test_strings_are_parsed_for_port_and_user(self):
self.assertEquals(
[{"host": "elastic.co", "port": 42}, {"host": "elastic.co", "http_auth": "user:secre]"}],
_normalize_hosts(["elastic.co:42", "user:secre%[email protected]"])
[
{"host": "elastic.co", "port": 42},
{"host": "elastic.co", "http_auth": "user:secre]"},
],
_normalize_hosts(["elastic.co:42", "user:secre%[email protected]"]),
)
def test_strings_are_parsed_for_scheme(self):
self.assertEquals(
[
{
"host": "elastic.co",
"port": 42,
"use_ssl": True,
},
{"host": "elastic.co", "port": 42, "use_ssl": True},
{
"host": "elastic.co",
"http_auth": "user:secret",
"use_ssl": True,
"port": 443,
'url_prefix': '/prefix'
}
"url_prefix": "/prefix",
},
],
_normalize_hosts(["https://elastic.co:42", "https://user:[email protected]/prefix"])
_normalize_hosts(
["https://elastic.co:42", "https://user:[email protected]/prefix"]
),
)
def test_dicts_are_left_unchanged(self):
self.assertEquals([{"host": "local", "extra": 123}], _normalize_hosts([{"host": "local", "extra": 123}]))
self.assertEquals(
[{"host": "local", "extra": 123}],
_normalize_hosts([{"host": "local", "extra": 123}]),
)
def test_single_string_is_wrapped_in_list(self):
self.assertEquals(
[{"host": "elastic.co"}],
_normalize_hosts("elastic.co")
)
self.assertEquals([{"host": "elastic.co"}], _normalize_hosts("elastic.co"))
class TestClient(ElasticsearchTestCase):
def test_request_timeout_is_passed_through_unescaped(self):
self.client.ping(request_timeout=.1)
calls = self.assert_url_called('HEAD', '/')
self.assertEquals([({'request_timeout': .1}, None)], calls)
self.client.ping(request_timeout=0.1)
calls = self.assert_url_called("HEAD", "/")
self.assertEquals([({"request_timeout": 0.1}, None)], calls)
def test_params_is_copied_when(self):
rt = object()
params = dict(request_timeout=rt)
self.client.ping(params=params)
self.client.ping(params=params)
calls = self.assert_url_called('HEAD', '/', 2)
calls = self.assert_url_called("HEAD", "/", 2)
self.assertEquals(
[
({'request_timeout': rt}, None),
({'request_timeout': rt}, None)
],
calls
[({"request_timeout": rt}, None), ({"request_timeout": rt}, None)], calls
)
self.assertFalse(calls[0][0] is calls[1][0])
def test_from_in_search(self):
self.client.search(index='i', doc_type='t', from_=10)
calls = self.assert_url_called('GET', '/i/t/_search')
self.assertEquals([({'from': '10'}, None)], calls)
self.client.search(index="i", from_=10)
calls = self.assert_url_called("GET", "/i/_search")
self.assertEquals([({"from": "10"}, None)], calls)
def test_repr_contains_hosts(self):
self.assertEquals('<Elasticsearch([{}])>', repr(self.client))
self.assertEquals("<Elasticsearch([{}])>", repr(self.client))
def test_repr_subclass(self):
class OtherElasticsearch(Elasticsearch): pass
self.assertEqual('<OtherElasticsearch([{}])>', repr(OtherElasticsearch()))
class OtherElasticsearch(Elasticsearch):
pass
self.assertEqual("<OtherElasticsearch([{}])>", repr(OtherElasticsearch()))
def test_repr_contains_hosts_passed_in(self):
self.assertIn("es.org", repr(Elasticsearch(['es.org:123'])))
self.assertIn("es.org", repr(Elasticsearch(["es.org:123"])))
def test_repr_truncates_host_to_5(self):
hosts = [{"host": "es" + str(i)} for i in range(10)]
es = Elasticsearch(hosts)
self.assertNotIn("es5", repr(es))
self.assertIn('...', repr(es))
self.assertIn("...", repr(es))
def test_index_uses_post_if_id_is_empty(self):
self.client.index(index='my-index', doc_type='test-doc', id='', body={})
self.client.index(index="my-index", id="", body={})
self.assert_url_called('POST', '/my-index/test-doc')
self.assert_url_called("POST", "/my-index/_doc")
def test_index_uses_put_if_id_is_not_empty(self):
self.client.index(index='my-index', doc_type='test-doc', id=0, body={})
self.client.index(index="my-index", id=0, body={})
self.assert_url_called('PUT', '/my-index/test-doc/0')
self.assert_url_called("PUT", "/my-index/_doc/0")
+45 -14
View File
@@ -2,7 +2,7 @@
import mock
import time
import threading
from nose.plugins.skip import SkipTest
from elasticsearch import helpers, Elasticsearch
from elasticsearch.serializer import JSONSerializer
@@ -10,6 +10,7 @@ from .test_cases import TestCase
lock_side_effect = threading.Lock()
def mock_process_bulk_chunk(*args, **kwargs):
"""
Threadsafe way of mocking process bulk chunk:
@@ -26,43 +27,73 @@ mock_process_bulk_chunk.call_count = 0
class TestParallelBulk(TestCase):
@mock.patch('elasticsearch.helpers.actions._process_bulk_chunk', side_effect=mock_process_bulk_chunk)
@mock.patch(
"elasticsearch.helpers.actions._process_bulk_chunk",
side_effect=mock_process_bulk_chunk,
)
def test_all_chunks_sent(self, _process_bulk_chunk):
actions = ({'x': i} for i in range(100))
actions = ({"x": i} for i in range(100))
list(helpers.parallel_bulk(Elasticsearch(), actions, chunk_size=2))
self.assertEquals(50, _process_bulk_chunk.call_count)
@SkipTest
@mock.patch(
'elasticsearch.helpers.actions._process_bulk_chunk',
"elasticsearch.helpers.actions._process_bulk_chunk",
# make sure we spend some time in the thread
side_effect=lambda *a: [(True, time.sleep(.001) or threading.current_thread().ident)]
side_effect=lambda *a: [
(True, time.sleep(0.001) or threading.current_thread().ident)
],
)
def test_chunk_sent_from_different_threads(self, _process_bulk_chunk):
actions = ({'x': i} for i in range(100))
results = list(helpers.parallel_bulk(Elasticsearch(), actions, thread_count=10, chunk_size=2))
actions = ({"x": i} for i in range(100))
results = list(
helpers.parallel_bulk(
Elasticsearch(), actions, thread_count=10, chunk_size=2
)
)
self.assertTrue(len(set([r[1] for r in results])) > 1)
class TestChunkActions(TestCase):
def setUp(self):
super(TestChunkActions, self).setUp()
self.actions = [({'index': {}}, {'some': u'datá', 'i': i}) for i in range(100)]
self.actions = [({"index": {}}, {"some": u"datá", "i": i}) for i in range(100)]
def test_chunks_are_chopped_by_byte_size(self):
self.assertEquals(100, len(list(helpers._chunk_actions(self.actions, 100000, 1, JSONSerializer()))))
self.assertEquals(
100,
len(
list(helpers._chunk_actions(self.actions, 100000, 1, JSONSerializer()))
),
)
def test_chunks_are_chopped_by_chunk_size(self):
self.assertEquals(10, len(list(helpers._chunk_actions(self.actions, 10, 99999999, JSONSerializer()))))
self.assertEquals(
10,
len(
list(
helpers._chunk_actions(self.actions, 10, 99999999, JSONSerializer())
)
),
)
def test_chunks_are_chopped_by_byte_size_properly(self):
max_byte_size = 170
chunks = list(helpers._chunk_actions(self.actions, 100000, max_byte_size, JSONSerializer()))
chunks = list(
helpers._chunk_actions(
self.actions, 100000, max_byte_size, JSONSerializer()
)
)
self.assertEquals(25, len(chunks))
for chunk_data, chunk_actions in chunks:
chunk = u''.join(chunk_actions)
chunk = chunk if isinstance(chunk, str) else chunk.encode('utf-8')
chunk = u"".join(chunk_actions)
chunk = chunk if isinstance(chunk, str) else chunk.encode("utf-8")
self.assertLessEqual(len(chunk), max_byte_size)
class TestExpandActions(TestCase):
def test_string_actions_are_marked_as_simple_inserts(self):
self.assertEquals(('{"index":{}}', "whatever"), helpers.expand_action('whatever'))
self.assertEquals(
('{"index":{}}', "whatever"), helpers.expand_action("whatever")
)
+133 -78
View File
@@ -18,59 +18,66 @@ from . import ElasticsearchTestCase
# some params had to be changed in python, keep track of them so we can rename
# those in the tests accordingly
PARAMS_RENAMES = {
'type': 'doc_type',
'from': 'from_',
}
PARAMS_RENAMES = {"type": "doc_type", "from": "from_"}
# mapping from catch values to http status codes
CATCH_CODES = {
'missing': 404,
'conflict': 409,
'unauthorized': 401,
}
CATCH_CODES = {"missing": 404, "conflict": 409, "unauthorized": 401}
# test features we have implemented
IMPLEMENTED_FEATURES = ('gtelte', 'stash_in_path', 'headers', 'catch_unauthorized')
IMPLEMENTED_FEATURES = ("gtelte", "stash_in_path", "headers", "catch_unauthorized")
# broken YAML tests on some releases
SKIP_TESTS = {
# wait for https://github.com/elastic/elasticsearch/issues/25694 to be fixed
# upstream
'*': set(('TestBulk10Basic', 'TestCatSnapshots10Basic',
'TestClusterPutSettings10Basic', 'TestIndex10WithId',
'TestClusterPutScript10Basic', 'TestIndicesPutMapping10Basic',
'TestIndicesPutTemplate10Basic', 'TestMsearch10Basic',
# Skipping some broken integration tests:
'TestIndicesOpen10Basic', 'TestClusterRemoteInfo10Info',
'TestIndicesSplit10Basic', 'TestIndicesSplit20SourceMapping'
))
"*": set(
(
"TestBulk10Basic",
"TestCatSnapshots10Basic",
"TestClusterPutSettings10Basic",
"TestIndex10WithId",
"TestClusterPutScript10Basic",
"TestIndicesPutMapping10Basic",
"TestIndicesPutTemplate10Basic",
"TestMsearch10Basic",
# Skipping some broken integration tests:
"TestIndicesOpen10Basic",
"TestClusterRemoteInfo10Info",
"TestIndicesSplit10Basic",
"TestIndicesSplit20SourceMapping",
)
)
}
class InvalidActionType(Exception):
pass
class YamlTestCase(ElasticsearchTestCase):
def setUp(self):
super(YamlTestCase, self).setUp()
if hasattr(self, '_setup_code'):
if hasattr(self, "_setup_code"):
self.run_code(self._setup_code)
self.last_response = None
self._state = {}
def tearDown(self):
if hasattr(self, '_teardown_code'):
if hasattr(self, "_teardown_code"):
self.run_code(self._teardown_code)
super(YamlTestCase, self).tearDown()
for repo, definition in self.client.snapshot.get_repository(repository='_all').items():
for repo, definition in self.client.snapshot.get_repository(
repository="_all"
).items():
self.client.snapshot.delete_repository(repository=repo)
if definition['type'] == 'fs':
rmtree('/tmp/%s' % definition['settings']['location'], ignore_errors=True)
if definition["type"] == "fs":
rmtree(
"/tmp/%s" % definition["settings"]["location"], ignore_errors=True
)
def _resolve(self, value):
# resolve variables
if isinstance(value, string_types) and value.startswith('$'):
if isinstance(value, string_types) and value.startswith("$"):
value = value[1:]
self.assertIn(value, self._state)
value = self._state[value]
@@ -85,13 +92,13 @@ class YamlTestCase(ElasticsearchTestCase):
def _lookup(self, path):
# fetch the possibly nested value from last_response
value = self.last_response
if path == '$body':
if path == "$body":
return value
path = path.replace(r'\.', '\1')
for step in path.split('.'):
path = path.replace(r"\.", "\1")
for step in path.split("."):
if not step:
continue
step = step.replace('\1', '.')
step = step.replace("\1", ".")
step = self._resolve(step)
if step.isdigit() and step not in value:
step = int(step)
@@ -108,24 +115,24 @@ class YamlTestCase(ElasticsearchTestCase):
self.assertEquals(1, len(action))
action_type, action = list(action.items())[0]
if hasattr(self, 'run_' + action_type):
getattr(self, 'run_' + action_type)(action)
if hasattr(self, "run_" + action_type):
getattr(self, "run_" + action_type)(action)
else:
raise InvalidActionType(action_type)
def run_do(self, action):
""" Perform an api call with given parameters. """
api = self.client
if 'headers' in action:
api = self._get_client(headers=action.pop('headers'))
if "headers" in action:
api = self._get_client(headers=action.pop("headers"))
catch = action.pop('catch', None)
catch = action.pop("catch", None)
self.assertEquals(1, len(action))
method, args = list(action.items())[0]
# locate api endpoint
for m in method.split('.'):
for m in method.split("."):
self.assertTrue(hasattr(api, m))
api = getattr(api, m)
@@ -147,55 +154,76 @@ class YamlTestCase(ElasticsearchTestCase):
self.run_catch(catch, e)
else:
if catch:
raise AssertionError('Failed to catch %r in %r.' % (catch, self.last_response))
raise AssertionError(
"Failed to catch %r in %r." % (catch, self.last_response)
)
def _get_nodes(self):
if not hasattr(self, '_node_info'):
self._node_info = list(self.client.nodes.info(node_id='_all', metric='clear')['nodes'].values())
if not hasattr(self, "_node_info"):
self._node_info = list(
self.client.nodes.info(node_id="_all", metric="clear")["nodes"].values()
)
return self._node_info
def _get_data_nodes(self):
return len([info for info in self._get_nodes() if info.get('attributes', {}).get('data', 'true') == 'true'])
return len(
[
info
for info in self._get_nodes()
if info.get("attributes", {}).get("data", "true") == "true"
]
)
def _get_benchmark_nodes(self):
return len([info for info in self._get_nodes() if info.get('attributes', {}).get('bench', 'false') == 'true'])
return len(
[
info
for info in self._get_nodes()
if info.get("attributes", {}).get("bench", "false") == "true"
]
)
def run_skip(self, skip):
if 'features' in skip:
features = skip['features']
if "features" in skip:
features = skip["features"]
if not isinstance(features, (tuple, list)):
features = [features]
for feature in features:
if feature in IMPLEMENTED_FEATURES:
continue
elif feature == 'requires_replica':
elif feature == "requires_replica":
if self._get_data_nodes() > 1:
continue
elif feature == 'benchmark':
elif feature == "benchmark":
if self._get_benchmark_nodes():
continue
raise SkipTest(skip.get('reason', 'Feature %s is not supported' % feature))
raise SkipTest(
skip.get("reason", "Feature %s is not supported" % feature)
)
if 'version' in skip:
version, reason = skip['version'], skip['reason']
if version == 'all':
if "version" in skip:
version, reason = skip["version"], skip["reason"]
if version == "all":
raise SkipTest(reason)
min_version, max_version = version.split('-')
min_version = _get_version(min_version) or (0, )
max_version = _get_version(max_version) or (999, )
if min_version <= self.es_version <= max_version:
min_version, max_version = version.split("-")
min_version = _get_version(min_version) or (0,)
max_version = _get_version(max_version) or (999,)
if min_version <= self.es_version <= max_version:
raise SkipTest(reason)
def run_catch(self, catch, exception):
if catch == 'param':
if catch == "param":
self.assertIsInstance(exception, TypeError)
return
self.assertIsInstance(exception, TransportError)
if catch in CATCH_CODES:
self.assertEquals(CATCH_CODES[catch], exception.status_code)
elif catch[0] == '/' and catch[-1] == '/':
self.assertTrue(re.search(catch[1:-1], exception.error + ' ' + repr(exception.info)), '%s not in %r' % (catch, exception.info))
elif catch[0] == "/" and catch[-1] == "/":
self.assertTrue(
re.search(catch[1:-1], exception.error + " " + repr(exception.info)),
"%s not in %r" % (catch, exception.info),
)
self.last_response = exception.info
def run_gt(self, action):
@@ -229,11 +257,11 @@ class YamlTestCase(ElasticsearchTestCase):
except AssertionError:
pass
else:
self.assertIn(value, ('', None, False, 0))
self.assertIn(value, ("", None, False, 0))
def run_is_true(self, action):
value = self._lookup(action)
self.assertNotIn(value, ('', None, False, 0))
self.assertNotIn(value, ("", None, False, 0))
def run_length(self, action):
for path, expected in action.items():
@@ -246,8 +274,11 @@ class YamlTestCase(ElasticsearchTestCase):
value = self._lookup(path)
expected = self._resolve(expected)
if isinstance(expected, string_types) and \
expected.startswith('/') and expected.endswith('/'):
if (
isinstance(expected, string_types)
and expected.startswith("/")
and expected.endswith("/")
):
expected = re.compile(expected[1:-1], re.VERBOSE)
self.assertTrue(expected.search(value))
else:
@@ -259,51 +290,75 @@ def construct_case(filename, name):
Parse a definition of a test case from a yaml file and construct the
TestCase subclass dynamically.
"""
def make_test(test_name, definition, i):
def m(self):
if name in SKIP_TESTS.get(self.es_version, ()) or name in SKIP_TESTS.get('*', ()):
if name in SKIP_TESTS.get(self.es_version, ()) or name in SKIP_TESTS.get(
"*", ()
):
raise SkipTest()
self.run_code(definition)
m.__doc__ = '%s:%s.test_from_yaml_%d (%s): %s' % (
__name__, name, i, '/'.join(filename.split('/')[-2:]), test_name)
m.__name__ = 'test_from_yaml_%d' % i
m.__doc__ = "%s:%s.test_from_yaml_%d (%s): %s" % (
__name__,
name,
i,
"/".join(filename.split("/")[-2:]),
test_name,
)
m.__name__ = "test_from_yaml_%d" % i
return m
with open(filename) as f:
tests = list(yaml.load_all(f))
attrs = {
'_yaml_file': filename
}
attrs = {"_yaml_file": filename}
i = 0
for test in tests:
for test_name, definition in test.items():
if test_name in ('setup', 'teardown'):
attrs['_%s_code' % test_name] = definition
if test_name in ("setup", "teardown"):
attrs["_%s_code" % test_name] = definition
continue
attrs['test_from_yaml_%d' % i] = make_test(test_name, definition, i)
attrs["test_from_yaml_%d" % i] = make_test(test_name, definition, i)
i += 1
return type(name, (YamlTestCase, ), attrs)
return type(name, (YamlTestCase,), attrs)
YAML_DIR = environ.get(
'TEST_ES_YAML_DIR',
"TEST_ES_YAML_DIR",
join(
dirname(__file__), pardir, pardir, pardir,
'elasticsearch', 'rest-api-spec', 'src', 'main', 'resources', 'rest-api-spec', 'test'
)
dirname(__file__),
pardir,
pardir,
pardir,
"elasticsearch",
"rest-api-spec",
"src",
"main",
"resources",
"rest-api-spec",
"test",
),
)
if exists(YAML_DIR):
# find all the test definitions in yaml files ...
# find all the test definitions in yaml files ...
for (path, dirs, files) in walk(YAML_DIR):
for filename in files:
if not filename.endswith(('.yaml', '.yml')):
if not filename.endswith((".yaml", ".yml")):
continue
# ... parse them
name = ('Test' + ''.join(s.title() for s in path[len(YAML_DIR) + 1:].split('/')) + filename.rsplit('.', 1)[0].title()).replace('_', '').replace('.', '')
name = (
(
"Test"
+ "".join(s.title() for s in path[len(YAML_DIR) + 1 :].split("/"))
+ filename.rsplit(".", 1)[0].title()
)
.replace("_", "")
.replace(".", "")
)
# and insert them into locals for test runner to find them
locals()[name] = construct_case(join(path, filename), name)
+272 -171
View File
@@ -5,7 +5,9 @@ from ..test_cases import SkipTest
class FailingBulkClient(object):
def __init__(self, client, fail_at=(2, ), fail_with=TransportError(599, "Error!", {})):
def __init__(
self, client, fail_at=(2,), fail_with=TransportError(599, "Error!", {})
):
self.client = client
self._called = 0
self._fail_at = fail_at
@@ -16,34 +18,44 @@ class FailingBulkClient(object):
self._called += 1
if self._called in self._fail_at:
raise self._fail_with
return self.client.bulk(*args, **kwargs)
return self.client.bulk(*args, **kwargs)
class TestStreamingBulk(ElasticsearchTestCase):
def test_actions_remain_unchanged(self):
actions = [{'_id': 1}, {'_id': 2}]
for ok, item in helpers.streaming_bulk(self.client, actions, index='test-index', doc_type='answers'):
actions = [{"_id": 1}, {"_id": 2}]
for ok, item in helpers.streaming_bulk(
self.client, actions, index="test-index"
):
self.assertTrue(ok)
self.assertEquals([{'_id': 1}, {'_id': 2}], actions)
self.assertEquals([{"_id": 1}, {"_id": 2}], actions)
def test_all_documents_get_inserted(self):
docs = [{"answer": x, '_id': x} for x in range(100)]
for ok, item in helpers.streaming_bulk(self.client, docs, index='test-index', doc_type='answers', refresh=True):
docs = [{"answer": x, "_id": x} for x in range(100)]
for ok, item in helpers.streaming_bulk(
self.client, docs, index="test-index", refresh=True
):
self.assertTrue(ok)
self.assertEquals(100, self.client.count(index='test-index', doc_type='answers')['count'])
self.assertEquals({"answer": 42}, self.client.get(index='test-index', doc_type='answers', id=42)['_source'])
self.assertEquals(100, self.client.count(index="test-index")["count"])
self.assertEquals(
{"answer": 42}, self.client.get(index="test-index", id=42)["_source"]
)
def test_all_errors_from_chunk_are_raised_on_failure(self):
self.client.indices.create("i",
self.client.indices.create(
"i",
{
"mappings": {"t": {"properties": {"a": {"type": "integer"}}}},
"settings": {"number_of_shards": 1, "number_of_replicas": 0}
})
"mappings": {"properties": {"a": {"type": "integer"}}},
"settings": {"number_of_shards": 1, "number_of_replicas": 0},
},
)
self.client.cluster.health(wait_for_status="yellow")
try:
for ok, item in helpers.streaming_bulk(self.client, [{"a": "b"},
{"a": "c"}], index="i", doc_type="t", raise_on_error=True):
for ok, item in helpers.streaming_bulk(
self.client, [{"a": "b"}, {"a": "c"}], index="i", raise_on_error=True
):
self.assertTrue(ok)
except helpers.BulkIndexError as e:
self.assertEquals(2, len(e.errors))
@@ -52,102 +64,136 @@ class TestStreamingBulk(ElasticsearchTestCase):
def test_different_op_types(self):
if self.es_version < (0, 90, 1):
raise SkipTest('update supported since 0.90.1')
self.client.index(index='i', doc_type='t', id=45, body={})
self.client.index(index='i', doc_type='t', id=42, body={})
raise SkipTest("update supported since 0.90.1")
self.client.index(index="i", id=45, body={})
self.client.index(index="i", id=42, body={})
docs = [
{'_index': 'i', '_type': 't', '_id': 47, 'f': 'v'},
{'_op_type': 'delete', '_index': 'i', '_type': 't', '_id': 45},
{'_op_type': 'update', '_index': 'i', '_type': 't', '_id': 42, 'doc': {'answer': 42}}
{"_index": "i", "_type": "_doc", "_id": 47, "f": "v"},
{"_op_type": "delete", "_index": "i", "_type": "_doc", "_id": 45},
{
"_op_type": "update",
"_index": "i",
"_type": "_doc",
"_id": 42,
"doc": {"answer": 42},
},
]
for ok, item in helpers.streaming_bulk(self.client, docs):
self.assertTrue(ok)
self.assertFalse(self.client.exists(index='i', doc_type='t', id=45))
self.assertEquals({'answer': 42}, self.client.get(index='i', doc_type='t', id=42)['_source'])
self.assertEquals({'f': 'v'}, self.client.get(index='i', doc_type='t', id=47)['_source'])
self.assertFalse(self.client.exists(index="i", id=45))
self.assertEquals({"answer": 42}, self.client.get(index="i", id=42)["_source"])
self.assertEquals({"f": "v"}, self.client.get(index="i", id=47)["_source"])
def test_transport_error_can_becaught(self):
failing_client = FailingBulkClient(self.client)
docs = [
{'_index': 'i', '_type': 't', '_id': 47, 'f': 'v'},
{'_index': 'i', '_type': 't', '_id': 45, 'f': 'v'},
{'_index': 'i', '_type': 't', '_id': 42, 'f': 'v'},
{"_index": "i", "_type": "_doc", "_id": 47, "f": "v"},
{"_index": "i", "_type": "_doc", "_id": 45, "f": "v"},
{"_index": "i", "_type": "_doc", "_id": 42, "f": "v"},
]
results = list(helpers.streaming_bulk(failing_client, docs, raise_on_exception=False, raise_on_error=False, chunk_size=1))
results = list(
helpers.streaming_bulk(
failing_client,
docs,
raise_on_exception=False,
raise_on_error=False,
chunk_size=1,
)
)
self.assertEquals(3, len(results))
self.assertEquals([True, False, True], [r[0] for r in results])
exc = results[1][1]['index'].pop('exception')
exc = results[1][1]["index"].pop("exception")
self.assertIsInstance(exc, TransportError)
self.assertEquals(599, exc.status_code)
self.assertEquals(
{
'index': {
'_index': 'i',
'_type': 't',
'_id': 45,
'data': {'f': 'v'},
'error': "TransportError(599, 'Error!')",
'status': 599
"index": {
"_index": "i",
"_type": "_doc",
"_id": 45,
"data": {"f": "v"},
"error": "TransportError(599, 'Error!')",
"status": 599,
}
},
results[1][1]
results[1][1],
)
def test_rejected_documents_are_retried(self):
failing_client = FailingBulkClient(self.client, fail_with=TransportError(429, 'Rejected!', {}))
failing_client = FailingBulkClient(
self.client, fail_with=TransportError(429, "Rejected!", {})
)
docs = [
{'_index': 'i', '_type': 't', '_id': 47, 'f': 'v'},
{'_index': 'i', '_type': 't', '_id': 45, 'f': 'v'},
{'_index': 'i', '_type': 't', '_id': 42, 'f': 'v'},
{"_index": "i", "_type": "_doc", "_id": 47, "f": "v"},
{"_index": "i", "_type": "_doc", "_id": 45, "f": "v"},
{"_index": "i", "_type": "_doc", "_id": 42, "f": "v"},
]
results = list(helpers.streaming_bulk(failing_client, docs,
raise_on_exception=False,
raise_on_error=False,
chunk_size=1, max_retries=1,
initial_backoff=0))
results = list(
helpers.streaming_bulk(
failing_client,
docs,
raise_on_exception=False,
raise_on_error=False,
chunk_size=1,
max_retries=1,
initial_backoff=0,
)
)
self.assertEquals(3, len(results))
self.assertEquals([True, True, True], [r[0] for r in results])
self.client.indices.refresh(index='i')
res = self.client.search(index='i')
self.assertEquals(3, res['hits']['total'])
self.client.indices.refresh(index="i")
res = self.client.search(index="i")
self.assertEquals({"value": 3, "relation": "eq"}, res["hits"]["total"])
self.assertEquals(4, failing_client._called)
def test_rejected_documents_are_retried_at_most_max_retries_times(self):
failing_client = FailingBulkClient(self.client, fail_at=(1, 2, ),
fail_with=TransportError(429, 'Rejected!', {}))
failing_client = FailingBulkClient(
self.client, fail_at=(1, 2), fail_with=TransportError(429, "Rejected!", {})
)
docs = [
{'_index': 'i', '_type': 't', '_id': 47, 'f': 'v'},
{'_index': 'i', '_type': 't', '_id': 45, 'f': 'v'},
{'_index': 'i', '_type': 't', '_id': 42, 'f': 'v'},
{"_index": "i", "_type": "_doc", "_id": 47, "f": "v"},
{"_index": "i", "_type": "_doc", "_id": 45, "f": "v"},
{"_index": "i", "_type": "_doc", "_id": 42, "f": "v"},
]
results = list(helpers.streaming_bulk(failing_client, docs,
raise_on_exception=False,
raise_on_error=False,
chunk_size=1, max_retries=1,
initial_backoff=0))
results = list(
helpers.streaming_bulk(
failing_client,
docs,
raise_on_exception=False,
raise_on_error=False,
chunk_size=1,
max_retries=1,
initial_backoff=0,
)
)
self.assertEquals(3, len(results))
self.assertEquals([False, True, True], [r[0] for r in results])
self.client.indices.refresh(index='i')
res = self.client.search(index='i')
self.assertEquals(2, res['hits']['total'])
self.client.indices.refresh(index="i")
res = self.client.search(index="i")
self.assertEquals({"value": 2, "relation": "eq"}, res["hits"]["total"])
self.assertEquals(4, failing_client._called)
def test_transport_error_is_raised_with_max_retries(self):
failing_client = FailingBulkClient(self.client, fail_at=(1, 2, 3, 4, ),
fail_with=TransportError(429, 'Rejected!', {}))
failing_client = FailingBulkClient(
self.client,
fail_at=(1, 2, 3, 4),
fail_with=TransportError(429, "Rejected!", {}),
)
def streaming_bulk():
results = list(helpers.streaming_bulk(
failing_client,
[{"a": 42}, {"a": 39}],
raise_on_exception=True,
max_retries=3,
initial_backoff=0
))
results = list(
helpers.streaming_bulk(
failing_client,
[{"a": 42}, {"a": 39}],
raise_on_exception=True,
max_retries=3,
initial_backoff=0,
)
)
return results
self.assertRaises(TransportError, streaming_bulk)
@@ -156,85 +202,103 @@ class TestStreamingBulk(ElasticsearchTestCase):
class TestBulk(ElasticsearchTestCase):
def test_bulk_works_with_single_item(self):
docs = [{"answer": 42, '_id': 1}]
success, failed = helpers.bulk(self.client, docs, index='test-index', doc_type='answers', refresh=True)
docs = [{"answer": 42, "_id": 1}]
success, failed = helpers.bulk(
self.client, docs, index="test-index", refresh=True
)
self.assertEquals(1, success)
self.assertFalse(failed)
self.assertEquals(1, self.client.count(index='test-index', doc_type='answers')['count'])
self.assertEquals({"answer": 42}, self.client.get(index='test-index', doc_type='answers', id=1)['_source'])
self.assertEquals(1, self.client.count(index="test-index")["count"])
self.assertEquals(
{"answer": 42}, self.client.get(index="test-index", id=1)["_source"]
)
def test_all_documents_get_inserted(self):
docs = [{"answer": x, '_id': x} for x in range(100)]
success, failed = helpers.bulk(self.client, docs, index='test-index', doc_type='answers', refresh=True)
docs = [{"answer": x, "_id": x} for x in range(100)]
success, failed = helpers.bulk(
self.client, docs, index="test-index", refresh=True
)
self.assertEquals(100, success)
self.assertFalse(failed)
self.assertEquals(100, self.client.count(index='test-index', doc_type='answers')['count'])
self.assertEquals({"answer": 42}, self.client.get(index='test-index', doc_type='answers', id=42)['_source'])
self.assertEquals(100, self.client.count(index="test-index")["count"])
self.assertEquals(
{"answer": 42}, self.client.get(index="test-index", id=42)["_source"]
)
def test_stats_only_reports_numbers(self):
docs = [{"answer": x} for x in range(100)]
success, failed = helpers.bulk(self.client, docs, index='test-index', doc_type='answers', refresh=True, stats_only=True)
success, failed = helpers.bulk(
self.client, docs, index="test-index", refresh=True, stats_only=True
)
self.assertEquals(100, success)
self.assertEquals(0, failed)
self.assertEquals(100, self.client.count(index='test-index', doc_type='answers')['count'])
self.assertEquals(100, self.client.count(index="test-index")["count"])
def test_errors_are_reported_correctly(self):
self.client.indices.create("i",
self.client.indices.create(
"i",
{
"mappings": {"t": {"properties": {"a": {"type": "integer"}}}},
"settings": {"number_of_shards": 1, "number_of_replicas": 0}
})
"mappings": {"properties": {"a": {"type": "integer"}}},
"settings": {"number_of_shards": 1, "number_of_replicas": 0},
},
)
self.client.cluster.health(wait_for_status="yellow")
success, failed = helpers.bulk(
self.client,
[{"a": 42}, {"a": "c", '_id': 42}],
[{"a": 42}, {"a": "c", "_id": 42}],
index="i",
doc_type="t",
raise_on_error=False
raise_on_error=False,
)
self.assertEquals(1, success)
self.assertEquals(1, len(failed))
error = failed[0]
self.assertEquals('42', error['index']['_id'])
self.assertEquals('t', error['index']['_type'])
self.assertEquals('i', error['index']['_index'])
print(error['index']['error'])
self.assertTrue('MapperParsingException' in repr(error['index']['error']) or 'mapper_parsing_exception' in repr(error['index']['error']))
self.assertEquals("42", error["index"]["_id"])
self.assertEquals("_doc", error["index"]["_type"])
self.assertEquals("i", error["index"]["_index"])
print(error["index"]["error"])
self.assertTrue(
"MapperParsingException" in repr(error["index"]["error"])
or "mapper_parsing_exception" in repr(error["index"]["error"])
)
def test_error_is_raised(self):
self.client.indices.create("i",
self.client.indices.create(
"i",
{
"mappings": {"t": {"properties": {"a": {"type": "integer"}}}},
"settings": {"number_of_shards": 1, "number_of_replicas": 0}
})
"mappings": {"properties": {"a": {"type": "integer"}}},
"settings": {"number_of_shards": 1, "number_of_replicas": 0},
},
)
self.client.cluster.health(wait_for_status="yellow")
self.assertRaises(helpers.BulkIndexError, helpers.bulk,
self.assertRaises(
helpers.BulkIndexError,
helpers.bulk,
self.client,
[{"a": 42}, {"a": "c"}],
index="i",
doc_type="t"
)
def test_errors_are_collected_properly(self):
self.client.indices.create("i",
self.client.indices.create(
"i",
{
"mappings": {"t": {"properties": {"a": {"type": "integer"}}}},
"settings": {"number_of_shards": 1, "number_of_replicas": 0}
})
"mappings": {"properties": {"a": {"type": "integer"}}},
"settings": {"number_of_shards": 1, "number_of_replicas": 0},
},
)
self.client.cluster.health(wait_for_status="yellow")
success, failed = helpers.bulk(
self.client,
[{"a": 42}, {"a": "c"}],
index="i",
doc_type="t",
stats_only=True,
raise_on_error=False
raise_on_error=False,
)
self.assertEquals(1, success)
self.assertEquals(1, failed)
@@ -244,131 +308,168 @@ class TestScan(ElasticsearchTestCase):
def test_order_can_be_preserved(self):
bulk = []
for x in range(100):
bulk.append({"index": {"_index": "test_index", "_type": "answers", "_id": x}})
bulk.append({"index": {"_index": "test_index", "_type": "_doc", "_id": x}})
bulk.append({"answer": x, "correct": x == 42})
self.client.bulk(bulk, refresh=True)
docs = list(helpers.scan(self.client, index="test_index", doc_type="answers", query={"sort": "answer"}, preserve_order=True))
docs = list(
helpers.scan(
self.client,
index="test_index",
query={"sort": "answer"},
preserve_order=True,
)
)
self.assertEquals(100, len(docs))
self.assertEquals(list(map(str, range(100))), list(d['_id'] for d in docs))
self.assertEquals(list(range(100)), list(d['_source']['answer'] for d in docs))
self.assertEquals(list(map(str, range(100))), list(d["_id"] for d in docs))
self.assertEquals(list(range(100)), list(d["_source"]["answer"] for d in docs))
def test_all_documents_are_read(self):
bulk = []
for x in range(100):
bulk.append({"index": {"_index": "test_index", "_type": "answers", "_id": x}})
bulk.append({"index": {"_index": "test_index", "_type": "_doc", "_id": x}})
bulk.append({"answer": x, "correct": x == 42})
self.client.bulk(bulk, refresh=True)
docs = list(helpers.scan(self.client, index="test_index", doc_type="answers", size=2))
docs = list(helpers.scan(self.client, index="test_index", size=2))
self.assertEquals(100, len(docs))
self.assertEquals(set(map(str, range(100))), set(d['_id'] for d in docs))
self.assertEquals(set(range(100)), set(d['_source']['answer'] for d in docs))
self.assertEquals(set(map(str, range(100))), set(d["_id"] for d in docs))
self.assertEquals(set(range(100)), set(d["_source"]["answer"] for d in docs))
class TestReindex(ElasticsearchTestCase):
def setUp(self):
super(TestReindex, self).setUp()
bulk = []
for x in range(100):
bulk.append({"index": {"_index": "test_index", "_type": "post", "_id": x}})
bulk.append({"answer": x, "correct": x == 42, "type": "answers" if x % 2 == 0 else "questions"})
bulk.append({"index": {"_index": "test_index", "_type": "_doc", "_id": x}})
bulk.append(
{
"answer": x,
"correct": x == 42,
"type": "answers" if x % 2 == 0 else "questions",
}
)
self.client.bulk(bulk, refresh=True)
def test_reindex_passes_kwargs_to_scan_and_bulk(self):
helpers.reindex(self.client, "test_index", "prod_index", scan_kwargs={'q': 'type:answers'}, bulk_kwargs={'refresh': True})
helpers.reindex(
self.client,
"test_index",
"prod_index",
scan_kwargs={"q": "type:answers"},
bulk_kwargs={"refresh": True},
)
self.assertTrue(self.client.indices.exists("prod_index"))
self.assertEquals(50, self.client.count(index='prod_index', q='type:answers')['count'])
self.assertEquals(
50, self.client.count(index="prod_index", q="type:answers")["count"]
)
self.assertEquals({"answer": 42, "correct": True, "type": "answers"}, self.client.get(index="prod_index", doc_type="post", id=42)['_source'])
self.assertEquals(
{"answer": 42, "correct": True, "type": "answers"},
self.client.get(index="prod_index", id=42)["_source"],
)
def test_reindex_accepts_a_query(self):
helpers.reindex(self.client, "test_index", "prod_index", query={"query": {"bool": {"filter": {"term": {"type": "answers"}}}}})
helpers.reindex(
self.client,
"test_index",
"prod_index",
query={"query": {"bool": {"filter": {"term": {"type": "answers"}}}}},
)
self.client.indices.refresh()
self.assertTrue(self.client.indices.exists("prod_index"))
self.assertEquals(50, self.client.count(index='prod_index', q='type:answers')['count'])
self.assertEquals(
50, self.client.count(index="prod_index", q="type:answers")["count"]
)
self.assertEquals({"answer": 42, "correct": True, "type": "answers"}, self.client.get(index="prod_index", doc_type="post", id=42)['_source'])
self.assertEquals(
{"answer": 42, "correct": True, "type": "answers"},
self.client.get(index="prod_index", id=42)["_source"],
)
def test_all_documents_get_moved(self):
helpers.reindex(self.client, "test_index", "prod_index")
self.client.indices.refresh()
self.assertTrue(self.client.indices.exists("prod_index"))
self.assertEquals(50, self.client.count(index='prod_index', q='type:questions')['count'])
self.assertEquals(50, self.client.count(index='prod_index', q='type:answers')['count'])
self.assertEquals(
50, self.client.count(index="prod_index", q="type:questions")["count"]
)
self.assertEquals(
50, self.client.count(index="prod_index", q="type:answers")["count"]
)
self.assertEquals(
{"answer": 42, "correct": True, "type": "answers"},
self.client.get(index="prod_index", id=42)["_source"],
)
self.assertEquals({"answer": 42, "correct": True, "type": "answers"}, self.client.get(index="prod_index", doc_type="post", id=42)['_source'])
class TestParentChildReindex(ElasticsearchTestCase):
def setUp(self):
super(TestParentChildReindex, self).setUp()
body={
'settings': {"number_of_shards": 1, "number_of_replicas": 0},
'mappings': {
'post': {
'properties': {
'question_answer': {
'type': 'join',
'relations': {'question': 'answer'}
}
body = {
"settings": {"number_of_shards": 1, "number_of_replicas": 0},
"mappings": {
"properties": {
"question_answer": {
"type": "join",
"relations": {"question": "answer"},
}
}
}
},
}
self.client.indices.create(index='test-index', body=body)
self.client.indices.create(index='real-index', body=body)
self.client.indices.create(index="test-index", body=body)
self.client.indices.create(index="real-index", body=body)
self.client.index(
index='test-index',
doc_type='post',
id=42,
body={'question_answer': 'question'},
index="test-index", id=42, body={"question_answer": "question"}
)
self.client.index(
index='test-index',
doc_type='post',
index="test-index",
id=47,
routing=42,
body={'some': 'data', 'question_answer': {'name': 'answer', 'parent': 42}},
body={"some": "data", "question_answer": {"name": "answer", "parent": 42}},
)
self.client.indices.refresh(index='test-index')
self.client.indices.refresh(index="test-index")
def test_children_are_reindexed_correctly(self):
helpers.reindex(self.client, 'test-index', 'real-index')
helpers.reindex(self.client, "test-index", "real-index")
q = self.client.get(
index='real-index',
doc_type='post',
id=42
)
q = self.client.get(index="real-index", id=42)
self.assertEquals(
{
'_id': '42',
'_index': 'real-index',
'_source': {'question_answer': 'question'},
'_type': 'post',
'_version': 1,
'found': True
}, q
)
q = self.client.get(
index='test-index',
doc_type='post',
id=47,
routing=42
"_id": "42",
"_index": "real-index",
"_primary_term": 1,
"_seq_no": 0,
"_source": {"question_answer": "question"},
"_type": "_doc",
"_version": 1,
"found": True,
},
q,
)
q = self.client.get(index="test-index", id=47, routing=42)
self.assertEquals(
{
'_routing': '42',
'_id': '47',
'_index': 'test-index',
'_source': {'some': 'data', 'question_answer': {'name': 'answer', 'parent': 42}},
'_type': 'post',
'_version': 1,
'found': True
}, q
"_routing": "42",
"_id": "47",
"_index": "test-index",
"_primary_term": 1,
"_seq_no": 1,
"_source": {
"some": "data",
"question_answer": {"name": "answer", "parent": 42},
},
"_type": "_doc",
"_version": 1,
"found": True,
},
q,
)