Python elasticsearch.exceptions 模块，ConnectionTimeout() 实例源码

def index_worker(self, queue, size=200):
        actions = []
        indexed = 0
        while True:
            item = queue.get()
            if item is None:
                break
            id_submission, analysis = item

            doc = {
                '_index': 'fcc-comments',
                '_type': 'document',
                '_op_type': 'update',
                '_id': id_submission,
                'doc': {'analysis': analysis},
            }
            actions.append(doc)

            if len(actions) == size:
                with warnings.catch_warnings():
                    warnings.simplefilter('ignore')
                    try:
                        response = bulk(self.es, actions)
                        indexed += response[0]
                        print('\tanalyzed %s/%s\t%s%%' % (indexed, self.limit,
                            int(indexed / self.limit * 100)))
                        actions = []
                    except ConnectionTimeout:
                        print('error indexing: connection timeout')

        with warnings.catch_warnings():
            warnings.simplefilter('ignore')
            response = bulk(self.es, actions)
            indexed += response[0]
            print('indexed %s' % (indexed))

def run(self):
        '''
            get documents without a sentiment tag that match significant terms:
            - significant terms from postive regex tagged vs others
            - extra multi match clause for stronger terms (in multiple term sets:
                positive vs negative, untagged, and all
            - phrase match net neutrality since both terms score high
        '''

        index_queue = multiprocessing.Queue()

        bulk_index_process = multiprocessing.Process(
            target=self.bulk_index, args=(index_queue,),
        )
        bulk_index_process.start()
        fetched = 0
        try:
            while fetched < self.limit:
                '''
                    use search instead of scan because keeping an ordered scan cursor
                    open negates the performance benefits
                '''
                resp = self.es.search(index='fcc-comments', body=self.query, size=self.limit)
                for doc in resp['hits']['hits']:
                    index_queue.put(doc['_id'])
                    fetched += 1
                    if not fetched % 100:
                        print('%s\t%s\t%s' % (fetched, doc['_score'],
                            doc['_source']['text_data']))
        except ConnectionTimeout:
            print('error fetching: connection timeout')

        index_queue.put(None)
        bulk_index_process.join()

def bulk_index(self, queue, size=20):

        actions = []
        indexed = 0
        ids = set()
        while True:
            item = queue.get()
            if item is None:
                break
            doc_id = item

            doc = {
                '_index': 'fcc-comments',
                '_type': 'document',
                '_op_type': 'update',
                '_id': doc_id,
                'doc': {'analysis.sentiment_sig_terms_ordered': True},
            }
            actions.append(doc)
            ids.add(doc_id)

            if len(actions) == size:
                with warnings.catch_warnings():
                    warnings.simplefilter('ignore')
                    try:
                        response = bulk(self.es, actions)
                        indexed += response[0]
                        if not indexed % 200:
                            print('\tindexed %s/%s\t%s%%' % (indexed, self.limit,
                                int(indexed / self.limit * 100)))
                        actions = []
                    except ConnectionTimeout:
                        print('error indexing: connection timeout')

        with warnings.catch_warnings():
            warnings.simplefilter('ignore')
            response = bulk(self.es, actions)
            indexed += response[0]
            print('indexed %s' % (indexed))
        ids = list(ids)
        #print('%s\n%s' % (len(ids), ' '.join(ids))

def create(self, *args, **kwargs):
        """
        The create method allows for documents passed to be elasticsearch handler
        directly.
        """
        # date = kwargs.get("date", None)
        # parser = kwargs.get('parser', 'default')
        # doctype = kwargs.get('type', 'default')
        # document = kwargs.get('document', {})
        # 
        # if date:
        #     # Index document into an index based on index_date field
        #     options = {
        #         "index": "dminer-%s-%s" % (parser, date),
        #         "doc_type": doctype,
        #         "body": document
        #     }
        # else:
        #     options = {
        #         "index": "dminer-%s" % parser,
        #         "doc_type": doctype,
        #         "body": document
        #     }

        while True:
            try:
                value = self.es.index(*args, **kwargs)
            except ConnectionTimeout:
                print "Connection Timeout"
                continue
            break
        return value

def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=(), headers=None):
        url_path = url
        if params:
            url_path = '%s?%s' % (url, urlencode(params or {}))
        url = self.base_url + url_path

        start = self.loop.time()
        response = None
        try:
            with aiohttp.Timeout(timeout or self.timeout, loop=self.loop):
                response = yield from self.session.request(method, url, data=body, headers=headers)
                raw_data = yield from response.text()
            duration = self.loop.time() - start

        except Exception as e:
            self.log_request_fail(method, url, url_path, body, self.loop.time() - start, exception=e)
            if isinstance(e, ServerFingerprintMismatch):
                raise SSLError('N/A', str(e), e)
            if isinstance(e, asyncio.TimeoutError):
                raise ConnectionTimeout('TIMEOUT', str(e), e)
            raise ConnectionError('N/A', str(e), e)

        finally:
            if response is not None:
                yield from response.release()

        # 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, url, url_path, body, duration, status_code=response.status, response=raw_data)
            self._raise_error(response.status, raw_data)

        self.log_request_success(method, url, url_path, body, response.status, raw_data, duration)

        return response.status, response.headers, raw_data

