in granule_ingester/granule_ingester/writers/ElasticsearchStore.py [0:0]
def get_connection(self) -> Elasticsearch:
if self.elastic_url:
if not self.username or not self.password:
return Elasticsearch([self.elastic_url])
else:
return Elasticsearch([self.elastic_url], http_auth=(self.username, self.password))
else:
raise RuntimeError('No Elasticsearch URL')