From 9aeeca8ebdf15493726bfef2363df9237c2661fc Mon Sep 17 00:00:00 2001 From: Hayder Mustafa Date: Fri, 6 Sep 2024 19:24:20 +0300 Subject: [PATCH] [FIX] passing token when calling a function inside another --- src/hetzner_dns_tools/hetzner_dns_helpers.py | 19 +- src/hetzner_dns_tools/record_create.py | 91 +++++---- src/hetzner_dns_tools/record_delete.py | 127 +++++++------ src/hetzner_dns_tools/record_get.py | 190 ++++++++++--------- src/hetzner_dns_tools/record_list.py | 41 ++-- src/hetzner_dns_tools/record_update.py | 114 ++++++----- src/hetzner_dns_tools/zone_create.py | 51 ++--- src/hetzner_dns_tools/zone_delete.py | 42 ++-- src/hetzner_dns_tools/zone_get.py | 69 +++---- src/hetzner_dns_tools/zone_list.py | 17 +- 10 files changed, 415 insertions(+), 346 deletions(-) diff --git a/src/hetzner_dns_tools/hetzner_dns_helpers.py b/src/hetzner_dns_tools/hetzner_dns_helpers.py index 7e9193e..dfdf229 100644 --- a/src/hetzner_dns_tools/hetzner_dns_helpers.py +++ b/src/hetzner_dns_tools/hetzner_dns_helpers.py @@ -1,6 +1,7 @@ -import requests import sys +import requests + def check_response_for_errors(response_dict): """ @@ -10,14 +11,14 @@ def check_response_for_errors(response_dict): - In Bash: Print error to the console and exit with error code 1 - In Python: Raise a ValueError exception """ - if response_dict.get('error') or response_dict.get('message'): + if response_dict.get("error") or response_dict.get("message"): error_message = "" - if response_dict.get('error'): - error_message = response_dict['error']['message'] - elif response_dict.get('message'): - error_message = response_dict['message'] + if response_dict.get("error"): + error_message = response_dict["error"]["message"] + elif response_dict.get("message"): + error_message = response_dict["message"] - if __name__ == '__main__': + if __name__ == "__main__": print(f"Error: {error_message}") sys.exit(1) # exit with error else: @@ -26,7 +27,7 @@ def check_response_for_errors(response_dict): def exit_with_error(error_message, exception_type=ValueError): """A boilerplate function to exit the program with an error message.""" - if __name__ == '__main__': + if __name__ == "__main__": print(f"Error: {error_message}") sys.exit(1) # exit with error else: @@ -36,7 +37,7 @@ def exit_with_error(error_message, exception_type=ValueError): def handle_request_exception(err): """A boilerplate function for handling request exceptions.""" # when running via the terminal, print output to console then exit - if __name__ == '__main__': + if __name__ == "__main__": print(f"Error: {err}") sys.exit(1) # exit with error else: diff --git a/src/hetzner_dns_tools/record_create.py b/src/hetzner_dns_tools/record_create.py index 582d112..17e7ab6 100644 --- a/src/hetzner_dns_tools/record_create.py +++ b/src/hetzner_dns_tools/record_create.py @@ -2,21 +2,24 @@ import json import os -import requests import sys +import requests + from . import hetzner_dns_helpers as helpers from .zone_list import zone_list -def record_create(hetzner_dns_token=None, - record_type=None, - name=None, - value=None, - ttl=None, - zone_id=None, - zone_name=None, - id_only=False): +def record_create( + hetzner_dns_token=None, + record_type=None, + name=None, + value=None, + ttl=None, + zone_id=None, + zone_name=None, + id_only=False, +): """ Create a new record. https://dns.hetzner.com/api-docs/#operation/CreateRecord @@ -47,99 +50,103 @@ def record_create(hetzner_dns_token=None, in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(record_create.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] if record_type is None: # get record_type from environment variable - record_type = os.environ['RECORD_TYPE']\ - if os.environ.get('RECORD_TYPE') else os.environ['TYPE'] + record_type = ( + os.environ["RECORD_TYPE"] + if os.environ.get("RECORD_TYPE") + else os.environ["TYPE"] + ) if name is None: # get name from environment variable - name = os.environ['NAME'] if os.environ.get('NAME') else '@' + name = os.environ["NAME"] if os.environ.get("NAME") else "@" if value is None: # get value from environment variable - value = os.environ['VALUE'] + value = os.environ["VALUE"] if ttl is None: - if os.environ.get('TTL'): + if os.environ.get("TTL"): # get ttl from environment variable - ttl = int(os.environ['TTL']) + ttl = int(os.environ["TTL"]) else: # use default value for TTL ttl = 86400 - if zone_name is None and os.environ.get('ZONE_NAME'): + if zone_name is None and os.environ.get("ZONE_NAME"): # get ttl from environment variable - zone_name = os.environ['ZONE_NAME'] + zone_name = os.environ["ZONE_NAME"] # if zone_name exists, use it to obtain the zone_id if zone_name: # get list of zones - response_dict = zone_list() + response_dict = zone_list(hetzner_dns_token=hetzner_dns_token) # check response for errors helpers.check_response_for_errors(response_dict) # check for matching zone - dns_zones = response_dict['zones'] + dns_zones = response_dict["zones"] for zone in dns_zones: - if zone['name'] == zone_name: - zone_id = zone['id'] + if zone["name"] == zone_name: + zone_id = zone["id"] break # if no matching zone found, then exit with error if zone_id is None: helpers.exit_with_error("zone not found") - if not zone_name and not zone_id and not os.environ.get('ZONE_ID'): + if not zone_name and not zone_id and not os.environ.get("ZONE_ID"): # if neither zone_name or zone_id exist, then exit with error helpers.exit_with_error("Must include one of: zone_id, zone_name") if zone_id is None: # get zone_id from environment variable - zone_id = os.environ['ZONE_ID'] + zone_id = os.environ["ZONE_ID"] try: - params = {'ttl': ttl, - 'type': record_type, - 'value': value, - 'zone_id': zone_id} + params = {"ttl": ttl, "type": record_type, "value": value, "zone_id": zone_id} if name: - params['name'] = name - - response = requests.post(url='https://dns.hetzner.com/api/v1/records', - headers={'Content-Type': 'application/json', - 'Auth-API-Token': hetzner_dns_token}, - data=json.dumps(params)) - - decoded_response = response.content.decode('utf-8') + params["name"] = name + + response = requests.post( + url="https://dns.hetzner.com/api/v1/records", + headers={ + "Content-Type": "application/json", + "Auth-API-Token": hetzner_dns_token, + }, + data=json.dumps(params), + ) + + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) # check response for errors helpers.check_response_for_errors(response_dict) # return the expected result - if id_only or os.environ.get('ID_ONLY') == '1': + if id_only or os.environ.get("ID_ONLY") == "1": # return the zone_id - result = response_dict['record']['id'] + result = response_dict["record"]["id"] - if __name__ == '__main__': + if __name__ == "__main__": print(result) sys.exit(0) # exit successfully return result else: # return all zone data - if __name__ == '__main__': + if __name__ == "__main__": print(json.dumps(response_dict)) sys.exit(0) # exit successfully @@ -151,5 +158,5 @@ def record_create(hetzner_dns_token=None, helpers.handle_request_exception(err) -if __name__ == '__main__': +if __name__ == "__main__": record_create() diff --git a/src/hetzner_dns_tools/record_delete.py b/src/hetzner_dns_tools/record_delete.py index 4a5f021..df4ade5 100644 --- a/src/hetzner_dns_tools/record_delete.py +++ b/src/hetzner_dns_tools/record_delete.py @@ -2,9 +2,10 @@ import json import os -import requests import sys +import requests + from . import hetzner_dns_helpers as helpers from .record_get import record_get @@ -12,17 +13,18 @@ def delete_record_by_id(hetzner_dns_token, record_id): try: response = requests.delete( - url=f'https://dns.hetzner.com/api/v1/records/{record_id}', - headers={'Auth-API-Token': hetzner_dns_token}) + url=f"https://dns.hetzner.com/api/v1/records/{record_id}", + headers={"Auth-API-Token": hetzner_dns_token}, + ) - decoded_response = response.content.decode('utf-8') + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) # check response for errors helpers.check_response_for_errors(response_dict) # when running via the terminal, print output to console - if __name__ == '__main__': + if __name__ == "__main__": print("OK") return "OK" @@ -31,17 +33,19 @@ def delete_record_by_id(hetzner_dns_token, record_id): helpers.handle_request_exception(err) -def record_delete(hetzner_dns_token=None, - record_id=None, - record_ids=None, - zone_id=None, - zone_name=None, - name=None, - record_type=None, - value=None, - first_record_only=None, - delete_multiple_records=False, - search_all_zones=False): +def record_delete( + hetzner_dns_token=None, + record_id=None, + record_ids=None, + zone_id=None, + zone_name=None, + name=None, + record_type=None, + value=None, + first_record_only=None, + delete_multiple_records=False, + search_all_zones=False, +): """ Delete an existing record. https://dns.hetzner.com/api-docs/#operation/DeleteRecord @@ -72,86 +76,90 @@ def record_delete(hetzner_dns_token=None, in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(record_delete.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] - if record_id is None and os.environ.get('RECORD_ID'): + if record_id is None and os.environ.get("RECORD_ID"): # get record_id from environment variable - record_id = os.environ['RECORD_ID'] + record_id = os.environ["RECORD_ID"] - if record_ids is None and os.environ.get('RECORD_IDS'): + if record_ids is None and os.environ.get("RECORD_IDS"): # get record_ids from environment variable - record_ids = os.environ['RECORD_IDS'] + record_ids = os.environ["RECORD_IDS"] - if not first_record_only\ - and os.environ.get('FIRST_RECORD_ONLY'): + if not first_record_only and os.environ.get("FIRST_RECORD_ONLY"): # get first_record_only from environment variable - first_record_only = os.environ['FIRST_RECORD_ONLY'] + first_record_only = os.environ["FIRST_RECORD_ONLY"] - if not search_all_zones\ - and os.environ.get('SEARCH_ALL_ZONES'): + if not search_all_zones and os.environ.get("SEARCH_ALL_ZONES"): # get search_all_zones from environment variable - search_all_zones = os.environ['SEARCH_ALL_ZONES'] + search_all_zones = os.environ["SEARCH_ALL_ZONES"] - if not delete_multiple_records\ - and os.environ.get('DELETE_MULTIPLE_RECORDS'): + if not delete_multiple_records and os.environ.get("DELETE_MULTIPLE_RECORDS"): # get delete_multiple_records from environment variable - delete_multiple_records = os.environ['DELETE_MULTIPLE_RECORDS'] + delete_multiple_records = os.environ["DELETE_MULTIPLE_RECORDS"] # do an indirect lookup of all relevant records that match the parameters # and allow a single match to be returned if not record_id: - if zone_id is None and os.environ.get('ZONE_ID'): + if zone_id is None and os.environ.get("ZONE_ID"): # get zone_id from environment variable - zone_id = os.environ['ZONE_ID'] + zone_id = os.environ["ZONE_ID"] - if zone_name is None and os.environ.get('ZONE_NAME'): + if zone_name is None and os.environ.get("ZONE_NAME"): # get zone_name from environment variable - zone_name = os.environ['ZONE_NAME'] + zone_name = os.environ["ZONE_NAME"] - if name is None and os.environ.get('NAME'): + if name is None and os.environ.get("NAME"): # get name from environment variable - name = os.environ.get('NAME', None) + name = os.environ.get("NAME", None) - if record_type is None and os.environ.get('TYPE'): + if record_type is None and os.environ.get("TYPE"): # get record_type from environment variable - record_type = os.environ['RECORD_TYPE']\ - if os.environ.get('RECORD_TYPE') else os.environ['TYPE'] + record_type = ( + os.environ["RECORD_TYPE"] + if os.environ.get("RECORD_TYPE") + else os.environ["TYPE"] + ) - if value is None and os.environ.get('VALUE'): + if value is None and os.environ.get("VALUE"): # get value from environment variable - value = os.environ['VALUE'] + value = os.environ["VALUE"] # this method will return a string if one record is returned, # and a list if multiple records are returned - record_get_id_result =\ - record_get(zone_id=zone_id, - zone_name=zone_name, - name=name, - record_type=record_type, - value=value, - id_only=True, - first_record_only=first_record_only, - allow_multiple_records=delete_multiple_records, - search_all_zones=search_all_zones) + record_get_id_result = record_get( + hetzner_dns_token=hetzner_dns_token, + zone_id=zone_id, + zone_name=zone_name, + name=name, + record_type=record_type, + value=value, + id_only=True, + first_record_only=first_record_only, + allow_multiple_records=delete_multiple_records, + search_all_zones=search_all_zones, + ) if type(record_get_id_result) == list: records_count = len(record_get_id_result) # do not delete multiple records unless explicitly requested if not delete_multiple_records: - error_message = f"Found {records_count} records. To "\ - "delete multiple records via a query, assign a truthy "\ - "value to 'delete_multiple_records', or use "\ - "'first_record_only' to get only the "\ - "first record. Capitalize these values if using "\ + error_message = ( + f"Found {records_count} records. To " + "delete multiple records via a query, assign a truthy " + "value to 'delete_multiple_records', or use " + "'first_record_only' to get only the " + "first record. Capitalize these values if using " "environment variables." + ) helpers.exit_with_error(error_message) else: @@ -164,7 +172,8 @@ def record_delete(hetzner_dns_token=None, if record_id and record_ids: helpers.exit_with_error( - "Cannot use 'record_id' and 'record_ids' at the same time.") + "Cannot use 'record_id' and 'record_ids' at the same time." + ) elif record_id: delete_record_by_id(hetzner_dns_token, record_id) elif record_ids: @@ -174,5 +183,5 @@ def record_delete(hetzner_dns_token=None, helpers.exit_with_error("No 'record_id' or 'record_ids' found.") -if __name__ == '__main__': +if __name__ == "__main__": record_delete() diff --git a/src/hetzner_dns_tools/record_get.py b/src/hetzner_dns_tools/record_get.py index e8a5ac3..913657e 100644 --- a/src/hetzner_dns_tools/record_get.py +++ b/src/hetzner_dns_tools/record_get.py @@ -2,25 +2,28 @@ import json import os -import requests import sys +import requests + from . import hetzner_dns_helpers as helpers from .record_list import record_list from .zone_get import zone_get -def record_get(hetzner_dns_token=None, - record_id=None, - zone_id=None, - zone_name=None, - record_type=None, - name=None, - value=None, - first_record_only=False, - allow_multiple_records=False, - search_all_zones=False, - id_only=False): +def record_get( + hetzner_dns_token=None, + record_id=None, + zone_id=None, + zone_name=None, + record_type=None, + name=None, + value=None, + first_record_only=False, + allow_multiple_records=False, + search_all_zones=False, + id_only=False, +): """ Get info about an existing record. https://dns.hetzner.com/api-docs/#operation/GetRecord @@ -65,33 +68,36 @@ def record_get(hetzner_dns_token=None, in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(record_get.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] - if record_id is None and os.environ.get('RECORD_ID'): + if record_id is None and os.environ.get("RECORD_ID"): # get record_id from environment variable - record_id = os.environ['RECORD_ID'] + record_id = os.environ["RECORD_ID"] - if not id_only and os.environ.get('ID_ONLY'): + if not id_only and os.environ.get("ID_ONLY"): # get id_only from environment variable - id_only = os.environ['ID_ONLY'] + id_only = os.environ["ID_ONLY"] # if record_id exists, do a direct lookup to obtain the record if record_id: # get response try: response = requests.get( - url=f'https://dns.hetzner.com/api/v1/records/{record_id}', - headers={'Auth-API-Token': hetzner_dns_token, - 'Content-Type': 'application/json; charset=utf-8'}) - - decoded_response = response.content.decode('utf-8') + url=f"https://dns.hetzner.com/api/v1/records/{record_id}", + headers={ + "Auth-API-Token": hetzner_dns_token, + "Content-Type": "application/json; charset=utf-8", + }, + ) + + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) # check response for errors @@ -99,14 +105,14 @@ def record_get(hetzner_dns_token=None, # if id_only, return just the record ID if id_only: - if __name__ == '__main__': - print(response_dict['record']['id']) + if __name__ == "__main__": + print(response_dict["record"]["id"]) sys.exit(0) # exit successfully else: - return response_dict['record']['id'] + return response_dict["record"]["id"] # when running via the terminal, print output to console then exit - if __name__ == '__main__': + if __name__ == "__main__": print(decoded_response) sys.exit(0) # exit successfully @@ -116,78 +122,88 @@ def record_get(hetzner_dns_token=None, helpers.handle_request_exception(err) # do an indirect lookup of all relevant records and filter via given params - if zone_id is None and os.environ.get('ZONE_ID'): + if zone_id is None and os.environ.get("ZONE_ID"): # get zone_id from environment variable - zone_id = os.environ['ZONE_ID'] + zone_id = os.environ["ZONE_ID"] - if zone_name is None and os.environ.get('ZONE_NAME'): + if zone_name is None and os.environ.get("ZONE_NAME"): # get zone_name from environment variable - zone_name = os.environ['ZONE_NAME'] + zone_name = os.environ["ZONE_NAME"] - if record_type is None and os.environ.get('TYPE'): + if record_type is None and os.environ.get("TYPE"): # get record_type from environment variable - record_type = os.environ['RECORD_TYPE']\ - if os.environ.get('RECORD_TYPE') else os.environ['TYPE'] + record_type = ( + os.environ["RECORD_TYPE"] + if os.environ.get("RECORD_TYPE") + else os.environ["TYPE"] + ) - if name is None and os.environ.get('NAME'): + if name is None and os.environ.get("NAME"): # get name from environment variable - name = os.environ.get('NAME', None) + name = os.environ.get("NAME", None) - if value is None and os.environ.get('VALUE'): + if value is None and os.environ.get("VALUE"): # get value from environment variable - value = os.environ['VALUE'] + value = os.environ["VALUE"] - if not first_record_only\ - and os.environ.get('FIRST_RECORD_ONLY'): + if not first_record_only and os.environ.get("FIRST_RECORD_ONLY"): # get first_record_only from environment variable - first_record_only = os.environ['FIRST_RECORD_ONLY'] + first_record_only = os.environ["FIRST_RECORD_ONLY"] - if not allow_multiple_records\ - and os.environ.get('ALLOW_MULTIPLE_RECORDS'): + if not allow_multiple_records and os.environ.get("ALLOW_MULTIPLE_RECORDS"): # get allow_multiple_records from environment variable - allow_multiple_records = os.environ['ALLOW_MULTIPLE_RECORDS'] + allow_multiple_records = os.environ["ALLOW_MULTIPLE_RECORDS"] if first_record_only and allow_multiple_records: - helpers\ - .exit_with_error("This combination of options doesn't make sense.") + helpers.exit_with_error("This combination of options doesn't make sense.") - if not search_all_zones\ - and os.environ.get('SEARCH_ALL_ZONES'): + if not search_all_zones and os.environ.get("SEARCH_ALL_ZONES"): # get search_all_zones from environment variable - search_all_zones = os.environ['SEARCH_ALL_ZONES'] + search_all_zones = os.environ["SEARCH_ALL_ZONES"] # BEGIN validation # # if no record_id exists, ensure that zone_name or zone_id exist, # in order to prevent pulling records from multiple zones - if not record_id and not zone_name and not zone_id\ - and not search_all_zones: - error_message = "In order to prevent records from being pulled from "\ - "more than one zone, you must specify one (or more) of: "\ - "record_id, zone_id, or zone_name. You can override this "\ + if not record_id and not zone_name and not zone_id and not search_all_zones: + error_message = ( + "In order to prevent records from being pulled from " + "more than one zone, you must specify one (or more) of: " + "record_id, zone_id, or zone_name. You can override this " "behavior by assigning a truthy value to 'search_all_zones'." + ) helpers.exit_with_error(error_message) # ensure that one or more optional parameters exist before doing # an indirect lookup - if not record_id and not search_all_zones and not zone_name\ - and not name and not record_type and not value: - error_message =\ - "You must provide a record_id or one or more of the following: "\ - "name, record_type (environment variable: TYPE), or value, "\ + if ( + not record_id + and not search_all_zones + and not zone_name + and not name + and not record_type + and not value + ): + error_message = ( + "You must provide a record_id or one or more of the following: " + "name, record_type (environment variable: TYPE), or value, " "*OR* you must set a truthy value for 'search_all_zones.'" + ) helpers.exit_with_error(error_message) # if zone_name passed, lookup the zone that matches it to get zone_id if zone_name: - zone_name_id = zone_get(zone_name=zone_name, id_only=True) + zone_name_id = zone_get( + hetzner_dns_token=hetzner_dns_token, zone_name=zone_name, id_only=True + ) if not zone_id: zone_id = zone_name_id elif zone_id != zone_name_id: - error_message =\ - "The zone_id you entered does not match the zone_id of "\ + error_message = ( + "The zone_id you entered does not match the zone_id of " "the zone_name you entered." - if __name__ == '__main__': + ) + if __name__ == "__main__": print(f"Error: {error_message}") sys.exit(1) # exit with error else: @@ -195,8 +211,9 @@ def record_get(hetzner_dns_token=None, # END validation # - records = record_list(hetzner_dns_token=hetzner_dns_token, - zone_id=zone_id)['records'] + records = record_list(hetzner_dns_token=hetzner_dns_token, zone_id=zone_id)[ + "records" + ] # iterate over the given parameters, adding any matching records that # are not yet in the list @@ -207,11 +224,11 @@ def record_get(hetzner_dns_token=None, for record in records: # if record does not meet any one qualifying criterion, # then skip over it and continue the loop - if name and record['name'] != name: + if name and record["name"] != name: continue - if record_type and record['type'] != record_type: + if record_type and record["type"] != record_type: continue - if value and record['value'] != value: + if value and record["value"] != value: continue # if the record has not fail any of the qualifications, # then add it to the list @@ -220,7 +237,7 @@ def record_get(hetzner_dns_token=None, # if no records found, return empty dictionary if len(filtered_records) == 0: # when running via the terminal, print output to console then exit - if __name__ == '__main__': + if __name__ == "__main__": print({}) sys.exit(0) # exit successfully return {} @@ -229,14 +246,14 @@ def record_get(hetzner_dns_token=None, if len(filtered_records) == 1: # if id_only, return just the record ID if id_only: - if __name__ == '__main__': - print(filtered_records[0]['id']) + if __name__ == "__main__": + print(filtered_records[0]["id"]) sys.exit(0) # exit successfully else: - return filtered_records[0]['id'] + return filtered_records[0]["id"] # when running via the terminal, print output to console then exit - if __name__ == '__main__': + if __name__ == "__main__": print(json.dumps(filtered_records[0])) sys.exit(0) # exit successfully return filtered_records[0] @@ -248,23 +265,23 @@ def record_get(hetzner_dns_token=None, if first_record_only: # if id_only, return just the record ID if id_only: - if __name__ == '__main__': - print(filtered_records[0]['id']) + if __name__ == "__main__": + print(filtered_records[0]["id"]) sys.exit(0) # exit successfully else: - return filtered_records[0]['id'] + return filtered_records[0]["id"] # when running via the terminal, print output to console then exit - if __name__ == '__main__': + if __name__ == "__main__": print(json.dumps(filtered_records[0])) sys.exit(0) # exit successfully return filtered_records[0] elif id_only and allow_multiple_records: id_list = [] for record in filtered_records: - id_list.append(record['id']) + id_list.append(record["id"]) - if __name__ == '__main__': + if __name__ == "__main__": print(json.dumps(id_list)) sys.exit(0) # exit successfully else: @@ -273,19 +290,20 @@ def record_get(hetzner_dns_token=None, elif allow_multiple_records and not id_only: # when running via the terminal, print output to console, # then exit - if __name__ == '__main__': + if __name__ == "__main__": print(json.dumps(filtered_records)) sys.exit(0) # exit successfully return filtered_records else: - error_message =\ - f"Found {records_count} record{plural}. Assign a truthy "\ - "value to 'allow_multiple_records' To get all relevant "\ - "records, or use 'first_record_only' to get only the "\ - "first record. Capitalize these values if using "\ + error_message = ( + f"Found {records_count} record{plural}. Assign a truthy " + "value to 'allow_multiple_records' To get all relevant " + "records, or use 'first_record_only' to get only the " + "first record. Capitalize these values if using " "environment variables." + ) helpers.exit_with_error(error_message) -if __name__ == '__main__': +if __name__ == "__main__": record_get() diff --git a/src/hetzner_dns_tools/record_list.py b/src/hetzner_dns_tools/record_list.py index b386e23..589a41f 100644 --- a/src/hetzner_dns_tools/record_list.py +++ b/src/hetzner_dns_tools/record_list.py @@ -3,6 +3,7 @@ import json import os import sys + import requests from . import hetzner_dns_helpers as helpers @@ -31,29 +32,31 @@ def record_list(hetzner_dns_token=None, zone_id=None, zone_name=None): assigned in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(record_list.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] - if zone_id is None and os.environ.get('ZONE_ID'): + if zone_id is None and os.environ.get("ZONE_ID"): # get zone_id from environment variable - zone_id = os.environ['ZONE_ID'] + zone_id = os.environ["ZONE_ID"] - if zone_name is None and os.environ.get('ZONE_NAME'): + if zone_name is None and os.environ.get("ZONE_NAME"): # get zone_name from environment variable - zone_name = os.environ['ZONE_NAME'] + zone_name = os.environ["ZONE_NAME"] # if zone_name exists, use it to obtain zone (skip if zone_id exists) - if (zone_name or 'ZONE_NAME' in os.environ) and not zone_id: + if (zone_name or "ZONE_NAME" in os.environ) and not zone_id: try: # get the desired zone - response_dict = zone_get(zone_name=zone_name) + response_dict = zone_get( + hetzner_dns_token=hetzner_dns_token, zone_name=zone_name + ) except ValueError: # if no matching zone found, halt and notify of error helpers.exit_with_error("record not found") @@ -62,31 +65,33 @@ def record_list(hetzner_dns_token=None, zone_id=None, zone_name=None): helpers.check_response_for_errors(response_dict) # get the zone_id - zone = response_dict['zone'] - zone_id = zone['id'] + zone = response_dict["zone"] + zone_id = zone["id"] # get zone_id from environment variable if zone_id is None: - zone_id = os.environ.get('ZONE_ID') + zone_id = os.environ.get("ZONE_ID") try: # build params dict params = {} if zone_id: - params['zone_id'] = zone_id + params["zone_id"] = zone_id # get response - response = requests.get(url='https://dns.hetzner.com/api/v1/records', - params=params, - headers={'Auth-API-Token': hetzner_dns_token}) + response = requests.get( + url="https://dns.hetzner.com/api/v1/records", + params=params, + headers={"Auth-API-Token": hetzner_dns_token}, + ) - decoded_response = response.content.decode('utf-8') + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) # check response for errors helpers.check_response_for_errors(response_dict) - if __name__ == '__main__': + if __name__ == "__main__": # when running via the terminal, print output to console then exit print(decoded_response) sys.exit(0) # exit successfully @@ -97,5 +102,5 @@ def record_list(hetzner_dns_token=None, zone_id=None, zone_name=None): helpers.handle_request_exception(err) -if __name__ == '__main__': +if __name__ == "__main__": record_list() diff --git a/src/hetzner_dns_tools/record_update.py b/src/hetzner_dns_tools/record_update.py index 2d4e35e..636e2f8 100644 --- a/src/hetzner_dns_tools/record_update.py +++ b/src/hetzner_dns_tools/record_update.py @@ -2,24 +2,27 @@ import json import os -import requests import sys +import requests + from . import hetzner_dns_helpers as helpers -from .zone_list import zone_list from .record_list import record_list +from .zone_list import zone_list -def record_update(hetzner_dns_token=None, - record_type=None, - name=None, - record_id=None, - value=None, - ttl=None, - zone_id=None, - zone_name=None, - debug=0, - id_only=False): +def record_update( + hetzner_dns_token=None, + record_type=None, + name=None, + record_id=None, + value=None, + ttl=None, + zone_id=None, + zone_name=None, + debug=0, + id_only=False, +): """ Update a record. https://dns.hetzner.com/api-docs/#operation/UpdateRecord @@ -50,90 +53,93 @@ def record_update(hetzner_dns_token=None, in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(record_update.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] if record_type is None: # get record_type from environment variable - record_type = os.environ['RECORD_TYPE']\ - if os.environ.get('RECORD_TYPE') else os.environ['TYPE'] + record_type = ( + os.environ["RECORD_TYPE"] + if os.environ.get("RECORD_TYPE") + else os.environ["TYPE"] + ) if name is None: # get name from environment variable - name = os.environ['NAME'] if os.environ.get('NAME') else '@' + name = os.environ["NAME"] if os.environ.get("NAME") else "@" if value is None: # get value from environment variable - value = os.environ['VALUE'] + value = os.environ["VALUE"] if ttl is None: - if os.environ.get('TTL'): + if os.environ.get("TTL"): # get ttl from environment variable - ttl = int(os.environ['TTL']) + ttl = int(os.environ["TTL"]) else: # use default value for TTL ttl = 86400 - if zone_name is None and os.environ.get('ZONE_NAME'): + if zone_name is None and os.environ.get("ZONE_NAME"): # get zone name from environment variable - zone_name = os.environ['ZONE_NAME'] + zone_name = os.environ["ZONE_NAME"] - if name is None and os.environ.get('NAME'): + if name is None and os.environ.get("NAME"): # get record name from environment variable - name = os.environ['NAME'] + name = os.environ["NAME"] - if debug == 0 and os.environ.get('DEBUG'): + if debug == 0 and os.environ.get("DEBUG"): # get debug from environment variable - debug = int(os.environ['DEBUG']) + debug = int(os.environ["DEBUG"]) # if zone_name exists, use it to obtain the zone_id if zone_name: # get list of zones - response_dict = zone_list() + response_dict = zone_list(hetzner_dns_token=hetzner_dns_token) # check response for errors helpers.check_response_for_errors(response_dict) # check for matching zone - dns_zones = response_dict['zones'] + dns_zones = response_dict["zones"] for zone in dns_zones: - if zone['name'] == zone_name: - zone_id = zone['id'] + if zone["name"] == zone_name: + zone_id = zone["id"] break # if no matching zone found, then exit with error if zone_id is None: helpers.exit_with_error("zone not found") - if not zone_name and not zone_id and not os.environ.get('ZONE_ID'): + if not zone_name and not zone_id and not os.environ.get("ZONE_ID"): # if neither zone_name or zone_id exist, then exit with error helpers.exit_with_error("Must include one of: zone_id, zone_name") if zone_id is None: # get zone_id from environment variable - zone_id = os.environ['ZONE_ID'] + zone_id = os.environ["ZONE_ID"] # if name exists, use it to obtain the record_id if name: # get list of records - response_dict = record_list() + response_dict = record_list(hetzner_dns_token=hetzner_dns_token) # check response for errors helpers.check_response_for_errors(response_dict) # check for matching record in zone - dns_records = response_dict['records'] + dns_records = response_dict["records"] record_id_count = 0 for record in dns_records: - if record['name'] == name and record['type'] == record_type: - record_id = record['id'] + if record["name"] == name and record["type"] == record_type: + record_id = record["id"] record_id_count += 1 # if no matching name found, then exit with error @@ -142,33 +148,41 @@ def record_update(hetzner_dns_token=None, # if no matching name found, then exit with error if record_id_count > 1: - helpers.exit_with_error("more than one record found for name and type, record_id must be provided") + helpers.exit_with_error( + "more than one record found for name and type, record_id must be provided" + ) if record_id is None: # get record_id from environment variable - record_id = os.environ['RECORD_ID'] + record_id = os.environ["RECORD_ID"] if record_id is None: # if record_id exist, then exit with error helpers.exit_with_error("Must include (or able to retrieve): record_id") try: - params = {'ttl': ttl, - 'type': record_type, - 'value': value, - 'name': name, - 'zone_id': zone_id} + params = { + "ttl": ttl, + "type": record_type, + "value": value, + "name": name, + "zone_id": zone_id, + } if debug > 0: print("DEBUG : request: record_id=%s" % record_id, file=sys.stderr) print(json.dumps(params), file=sys.stderr) - response = requests.put(url='https://dns.hetzner.com/api/v1/records/' + record_id, - headers={'Content-Type': 'application/json', - 'Auth-API-Token': hetzner_dns_token}, - data=json.dumps(params)) + response = requests.put( + url="https://dns.hetzner.com/api/v1/records/" + record_id, + headers={ + "Content-Type": "application/json", + "Auth-API-Token": hetzner_dns_token, + }, + data=json.dumps(params), + ) - decoded_response = response.content.decode('utf-8') + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) if debug > 0: print("DEBUG : response:", file=sys.stderr) @@ -178,7 +192,7 @@ def record_update(hetzner_dns_token=None, helpers.check_response_for_errors(response_dict) # return all zone data - if __name__ == '__main__': + if __name__ == "__main__": print(json.dumps(response_dict)) sys.exit(0) # exit successfully @@ -188,5 +202,5 @@ def record_update(hetzner_dns_token=None, helpers.handle_request_exception(err) -if __name__ == '__main__': +if __name__ == "__main__": record_update() diff --git a/src/hetzner_dns_tools/zone_create.py b/src/hetzner_dns_tools/zone_create.py index abab3e0..f0ecfd4 100644 --- a/src/hetzner_dns_tools/zone_create.py +++ b/src/hetzner_dns_tools/zone_create.py @@ -3,16 +3,15 @@ import json import os import sys + import requests from . import hetzner_dns_helpers as helpers -def zone_create(hetzner_dns_token=None, - name=None, - id_only=False, - ttl=None, - zone_name=None): +def zone_create( + hetzner_dns_token=None, name=None, id_only=False, ttl=None, zone_name=None +): """ Create a new zone. https://dns.hetzner.com/api-docs/#operation/CreateZone @@ -39,58 +38,64 @@ def zone_create(hetzner_dns_token=None, in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(zone_create.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] if name is None: # get domain name from environment variable - name = os.environ['ZONE_NAME']\ - if os.environ.get('ZONE_NAME') else os.environ['NAME'] + name = ( + os.environ["ZONE_NAME"] + if os.environ.get("ZONE_NAME") + else os.environ["NAME"] + ) elif zone_name: # allow zone_name and name to be used interchangeably # for zone_create and zone_delete name = zone_name if ttl is None: - if os.environ.get('TTL'): + if os.environ.get("TTL"): # get TTL from environment variable - ttl = int(os.environ['TTL']) + ttl = int(os.environ["TTL"]) else: # use default value for TTL ttl = 86400 try: - response = requests.post(url='https://dns.hetzner.com/api/v1/zones', - headers={'Content-Type': 'application/json', - 'Auth-API-Token': hetzner_dns_token}, - data=json.dumps({'name': name, - 'ttl': ttl})) - - decoded_response = response.content.decode('utf-8') + response = requests.post( + url="https://dns.hetzner.com/api/v1/zones", + headers={ + "Content-Type": "application/json", + "Auth-API-Token": hetzner_dns_token, + }, + data=json.dumps({"name": name, "ttl": ttl}), + ) + + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) # check response for errors helpers.check_response_for_errors(response_dict) # return the expected value - if id_only or os.environ.get('ID_ONLY') == '1': + if id_only or os.environ.get("ID_ONLY") == "1": # return the zone_id - result = response_dict['zone']['id'] + result = response_dict["zone"]["id"] - if __name__ == '__main__': + if __name__ == "__main__": print(result) sys.exit(0) # exit successfully return result else: # return all zone data - if __name__ == '__main__': + if __name__ == "__main__": print(json.dumps(response_dict)) sys.exit(0) # exit successfully @@ -100,5 +105,5 @@ def zone_create(hetzner_dns_token=None, helpers.handle_request_exception(err) -if __name__ == '__main__': +if __name__ == "__main__": zone_create() diff --git a/src/hetzner_dns_tools/zone_delete.py b/src/hetzner_dns_tools/zone_delete.py index 1c7c058..54b6356 100644 --- a/src/hetzner_dns_tools/zone_delete.py +++ b/src/hetzner_dns_tools/zone_delete.py @@ -3,16 +3,14 @@ import json import os import sys + import requests from . import hetzner_dns_helpers as helpers from .zone_list import zone_list -def zone_delete(hetzner_dns_token=None, - zone_id=None, - zone_name=None, - name=None): +def zone_delete(hetzner_dns_token=None, zone_id=None, zone_name=None, name=None): """ Delete an existing zone. @@ -32,41 +30,44 @@ def zone_delete(hetzner_dns_token=None, in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(zone_delete.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] if zone_name is None: # get name from environment variable - zone_name = os.environ['NAME']\ - if os.environ.get('NAME') else os.environ.get('ZONE_NAME', None) + zone_name = ( + os.environ["NAME"] + if os.environ.get("NAME") + else os.environ.get("ZONE_NAME", None) + ) elif name: # allow zone_name and name to be used interchangeably # for zone_create and zone_delete zone_name = name - if os.environ.get('ZONE_ID'): - zone_id = os.environ['ZONE_ID'] + if os.environ.get("ZONE_ID"): + zone_id = os.environ["ZONE_ID"] # if zone_name exists, use it to obtain zone (skip if zone_id exists) if zone_name and not zone_id: # get list of zones - response_dict = zone_list() + response_dict = zone_list(hetzner_dns_token=hetzner_dns_token) # check response for errors helpers.check_response_for_errors(response_dict) # check for matching zone - dns_zones = response_dict['zones'] + dns_zones = response_dict["zones"] for zone in dns_zones: - if zone['name'] == zone_name: - zone_id = zone['id'] + if zone["name"] == zone_name: + zone_id = zone["id"] break # if no matching zone found, halt and notify of error @@ -79,21 +80,22 @@ def zone_delete(hetzner_dns_token=None, if zone_id is None: # get zone_id from environment variable - zone_id = os.environ['ZONE_ID'] + zone_id = os.environ["ZONE_ID"] try: response = requests.delete( - url=f'https://dns.hetzner.com/api/v1/zones/{zone_id}', - headers={'Auth-API-Token': hetzner_dns_token}) + url=f"https://dns.hetzner.com/api/v1/zones/{zone_id}", + headers={"Auth-API-Token": hetzner_dns_token}, + ) - decoded_response = response.content.decode('utf-8') + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) # check response for errors helpers.check_response_for_errors(response_dict) # when running via the terminal, print output to console - if __name__ == '__main__': + if __name__ == "__main__": print("OK") sys.exit(0) # exit successfully @@ -103,5 +105,5 @@ def zone_delete(hetzner_dns_token=None, helpers.handle_request_exception(err) -if __name__ == '__main__': +if __name__ == "__main__": zone_delete() diff --git a/src/hetzner_dns_tools/zone_get.py b/src/hetzner_dns_tools/zone_get.py index bbaad94..4c4804a 100644 --- a/src/hetzner_dns_tools/zone_get.py +++ b/src/hetzner_dns_tools/zone_get.py @@ -2,19 +2,17 @@ import json import os -import requests import sys -from . import hetzner_dns_helpers as helpers +import requests +from . import hetzner_dns_helpers as helpers from .zone_list import zone_list -def zone_get(hetzner_dns_token=None, - zone_id=None, - name=None, - zone_name=None, - id_only=False): +def zone_get( + hetzner_dns_token=None, zone_id=None, name=None, zone_name=None, id_only=False +): """ Get info about an existing zone. https://dns.hetzner.com/api-docs/#operation/GetZone @@ -44,49 +42,53 @@ def zone_get(hetzner_dns_token=None, in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(zone_get.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] - if (name is None and os.environ.get('NAME'))\ - or (zone_name is None and os.environ.get('ZONE_NAME')): + if (name is None and os.environ.get("NAME")) or ( + zone_name is None and os.environ.get("ZONE_NAME") + ): # get name/zone_name from environment variable - zone_name = os.environ['ZONE_NAME']\ - if os.environ.get('ZONE_NAME') else os.environ.get('NAME') + zone_name = ( + os.environ["ZONE_NAME"] + if os.environ.get("ZONE_NAME") + else os.environ.get("NAME") + ) - if not id_only and os.environ.get('ID_ONLY'): + if not id_only and os.environ.get("ID_ONLY"): # get id_only from environment variable - id_only = os.environ['ID_ONLY'] + id_only = os.environ["ID_ONLY"] zone = None # if zone_name exists, use it to obtain the zone (skip if zone_id exists) - if (zone_name or 'ZONE_NAME' in os.environ) and zone_id is None: + if (zone_name or "ZONE_NAME" in os.environ) and zone_id is None: # get list of zones - response_dict = zone_list() + response_dict = zone_list(hetzner_dns_token=hetzner_dns_token) # check response for errors helpers.check_response_for_errors(response_dict) # check for matching zone - dns_zones = response_dict['zones'] + dns_zones = response_dict["zones"] for dns_zone in dns_zones: - if dns_zone['name'] == zone_name: - zone = {'zone': dns_zone} - zone_id = zone['zone']['id'] + if dns_zone["name"] == zone_name: + zone = {"zone": dns_zone} + zone_id = zone["zone"]["id"] break # if no matching zone found, halt and notify of error if zone is None: error_message = "zone not found" - if __name__ == '__main__': + if __name__ == "__main__": print(f"Error: {error_message}") sys.exit(1) # exit with error else: @@ -94,7 +96,7 @@ def zone_get(hetzner_dns_token=None, # if we don't have a zone_id by now, then get it from environment variable if zone_id is None: - zone_id = os.environ.get('ZONE_ID') + zone_id = os.environ.get("ZONE_ID") # raise an exception if no zone_id or zone_name have been passed if not zone_id and not zone_name: @@ -103,11 +105,14 @@ def zone_get(hetzner_dns_token=None, # get response try: response = requests.get( - url=f'https://dns.hetzner.com/api/v1/zones/{zone_id}', - headers={'Auth-API-Token': hetzner_dns_token, - 'Content-Type': 'application/json; charset=utf-8'}) - - decoded_response = response.content.decode('utf-8') + url=f"https://dns.hetzner.com/api/v1/zones/{zone_id}", + headers={ + "Auth-API-Token": hetzner_dns_token, + "Content-Type": "application/json; charset=utf-8", + }, + ) + + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) # check response for errors @@ -122,21 +127,21 @@ def zone_get(hetzner_dns_token=None, # return the expected zone ID or zone if id_only: # return the zone_id - zone_id = zone['zone']['id'] + zone_id = zone["zone"]["id"] - if __name__ == '__main__': + if __name__ == "__main__": print(zone_id) sys.exit(0) # exit successfully return zone_id else: # return all zone data - if __name__ == '__main__': + if __name__ == "__main__": print(json.dumps(zone)) sys.exit(0) # exit successfully return zone -if __name__ == '__main__': +if __name__ == "__main__": zone_get() diff --git a/src/hetzner_dns_tools/zone_list.py b/src/hetzner_dns_tools/zone_list.py index 775b8c9..f9b8f96 100644 --- a/src/hetzner_dns_tools/zone_list.py +++ b/src/hetzner_dns_tools/zone_list.py @@ -3,6 +3,7 @@ import json import os import sys + import requests from . import hetzner_dns_helpers as helpers @@ -22,27 +23,29 @@ def zone_list(hetzner_dns_token=None): in ALL_CAPS. - e.g. zone_id in Python -> ZONE_ID in environment variable """ - if os.environ.get('SHOW_HELP'): + if os.environ.get("SHOW_HELP"): # print the docstring and exit print(zone_list.__doc__) sys.exit(0) if hetzner_dns_token is None: # get token from environment variable - hetzner_dns_token = os.environ['HETZNER_DNS_TOKEN'] + hetzner_dns_token = os.environ["HETZNER_DNS_TOKEN"] # get response try: - response = requests.get(url='https://dns.hetzner.com/api/v1/zones', - headers={'Auth-API-Token': hetzner_dns_token}) + response = requests.get( + url="https://dns.hetzner.com/api/v1/zones", + headers={"Auth-API-Token": hetzner_dns_token}, + ) - decoded_response = response.content.decode('utf-8') + decoded_response = response.content.decode("utf-8") response_dict = json.loads(decoded_response) # check response for errors helpers.check_response_for_errors(response_dict) - if __name__ == '__main__': + if __name__ == "__main__": # when running via the terminal, print output to console print(decoded_response) sys.exit(0) # exit successfully @@ -53,5 +56,5 @@ def zone_list(hetzner_dns_token=None): helpers.handle_request_exception(err) -if __name__ == '__main__': +if __name__ == "__main__": zone_list()