def lookup_portal_item(esclient, index_name, item_id):
    """
    Returns an array of the collections that this item belongs to, according to Portal's ES index.
    :param esclient: Elastic search client object to use
    :param index_name: Name of the index to search within
    :param item_id: item ID to look for
    :return: array of collection names that this belongs to. Blank array if it does not belong.
    """
    parts = id_xplodr.match(item_id)
    query = {
        'query': {
            'filtered': {
                'filter': {
                    'term': {
                        'vidispine_id_str_ex': item_id
                    }
                }
            }
        }
    }

    wait_time = 2
    while True:
        try:
            result = esclient.search(index=index_name, doc_type='item', body=query)
            break
        except ReadTimeoutError as e:
            logger.warning(str(e))
            sleep(wait_time)
            wait_time *= 2
        except ConnectionTimeout as e:
            logger.warning(str(e))
            sleep(wait_time)
            wait_time *= 2
    hits = result['hits']['hits']
    if len(hits) == 0: raise PortalItemNotFound(item_id)

    # pprint(hits[0]['_source'])
    if not 'f___collection_str' in hits[0]['_source']:
        return None
    return hits[0]['_source']['f___collection_str']  # this is an array

def collect(self):
        try:
            response = self.es_client.cluster.health(level=self.level, request_timeout=self.timeout)

            metrics = cluster_health_parser.parse_response(response, self.metric_name_list)
        except ConnectionTimeout:
            logging.warn('Timeout while fetching %s (timeout %ss).', self.description, self.timeout)
            yield collector_up_gauge(self.metric_name_list, self.description, succeeded=False)
        except Exception:
            logging.exception('Error while fetching %s.', self.description)
            yield collector_up_gauge(self.metric_name_list, self.description, succeeded=False)
        else:
            yield from gauge_generator(metrics)
            yield collector_up_gauge(self.metric_name_list, self.description)

def collect(self):
        try:
            response = self.es_client.nodes.stats(metric=self.metrics, request_timeout=self.timeout)

            metrics = nodes_stats_parser.parse_response(response, self.metric_name_list)
        except ConnectionTimeout:
            logging.warn('Timeout while fetching %s (timeout %ss).', self.description, self.timeout)
            yield collector_up_gauge(self.metric_name_list, self.description, succeeded=False)
        except Exception:
            logging.exception('Error while fetching %s.', self.description)
            yield collector_up_gauge(self.metric_name_list, self.description, succeeded=False)
        else:
            yield from gauge_generator(metrics)
            yield collector_up_gauge(self.metric_name_list, self.description)

def collect(self):
        try:
            response = self.es_client.indices.stats(metric=self.metrics, fields=self.fields, request_timeout=self.timeout)

            metrics = indices_stats_parser.parse_response(response, self.parse_indices, self.metric_name_list)
        except ConnectionTimeout:
            logging.warn('Timeout while fetching %s (timeout %ss).', self.description, self.timeout)
            yield collector_up_gauge(self.metric_name_list, self.description, succeeded=False)
        except Exception:
            logging.exception('Error while fetching %s.', self.description)
            yield collector_up_gauge(self.metric_name_list, self.description, succeeded=False)
        else:
            yield from gauge_generator(metrics)
            yield collector_up_gauge(self.metric_name_list, self.description)

def test_failed_es_request(self):
        request = mock_request()
        exc = es_exceptions.ConnectionTimeout()
        request.es.cluster.health.side_effect = exc

        with pytest.raises(views.FailedHealthcheck) as e:
            views.healthcheck(request)

        assert e.value.__cause__ == exc

def exists_es_index(str_valid_index):
    """Returns if given index exists in Elasticsearch cluster"""
    connection_attempts = 0
    while connection_attempts < 3:
        try:
            es = get_es_object()
            es_indices = es.indices
            return es_indices.exists(index=str_valid_index)
        except exceptions.ConnectionTimeout:
            connection_attempts += 1
    sys.exit('Elasticsearch connection timeout, exiting now...')

def perform_request(
            self, method, url, params=None, body=None,
            timeout=None, ignore=()):
        url = self.url_prefix + url
        if params:
            url = '{}?{}'.format(url, urlencode(params))
        full_url = self.host + url

        start = self._loop.time()
        try:
            with async_timeout.timeout(timeout, loop=self._loop):
                async with self._session.request(
                        method, full_url, data=body) as resp:
                    raw_data = await resp.text()
        except Exception as e:
            self.log_request_fail(
                method, full_url, url, body,
                self._loop.time() - start, exception=e)
            if isinstance(e, asyncio.TimeoutError):
                raise ConnectionTimeout('TIMEOUT', str(e), e)
            raise ConnectionError('N/A', str(e), e)

        duration = self._loop.time() - start
        # raise errors based on http status codes,
        # let the client handle those if needed
        if not (200 <= resp.status < 300) and resp.status not in ignore:
            self.log_request_fail(
                method, full_url, url, body, duration, resp.status, raw_data)
            self._raise_error(resp.status, raw_data)

        self.log_request_success(
            method, full_url, url, body, resp.status, raw_data, duration)

        return resp.status, resp.headers, raw_data