A function to update DynamoDB Route 53 with your public ip address

public_ip_update_domains.py

#pylint: disable=broad-except,literal-comparison
"""
A function to grab the public ip and update a set of domains in Route 53
"""
import argparse
import logging
from requests import get
from boto3 import client


class AwsIPUpdate:
    """
    A class to update domains with an IP
    """
    dns_client = client('route53')
    zone_id = None
    def __init__(self, domains: [str], ip_address: str):
        self.domains = domains
        self.ip_address = ip_address
        try:
            self.zone_id = self.dns_client.list_hosted_zones()['HostedZones'][0]['Id']
            logging.debug("Zone ID: %s", self.zone_id)
        except Exception as exc:
            logging.error(exc)
            raise Exception("No domains exist")

    def get_a_record(self, domain: str) -> str:
        """
        A function to get an individual record set
        :param
        """
        return self.dns_client.list_resource_record_sets( \
                    HostedZoneId=self.zone_id, \
                    StartRecordName=domain, \
                    StartRecordType='A', \
                    MaxItems='1')['ResourceRecordSets'][0]['ResourceRecords'][0]['Value']

    def update(self):
        """
        A wrapper function to get an a record for every domain and update the domain
        """
        for domain in self.domains:
            match = False
            try:
                logging.debug("Checking for A record match for %s", domain)
                current: str = self.get_a_record(domain)
                logging.debug("Current ip is %s", current)
                match = (self.ip_address == current)
            except Exception as exc:
                logging.error(exc)
                logging.debug("No resource record exists")

            if not match:
                self.__update_record(domain)
            else:
                logging.debug("Domain %s is already set for %s", domain, self.ip_address)

    def __update_record(self, domain: str):
        """ A private function to update ips in route 53
        :param ip the ip for the A record
        :param domain the domain for the A record
        """
        logging.info("Updating record for %s: to %s", domain, self.ip_address)
        record = {
            "Comment": "Automatic DNS update",
            'Changes': [
                {
                    'Action': 'UPSERT',
                    'ResourceRecordSet': {
                        'Name': domain,
                        'Type': 'A',
                        'TTL': 300,
                        'ResourceRecords': [
                            {
                                'Value': self.ip_address
                            },
                        ]}
                    },
                ]
        }
        logging.debug(record)
        response = self.dns_client.change_resource_record_sets(
            HostedZoneId=self.zone_id,
            ChangeBatch=record
        )
        if response['ResponseMetadata']['HTTPStatusCode'] is 200:
            logging.info("update success")

def main(domains: [str]):
    """
    The main function that instantiates the AWSIPUpdate class
    :param domains a list of domains in which to update
    """
    ip_address = get('https://api.ipify.org').text
    AwsIPUpdate(domains, ip_address).update()

def set_logging(level):
    """ A function to set the logging level
    :param level: Log level
    """
    logging.basicConfig(format='updateip: %(asctime)s %(message)s', level=level)
    logging.getLogger("requests").setLevel(logging.WARNING)
    logging.getLogger('boto3').setLevel(logging.WARNING)
    logging.getLogger('botocore').setLevel(logging.CRITICAL)
    logging.getLogger('urllib3').setLevel(logging.CRITICAL)


if __name__ == '__main__':
    PARSER = argparse.ArgumentParser()
    PARSER.add_argument('--domains', \
        dest='domains', \
        nargs='+',  \
        default=[],  \
        help='List of domains to update ip for')
    PARSER.add_argument('--debug', \
    action='store_true', \
    default=False, \
    dest='verbose_logging', \
    help='Debug output')
    RESULTS = PARSER.parse_args()

    if RESULTS.verbose_logging:
        set_logging(logging.DEBUG)
    else:
        set_logging(logging.INFO)

    main(RESULTS.domains)