Make ignore a global parameter
This commit is contained in:
@@ -171,7 +171,7 @@ class Elasticsearch(object):
|
||||
|
||||
@query_params('_source', '_source_exclude', '_source_include', 'fields',
|
||||
'parent', 'preference', 'realtime', 'refresh', 'routing')
|
||||
def get(self, index, id, doc_type='_all', ignore=(), params=None):
|
||||
def get(self, index, id, doc_type='_all', params=None):
|
||||
"""
|
||||
Get a typed JSON document from the index based on its id.
|
||||
`<http://elasticsearch.org/guide/reference/api/get/>`_
|
||||
@@ -180,8 +180,6 @@ class Elasticsearch(object):
|
||||
:arg id: The document ID
|
||||
:arg doc_type: The type of the document (uses `_all` by default to
|
||||
fetch the first document matching the ID across all types)
|
||||
:arg ignore: a list of http status number that should be ignored
|
||||
instead of raised as exceptions
|
||||
:arg _source: True or false to return the _source field or not, or a
|
||||
list of fields to return
|
||||
:arg _source_exclude: A list of fields to exclude from the returned
|
||||
@@ -198,18 +196,13 @@ class Elasticsearch(object):
|
||||
performing the operation
|
||||
:arg routing: Specific routing value
|
||||
"""
|
||||
try:
|
||||
_, data = self.transport.perform_request('GET', _make_path(index, doc_type, id),
|
||||
params=params)
|
||||
except TransportError as e:
|
||||
if e.status_code == ignore:
|
||||
return
|
||||
raise
|
||||
_, data = self.transport.perform_request('GET', _make_path(index, doc_type, id),
|
||||
params=params)
|
||||
return data
|
||||
|
||||
@query_params('_source_exclude', '_source_include', 'parent', 'preference',
|
||||
'realtime', 'refresh', 'routing')
|
||||
def get_source(self, index, id, doc_type='_all', ignore=(), params=None):
|
||||
def get_source(self, index, id, doc_type='_all', params=None):
|
||||
"""
|
||||
Get the source of a document by it's index, type and id.
|
||||
`<http://elasticsearch.org/guide/reference/api/get/>`_
|
||||
@@ -218,8 +211,6 @@ class Elasticsearch(object):
|
||||
:arg doc_type: The type of the document (uses `_all` by default to
|
||||
fetch the first document matching the ID across all types)
|
||||
:arg id: The document ID
|
||||
:arg ignore: a list of http status number that should be ignored
|
||||
instead of raised as exceptions
|
||||
:arg _source_exclude: A list of fields to exclude from the returned
|
||||
_source field
|
||||
:arg _source_include: A list of fields to extract and return from the
|
||||
@@ -232,13 +223,8 @@ class Elasticsearch(object):
|
||||
performing the operation
|
||||
:arg routing: Specific routing value
|
||||
"""
|
||||
try:
|
||||
_, data = self.transport.perform_request('GET', _make_path(index, doc_type, id, '_source'),
|
||||
params=params)
|
||||
except TransportError as e:
|
||||
if e.status_code == ignore:
|
||||
return
|
||||
raise
|
||||
_, data = self.transport.perform_request('GET', _make_path(index, doc_type, id, '_source'),
|
||||
params=params)
|
||||
return data
|
||||
|
||||
@query_params('_source', '_source_exclude', '_source_include', 'fields',
|
||||
@@ -274,7 +260,7 @@ class Elasticsearch(object):
|
||||
@query_params('consistency', 'fields', 'lang', 'parent', 'percolate',
|
||||
'refresh', 'replication', 'retry_on_conflict', 'routing', 'script',
|
||||
'timeout', 'timestamp', 'ttl', 'version', 'version_type')
|
||||
def update(self, index, doc_type, id, body=None, ignore=(), params=None):
|
||||
def update(self, index, doc_type, id, body=None, params=None):
|
||||
"""
|
||||
Update a document based on a script or partial data provided.
|
||||
`<http://elasticsearch.org/guide/reference/api/update/>`_
|
||||
@@ -283,8 +269,6 @@ class Elasticsearch(object):
|
||||
:arg doc_type: The type of the document
|
||||
:arg id: Document ID
|
||||
:arg body: The request definition using either `script` or partial `doc`
|
||||
:arg ignore: a list of http status number that should be ignored
|
||||
instead of raised as exceptions
|
||||
:arg consistency: Explicit write consistency setting for the operation
|
||||
:arg fields: A comma-separated list of fields to return in the response
|
||||
:arg lang: The script language (default: mvel)
|
||||
@@ -303,13 +287,8 @@ class Elasticsearch(object):
|
||||
:arg version: Explicit version number for concurrency control
|
||||
:arg version_type: Explicit version number for concurrency control
|
||||
"""
|
||||
try:
|
||||
_, data = self.transport.perform_request('POST', _make_path(index, doc_type, id, '_update'),
|
||||
params=params, body=body)
|
||||
except TransportError as e:
|
||||
if e.status_code == ignore:
|
||||
return
|
||||
raise
|
||||
_, data = self.transport.perform_request('POST', _make_path(index, doc_type, id, '_update'),
|
||||
params=params, body=body)
|
||||
return data
|
||||
|
||||
@query_params('_source', '_source_exclude', '_source_include',
|
||||
@@ -450,7 +429,7 @@ class Elasticsearch(object):
|
||||
|
||||
@query_params('consistency', 'parent', 'refresh', 'replication', 'routing',
|
||||
'timeout', 'version', 'version_type')
|
||||
def delete(self, index, doc_type, id, ignore=(), params=None):
|
||||
def delete(self, index, doc_type, id, params=None):
|
||||
"""
|
||||
Delete a typed JSON document from a specific index based on its id.
|
||||
`<http://elasticsearch.org/guide/reference/api/delete/>`_
|
||||
@@ -458,8 +437,6 @@ class Elasticsearch(object):
|
||||
:arg index: The name of the index
|
||||
:arg doc_type: The type of the document
|
||||
:arg id: The document ID
|
||||
:arg ignore: a list of http status number that should be ignored
|
||||
instead of raised as exceptions
|
||||
:arg consistency: Specific write consistency setting for the operation
|
||||
:arg parent: ID of parent document
|
||||
:arg refresh: Refresh the index after performing the operation
|
||||
@@ -469,12 +446,7 @@ class Elasticsearch(object):
|
||||
:arg version: Explicit version number for concurrency control
|
||||
:arg version_type: Specific version type
|
||||
"""
|
||||
try:
|
||||
_, data = self.transport.perform_request('DELETE', _make_path(index, doc_type, id), params=params)
|
||||
except TransportError as e:
|
||||
if e.status_code == ignore:
|
||||
return
|
||||
raise
|
||||
_, data = self.transport.perform_request('DELETE', _make_path(index, doc_type, id), params=params)
|
||||
return data
|
||||
|
||||
@query_params('ignore_indices', 'min_score', 'preference', 'routing', 'source')
|
||||
|
||||
@@ -62,6 +62,10 @@ def query_params(*es_query_params):
|
||||
for p in es_query_params + GLOBAL_PARAMS:
|
||||
if p in kwargs:
|
||||
params[p] = _escape(kwargs.pop(p))
|
||||
|
||||
# don't treat ignore as other params to avoid escaping
|
||||
if 'ignore' in kwargs:
|
||||
params['ignore'] = kwargs.pop('ignore')
|
||||
return func(*args, params=params, **kwargs)
|
||||
return _wrapped
|
||||
return _wrapper
|
||||
|
||||
@@ -30,7 +30,7 @@ class RequestsHttpConnection(Connection):
|
||||
)
|
||||
|
||||
|
||||
def perform_request(self, method, url, params=None, body=None, timeout=None):
|
||||
def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()):
|
||||
url = self.base_url + url
|
||||
|
||||
# use prepared requests so that requests formats url and params for us to log
|
||||
@@ -45,7 +45,7 @@ class RequestsHttpConnection(Connection):
|
||||
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_code < 300):
|
||||
if not (200 <= response.status_code < 300) and response.status_code not in ignore:
|
||||
self.log_request_fail(method, request.url, duration, response.status_code)
|
||||
self._raise_error(response.status_code, raw_data)
|
||||
|
||||
@@ -75,7 +75,7 @@ class Urllib3HttpConnection(Connection):
|
||||
|
||||
self.pool = pool_class(host, port=port, timeout=kwargs.get('timeout', None), headers=headers)
|
||||
|
||||
def perform_request(self, method, url, params=None, body=None, timeout=None):
|
||||
def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()):
|
||||
url = self.url_prefix + url
|
||||
if params:
|
||||
url = '%s?%s' % (url, urlencode(params or {}))
|
||||
@@ -93,7 +93,7 @@ class Urllib3HttpConnection(Connection):
|
||||
self.log_request_fail(method, full_url, time.time() - start, exception=e)
|
||||
raise ConnectionError('N/A', str(e), e)
|
||||
|
||||
if not (200 <= response.status < 300):
|
||||
if not (200 <= response.status < 300) and response.status not in ignore:
|
||||
self.log_request_fail(method, url, duration, response.status)
|
||||
self._raise_error(response.status, raw_data)
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ class MemcachedConnection(PoolingConnection):
|
||||
super(MemcachedConnection, self).__init__(host=host, port=port, **kwargs)
|
||||
self._make_connection = lambda: pylibmc.Client(['%s:%s' % (host, port)], behaviors={"tcp_nodelay": True})
|
||||
|
||||
def perform_request(self, method, url, params=None, body=None, timeout=None):
|
||||
def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()):
|
||||
mc = self._get_connection()
|
||||
url = self.url_prefix + url
|
||||
if params:
|
||||
@@ -70,7 +70,7 @@ class MemcachedConnection(PoolingConnection):
|
||||
elif 'error' in data:
|
||||
raise TransportError('N/A', data['error'])
|
||||
|
||||
if not (200 <= status < 300):
|
||||
if not (200 <= status < 300) and status not in ignore:
|
||||
self.log_request_fail(method, url, duration, status)
|
||||
self._raise_error(status, response)
|
||||
|
||||
|
||||
@@ -51,7 +51,7 @@ class ThriftConnection(PoolingConnection):
|
||||
transport.open()
|
||||
return client
|
||||
|
||||
def perform_request(self, method, url, params=None, body=None, timeout=None):
|
||||
def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()):
|
||||
request = RestRequest(method=Method._NAMES_TO_VALUES[method.upper()], uri=url,
|
||||
parameters=params, body=body)
|
||||
|
||||
@@ -66,7 +66,7 @@ class ThriftConnection(PoolingConnection):
|
||||
finally:
|
||||
self._release_connection(tclient)
|
||||
|
||||
if not (200 <= response.status < 300):
|
||||
if not (200 <= response.status < 300) and response.status not in ignore:
|
||||
self.log_request_fail(method, url, duration, response.status)
|
||||
self._raise_error(response.status, response.body)
|
||||
|
||||
|
||||
@@ -210,11 +210,17 @@ class Transport(object):
|
||||
if body is not None:
|
||||
body = self.serializer.dumps(body)
|
||||
|
||||
ignore = ()
|
||||
if params and 'ignore' in params:
|
||||
ignore = params.pop('ignore')
|
||||
if isinstance(ignore, int):
|
||||
ignore = (ignore, )
|
||||
|
||||
for attempt in range(self.max_retries + 1):
|
||||
connection = self.get_connection()
|
||||
|
||||
try:
|
||||
status, raw_data = connection.perform_request(method, url, params, body)
|
||||
status, raw_data = connection.perform_request(method, url, params, body, ignore=ignore)
|
||||
except ConnectionError:
|
||||
self.mark_dead(connection)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user