From 672fdf7d5f943d24ed29e06e1b95e5fc431653ec Mon Sep 17 00:00:00 2001 From: Honza Kral Date: Wed, 25 Sep 2013 23:09:50 +0200 Subject: [PATCH] Make ignore a global parameter --- elasticsearch/client/__init__.py | 50 ++++++--------------------- elasticsearch/client/utils.py | 4 +++ elasticsearch/connection/http.py | 8 ++--- elasticsearch/connection/memcached.py | 4 +-- elasticsearch/connection/thrift.py | 4 +-- elasticsearch/transport.py | 8 ++++- 6 files changed, 30 insertions(+), 48 deletions(-) diff --git a/elasticsearch/client/__init__.py b/elasticsearch/client/__init__.py index 462c9608..752ab048 100644 --- a/elasticsearch/client/__init__.py +++ b/elasticsearch/client/__init__.py @@ -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. ``_ @@ -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. ``_ @@ -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. ``_ @@ -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. ``_ @@ -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') diff --git a/elasticsearch/client/utils.py b/elasticsearch/client/utils.py index c8394e0d..722f9821 100644 --- a/elasticsearch/client/utils.py +++ b/elasticsearch/client/utils.py @@ -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 diff --git a/elasticsearch/connection/http.py b/elasticsearch/connection/http.py index 24386849..4b20d4b2 100644 --- a/elasticsearch/connection/http.py +++ b/elasticsearch/connection/http.py @@ -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) diff --git a/elasticsearch/connection/memcached.py b/elasticsearch/connection/memcached.py index 47636d34..3504e649 100644 --- a/elasticsearch/connection/memcached.py +++ b/elasticsearch/connection/memcached.py @@ -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) diff --git a/elasticsearch/connection/thrift.py b/elasticsearch/connection/thrift.py index 85e64312..f695a80d 100644 --- a/elasticsearch/connection/thrift.py +++ b/elasticsearch/connection/thrift.py @@ -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) diff --git a/elasticsearch/transport.py b/elasticsearch/transport.py index b6f5adef..e87e4cc5 100644 --- a/elasticsearch/transport.py +++ b/elasticsearch/transport.py @@ -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)