(self, host='127.0.0.1', port=9200)
Create a OpenSearch client.
Create a OpenSearch client.
Create a OpenSearch client .
def __init__(self, host='127.0.0.1', port=9200):
        """Create a OpenSearch client."""
        super().__init__()
        self._error_container = {}

        self.user = current_app.config.get('OPENSEARCH_USER', 'user')
        self.password = current_app.config.get('OPENSEARCH_PASSWORD', 'pass')
        self.ssl = current_app.config.get('OPENSEARCH_SSL', False)
        self.verify = current_app.config.get('OPENSEARCH_VERIFY_CERTS', True)
        self.timeout = current_app.config.get('OPENSEARCH_TIMEOUT', 10)

        parameters = {}
        if self.ssl:
            parameters['use_ssl'] = self.ssl
            parameters['verify_certs'] = self.verify

        if self.user and self.password:
            parameters['http_auth'] = (self.user, self.password)
        if self.timeout:
            parameters['timeout'] = self.timeout

        self.client = OpenSearch(
            [{'host': host, 'port': port}], **parameters)

        self.import_counter = Counter()
        self.import_events = []
        self._request_timeout = current_app.config.get(
            'TIMEOUT_FOR_EVENT_IMPORT', self.DEFAULT_EVENT_IMPORT_TIMEOUT)
def __init__ ( self , host = '127.0.0.1' , port = 9200 ) : super ( ) . __init__ ( ) self . _error_container = { } self . user = current_app . config . get ( 'OPENSEARCH_USER' , 'user' ) self . password = current_app . config . get ( 'OPENSEARCH_PASSWORD' , 'pass' ) self . ssl = current_app . config . get ( 'OPENSEARCH_SSL' , False ) self . verify = current_app . config . get ( 'OPENSEARCH_VERIFY_CERTS' , True ) self . timeout = current_app . config . get ( 'OPENSEARCH_TIMEOUT' , 10 ) parameters = { } if self . ssl : parameters [ 'use_ssl' ] = self . ssl parameters [ 'verify_certs' ] = self . verify if self . user and self . password : parameters [ 'http_auth' ] = ( self . user , self . password ) if self . timeout : parameters [ 'timeout' ] = self . timeout self . client = OpenSearch ( [ { 'host' : host , 'port' : port } ] , * * parameters ) self . import_counter = Counter ( ) self . import_events = [ ] self . _request_timeout = current_app . config . get ( 'TIMEOUT_FOR_EVENT_IMPORT' , self . DEFAULT_EVENT_IMPORT_TIMEOUT )