import elasticsearch
from biothings.utils.exceptions import RepositoryVerificationFailed
[docs]
class Repository:
def __init__(self, client, repository):
# Corresponds to HTTP operations on
# /_snapshot/<repository>
self.client = client
self.name = repository
[docs]
def exists(self):
try:
self.client.snapshot.get_repository(self.name)
except elasticsearch.exceptions.NotFoundError:
return False
return True
[docs]
def create(self, **body):
# https://www.elastic.co/guide/en/elasticsearch/plugins/current/repository-s3-client.html
return self.client.snapshot.create_repository(self.name, body=body)
[docs]
def delete(self):
self.client.snapshot.delete_repository(self.name)
[docs]
def verify(self, config):
"""A repository is consider properly setup and working, when:
- passes verification of ElasticSearch
- it's settings must match with the snapshot's config.
"""
# Check if the repo pass ElasticSearch's verification
self.client.snapshot.verify_repository(self.name)
# Check if the repo's settings match with the snapshot's config
repo_settings = self.client.snapshot.get_repository(self.name)[self.name]
incorrect_data = {}
for field in ["type", "settings"]:
if config[field] != repo_settings[field]:
incorrect_data[field] = {"config": config[field], "repo": repo_settings[field]}
if incorrect_data:
raise RepositoryVerificationFailed(
{
"error": "repository_verification_exception",
"detail": {
"message": "the repository's settings is not match with snapshot config.",
"diff": incorrect_data,
},
}
)
def __str__(self):
return (
f"<Repository {'READY' if self.exists() else 'MISSING'}"
f" name='{self.name}'"
f" client={self.client}"
f">"
)
[docs]
def test_01():
from elasticsearch import Elasticsearch
client = Elasticsearch()
snapshot = Repository(client, "mynews")
print(snapshot)
[docs]
def test_02():
from elasticsearch import Elasticsearch
client = Elasticsearch()
snapshot = Repository(client, "______")
print(snapshot)
if __name__ == "__main__":
test_01()