Source code for gordon_gcp.clients.gdns

# -*- coding: utf-8 -*-
# Copyright 2018 Spotify AB
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
Client module to interact with the Google Cloud DNS API.

This client makes use of the asynchronus HTTP client as defined in
:class:`.AIOConnection`, and therefore must use service account/JWT
authentication (for now).

To use:

.. code-block:: python

    import asyncio
    import gordon_gcp

    keyfile = '/path/to/service_account_keyfile.json'
    auth_client = gordon_gcp.GAuthClient(keyfile=keyfile)
    client = gordon_gcp.GDNSClient(
        project='my-dns-project', auth_client=auth_client)

    async def print_first_record(client)
        records = await client.get_records_for_zone('testzone')

    loop = asyncio.get_event_loop()

import json
import logging

from gordon_gcp.clients import http

__all__ = ('GDNSClient',)

[docs]class GDNSClient(http.AIOConnection): """Async HTTP client to interact with Google Cloud DNS API. Attributes: BASE_URL (str): base call url for the DNS API Args: project (str): Google project ID that hosts the managed DNS. auth_client (.GAuthClient): client to manage authentication for HTTP API requests. api_version (str): DNS API endpoint version. Defaults to ``v1``. session (aiohttp.ClientSession): (optional) ``aiohttp`` HTTP session to use for sending requests. Defaults to the session object attached to :obj:`auth_client` if not provided. """ BASE_URL = '' # see DNS_CHANGES_DONE = 'done' REVERSE_PREFIX = 'reverse-' def __init__(self, project=None, auth_client=None, api_version='v1', session=None, default_zone_prefix=None): super().__init__(auth_client=auth_client, session=session) self.project = project self._base_url = f'{self.BASE_URL}/{api_version}/projects/{project}' prefix = f'{default_zone_prefix}-' if default_zone_prefix else '' self.forward_prefix = f'{prefix}' self.reverse_prefix = f'{prefix}{self.REVERSE_PREFIX}'
[docs] def get_managed_zone(self, zone): """Get the GDNS managed zone name for a DNS zone. Google uses custom string names with specific `requirements <>`_ for storing records. The scheme implemented here chooses a managed zone name which removes the trailing dot and replaces other dots with dashes, and in the case of reverse records, uses only the two most significant octets, prepended with 'reverse'. At least two octets are required for reverse DNS zones. Example: get_managed_zone('') = 'example-com' get_managed_zone(' = 'reverse-20-10' get_managed_zone(' = 'reverse-20-10' get_managed_zone(' = 'reverse-20-10' Args: zone (str): DNS zone. Returns: str of managed zone name. """ if zone.endswith(''): return self.reverse_prefix + '-'.join(zone.split('.')[-5:-3]) return self.forward_prefix + '-'.join(zone.split('.')[:-1])
[docs] async def get_records_for_zone(self, dns_zone, params=None): """Get all resource record sets for a managed zone, using the DNS zone. Args: dns_zone (str): Desired DNS zone to query. params (dict): (optional) Additional query parameters for HTTP requests to the GDNS API. Returns: list of dicts representing rrsets. """ managed_zone = self.get_managed_zone(dns_zone) url = f'{self._base_url}/managedZones/{managed_zone}/rrsets' if not params: params = {} if 'fields' not in params: # Get only the fields we care about params['fields'] = ('rrsets/name,rrsets/kind,rrsets/rrdatas,' 'rrsets/type,rrsets/ttl,nextPageToken') next_page_token = None records = [] while True: if next_page_token: params['pageToken'] = next_page_token response = await self.get_json(url, params=params) records.extend(response['rrsets']) next_page_token = response.get('nextPageToken') if not next_page_token: break'Found {len(records)} rrsets for zone "{dns_zone}".') return records
[docs] async def is_change_done(self, zone, change_id): """Check if a DNS change has completed. Args: zone (str): DNS zone of the change. change_id (str): Identifier of the change. Returns: Boolean """ zone_id = self.get_managed_zone(zone) url = f'{self._base_url}/managedZones/{zone_id}/changes/{change_id}' resp = await self.get_json(url) return resp['status'] == self.DNS_CHANGES_DONE
[docs] async def publish_changes(self, zone, changes): """Post changes to a zone. Args: zone (str): DNS zone of the change. changes (dict): JSON compatible dict of a `Change <>`_. Returns: string identifier of the change. """ zone_id = self.get_managed_zone(zone) url = f'{self._base_url}/managedZones/{zone_id}/changes' resp = await self.request('post', url, json=changes) return json.loads(resp)['id']