#!/usr/bin/python
# -*- coding: utf-8 -*-
"""Metadata Client Module."""
from json import loads, dumps
import requests
[docs]class PMClientError(Exception):
"""Base Exception Error Class."""
pass
[docs]class PMClient(object):
"""
Pacifica Metadata Client.
This class provides client API to connect to the metadata service
"""
headers = {'content-type': 'application/json'}
[docs] def __init__(self, url):
"""Constructor takes the url to the endpoint."""
self.url = url
[docs] def create(self, cls_type, set_hash):
"""Create the object of type based on hash."""
ret = requests.put('{0}/{1}'.format(self.url, cls_type),
data=dumps(set_hash), headers=self.headers)
if int(ret.status_code / 100) == 2:
return True
elif int(ret.status_code / 100) == 5:
raise PMClientError('Internal Server Error ({0}) {1}'.format(
ret.status_code, ret.content))
else:
raise PMClientError('Unknown Error ({0}) {1}'.format(
ret.status_code, ret.content))
[docs] def update(self, cls_type, query_hash, set_hash):
"""
Update the object.
Update object of type returned from query_hash and
set the values in set_hash
"""
ret = requests.post('{0}/{1}'.format(self.url, cls_type),
params=query_hash,
data=dumps(set_hash),
headers=self.headers)
if int(ret.status_code / 100) == 2:
return True
if int(ret.status_code / 100) == 4:
return False
elif int(ret.status_code / 100) == 5:
raise PMClientError('Internal Server Error ({0}) {1}'.format(
ret.status_code, ret.content))
else:
raise PMClientError('Unknown Error ({0}) {1}'.format(
ret.status_code, ret.content))
[docs] def get(self, cls_type, query_hash):
"""Get the object of type from query_hash."""
ret = requests.get('{0}/{1}'.format(self.url, cls_type),
params=query_hash, allow_redirects=True)
if int(ret.status_code / 100) == 2:
return loads(ret.content.decode('UTF-8'))
elif int(ret.status_code / 100) == 4:
return {}
elif int(ret.status_code / 100) == 5:
raise PMClientError('Internal Server Error ({0}) {1}'.format(
ret.status_code, ret.content))
else:
raise PMClientError('Unknown Error ({0}) {1}'.format(
ret.status_code, ret.content))
[docs] def delete(self, cls_type, query_hash):
"""Delete the object of type from query_hash."""
ret = requests.delete(
'{0}/{1}'.format(self.url, cls_type), params=query_hash, allow_redirects=True)
if int(ret.status_code / 100) == 2 or int(ret.status_code / 100) == 4:
return True
elif int(ret.status_code / 100) == 5:
raise PMClientError('Internal Server Error ({0}) {1}'.format(
ret.status_code, ret.content))
else:
raise PMClientError('Unknown Error ({0}) {1}'.format(
ret.status_code, ret.content))