64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
import requests
|
|
import requests.packages
|
|
from typing import List, Dict
|
|
|
|
|
|
class PdnsRestAdapter:
|
|
def __init__(self, hostname: str, api_key: str = '', ver: str = 'v1'):
|
|
self.url = f"http://{hostname}/api/{ver}/"
|
|
self._api_key = api_key
|
|
|
|
def get(self, endpoint: str) -> List[Dict]:
|
|
full_url = self.url + endpoint
|
|
headers = {'X-API-key': self._api_key}
|
|
response = requests.get(url=full_url, headers=headers)
|
|
data_out = response.json()
|
|
# uhm...
|
|
# if response.status.code >= 200 and response.status.code <= 299:
|
|
return data_out
|
|
# raise Exception(data_out['error']) # TODO: raise custom exception later
|
|
|
|
def post(self, endpoint: str, ep_params: Dict = None, data: Dict = None):
|
|
full_url = self.url + endpoint
|
|
headers = {'X-API-key': self._api_key}
|
|
response = requests.post(url=full_url, headers=headers, params=ep_params, json=data)
|
|
if response.status_code >= 200 and response.status_code <= 299: # OK
|
|
return
|
|
print("PdnsRestAdapter::post - error")
|
|
data_out = response.json()
|
|
raise Exception(data_out['error'])
|
|
|
|
def patch(self, endpoint: str, ep_params: Dict = None, data: Dict = None):
|
|
full_url = self.url + endpoint
|
|
headers = {'X-API-key': self._api_key}
|
|
response = requests.patch(url=full_url, headers=headers, params=ep_params, json=data)
|
|
if response.status_code >= 200 and response.status_code <= 299: # OK
|
|
return
|
|
# if response.status_code < 500:
|
|
# data_out = response.json()
|
|
# else:
|
|
data_out = {"error": f"{response.status_code}"}
|
|
raise Exception(data_out['error'])
|
|
|
|
# def post(self, endpoint: str, params: Dict = None, data: Dict = None):
|
|
# full_url = self.url + endpoint
|
|
# headers = {'X-API-key': self._api_key}
|
|
# response = requests.post(url=full_url, params=params, headers=headers, json=data)
|
|
# # data_out = response.json()
|
|
# return response
|
|
# if response.status.code >= 200 and response.status.code <= 299:
|
|
# return
|
|
# raise Exception(data_out['message']) # TODO: raise custom exception later
|
|
|
|
def get_record(self, name=None, type=None):
|
|
domain = '.'.join(name.split('.')[1:])
|
|
if name[:-1] != '.':
|
|
name += '.'
|
|
result = self.get('servers/localhost/zones/' + domain)
|
|
record = None
|
|
for r in result["rrsets"]:
|
|
if r['name'] == name and r['type'] == type:
|
|
rrset = r
|
|
print(f"PdnsRestAdapter::get_record(): {rrset=}")
|
|
return rrset
|