mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

removed unused HetznerClient methods to fix imparial coverage

Browse Source
This commit is contained in:
Ricard Bejarano
2021-04-27 07:31:05 +02:00
parent 612738b327
commit 0de9efd032
1 changed files with 17 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
octodns/provider/hetzner.py
View File

@@ -32,11 +32,6 @@ class HetznerClientUnauthorized(HetznerClientException):
class HetznerClient(object):
'''
Hetzner DNS Public API v1 client class.
Zone and Record resources are (almost) fully supported, even if unnecessary
to future-proof this client. Bulk Record create/update is not supported.
No support for Primary Servers.
'''
BASE_URL = 'https://dns.hetzner.com/api/v1'
@@ -46,8 +41,8 @@ class HetznerClient(object):
session.headers.update({'Auth-API-Token': token})
self._session = session
def _do_raw(self, method, path, params=None, data=None):
url = self.BASE_URL + path
def _do(self, method, path, params=None, data=None):
url = '{}{}'.format(self.BASE_URL, path)
response = self._session.request(method, url, params=params, json=data)
if response.status_code == 401:
raise HetznerClientUnauthorized()
@@ -56,20 +51,15 @@ class HetznerClient(object):
response.raise_for_status()
return response
def _do(self, method, path, params=None, data=None):
return self._do_raw(method, path, params, data).json()
def _do_with_pagination(self, method, path, key, params=None, data=None,
per_page=100):
pagination_params = {'page': 1, 'per_page': per_page}
if params is not None:
params = {**params, **pagination_params}
else:
params = pagination_params
def _do_json(self, method, path, params=None, data=None):
return self._do(method, path, params, data).json()
def _do_json_paginate(self, method, path, key, params=None, data=None,
per_page=100):
items = []
params = {**{'page': 1, 'per_page': per_page}, **params}
while True:
response = self._do(method, path, params, data)
response = self._do_json(method, path, params, data)
items += response[key]
if response['meta']['pagination']['page'] >= \
response['meta']['pagination']['last_page']:
@@ -79,54 +69,28 @@ class HetznerClient(object):
def zones_get(self, name=None, search_name=None):
params = {'name': name, 'search_name': search_name}
return self._do_with_pagination('GET', '/zones', 'zones',
params=params)
def zone_get(self, zone_id):
return self._do('GET', '/zones/' + zone_id)['zone']
return self._do_json_paginate('GET', '/zones', 'zones', params=params)
def zone_create(self, name, ttl=None):
data = {'name': name, 'ttl': ttl}
return self._do('POST', '/zones', data=data)['zone']
def zone_update(self, zone_id, name, ttl=None):
data = {'name': name, 'ttl': ttl}
return self._do('PUT', '/zones/' + zone_id, data=data)['zone']
def zone_delete(self, zone_id):
return self._do('DELETE', '/zones/' + zone_id)
def _replace_at(self, record):
record['name'] = '' if record['name'] == '@' else record['name']
return record
return self._do_json('POST', '/zones', data=data)['zone']
def zone_records_get(self, zone_id):
params = {'zone_id': zone_id}
# No need to handle pagination as it returns all records by default.
return [
self._replace_at(record)
for record in self._do('GET', '/records', params=params)['records']
]
def zone_record_get(self, record_id):
record = self._do('GET', '/records/' + record_id)['record']
return self._replace_at(record)
records = self._do_json('GET', '/records', params=params)['records']
for record in records:
if record['name'] == '@':
record['name'] = ''
return records
def zone_record_create(self, zone_id, name, _type, value, ttl=None):
data = {'name': name or '@', 'ttl': ttl, 'type': _type, 'value': value,
'zone_id': zone_id}
record = self._do('POST', '/records', data=data)['record']
return self._replace_at(record)
def zone_record_update(self, zone_id, record_id, name, _type, value,
ttl=None):
data = {'name': name or '@', 'ttl': ttl, 'type': _type, 'value': value,
'zone_id': zone_id}
record = self._do('PUT', '/records/' + record_id, data=data)['record']
return self._replace_at(record)
self._do('POST', '/records', data=data)
def zone_record_delete(self, zone_id, record_id):
return self._do('DELETE', '/records/' + record_id)
self._do('DELETE', '/records/{}'.format(record_id))
class HetznerProvider(BaseProvider):