<commit_before>"""Provide support to Lexicon for AWS Route 53 DNS changes."""
from base import Provider as BaseProvider
import boto3
import botocore


def ProviderParser(subparser):
    """Specify arguments for AWS Route 53 Lexicon Provider."""
    pass


class RecordSetPaginator(object):
    """Paginate through complete list of record sets."""

    def __init__(self, r53_client, hosted_zone_id, max_items=None):
        """Initialize paginator."""
        self.r53_client = r53_client
        self.hosted_zone_id = hosted_zone_id
        self.max_items = max_items

    def get_record_sets(self, **kwargs):
        """Retrieve a page from API."""
        return self.r53_client.list_resource_record_sets(**kwargs)

    def get_base_kwargs(self):
        """Get base kwargs for API call."""
        kwargs = {
            'HostedZoneId': self.hosted_zone_id
        }
        if self.max_items is not None:
            kwargs.update({
                'MaxItems': str(self.max_items)
            })
        return kwargs

    def all_record_sets(self):
        """Generator to loop through current record set.

        Call next page if it exists.
        """
        is_truncated = True
        start_record_name = None
        start_record_type = None
        kwargs = self.get_base_kwargs()
        while is_truncated:
            if start_record_name is not None:
                kwargs.update({
                    'StartRecordName': start_record_name,
                    'StartRecordType': start_record_type
                })
            result = self.get_record_sets(**kwargs)
            for record_set in result.get('ResourceRecordSets', []):
                yield record_set

            is_truncated = result.get('IsTruncated', False)

            start_record_name = result.get('NextRecordName', None)
            start_record_type = result.get('NextRecordType', None)


class Provider(BaseProvider):
    """Provide AWS Route 53 implementation of Lexicon Provider interface."""

    def __init__(self, options, provider_options={}):
        """Initialize AWS Route 53 DNS provider."""
        super(Provider, self).__init__(options)
        self.domain_id = None
        self.r53_client = boto3.client('route53')<commit_msg>catch no credentials error to let tests run<commit_after>"""Provide support to Lexicon for AWS Route 53 DNS changes."""
from base import Provider as BaseProvider
import boto3
import botocore


def ProviderParser(subparser):
    """Specify arguments for AWS Route 53 Lexicon Provider."""
    pass


class RecordSetPaginator(object):
    """Paginate through complete list of record sets."""

    def __init__(self, r53_client, hosted_zone_id, max_items=None):
        """Initialize paginator."""
        self.r53_client = r53_client
        self.hosted_zone_id = hosted_zone_id
        self.max_items = max_items

    def get_record_sets(self, **kwargs):
        """Retrieve a page from API."""
        return self.r53_client.list_resource_record_sets(**kwargs)

    def get_base_kwargs(self):
        """Get base kwargs for API call."""
        kwargs = {
            'HostedZoneId': self.hosted_zone_id
        }
        if self.max_items is not None:
            kwargs.update({
                'MaxItems': str(self.max_items)
            })
        return kwargs

    def all_record_sets(self):
        """Generator to loop through current record set.

        Call next page if it exists.
        """
        is_truncated = True
        start_record_name = None
        start_record_type = None
        kwargs = self.get_base_kwargs()
        while is_truncated:
            if start_record_name is not None:
                kwargs.update({
                    'StartRecordName': start_record_name,
                    'StartRecordType': start_record_type
                })
            result = self.get_record_sets(**kwargs)
            for record_set in result.get('ResourceRecordSets', []):
                yield record_set

            is_truncated = result.get('IsTruncated', False)

            start_record_name = result.get('NextRecordName', None)
            start_record_type = result.get('NextRecordType', None)


class Provider(BaseProvider):
    """Provide AWS Route 53 implementation of Lexicon Provider interface."""

    def __init__(self, options, provider_options={}):
        """Initialize AWS Route 53 DNS provider."""
        super(Provider, self).__init__(options)
        self.domain_id = None
        try:
            self.r53_client = boto3.client('route53')
        except botocore.exceptions.NoCredentialsError:
            # instantiate the client anyway for tests
            print 'no credentials found - using dummy credentials for tests'
            self.r53_client = boto3.client(
                'route53',
                aws_access_key_id='DUMMY',
                aws_secret_access_key='DUMMY'
            )
{'fraction_non_alphanumeric': 0.08942091262533251, 'fraction_numerical': 0.01002660118682218, 'mean_word_length': 2.3206521739130435, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '897766', 'n_tokens_mistral': 1382, 'n_tokens_neox': 1302, 'n_words': 355}