Source code for libcloud.dns.drivers.gandi

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.

from libcloud.dns.base import Zone, Record, DNSDriver
from libcloud.dns.types import (
from libcloud.common.gandi import GandiResponse, BaseGandiDriver, GandiConnection

__all__ = ["GandiDNSDriver"]

TTL_MIN = 30
TTL_MAX = 2592000  # 30 days

class NewZoneVersion:
    Changes to a zone in the Gandi DNS service need to be wrapped in a new
    version object. The changes are made to the new version, then that
    version is made active.

    In effect, this is a transaction.

    Any calls made inside this context manager will be applied to a new version
    id. If your changes are successful (and only if they are successful) they
    are activated.

    def __init__(self, driver, zone):
        self.driver = driver
        self.connection = driver.connection = zone

    def __enter__(self):
        zid = int(
        vid = self.connection.request("", zid).object
        self.vid = vid
        return vid

    def __exit__(self, type, value, traceback):
        if not traceback:
            zid = int(
            con = self.connection
            con.request("", zid, self.vid).object

class GandiDNSResponse(GandiResponse):
    exceptions = {
        581042: ZoneDoesNotExistError,

class GandiDNSConnection(GandiConnection):
    responseCls = GandiDNSResponse

[docs]class GandiDNSDriver(BaseGandiDriver, DNSDriver): """ API reference can be found at: """ type = Provider.GANDI name = "Gandi DNS" website = "" connectionCls = GandiDNSConnection RECORD_TYPE_MAP = { RecordType.A: "A", RecordType.AAAA: "AAAA", RecordType.CNAME: "CNAME", RecordType.LOC: "LOC", RecordType.MX: "MX", RecordType.NS: "NS", RecordType.SPF: "SPF", RecordType.SRV: "SRV", RecordType.TXT: "TXT", RecordType.WKS: "WKS", } def _to_zone(self, zone): return Zone( id=str(zone["id"]), domain=zone["name"], type="master", ttl=0, driver=self, extra={}, ) def _to_zones(self, zones): ret = [] for z in zones: ret.append(self._to_zone(z)) return ret
[docs] def list_zones(self): zones = self.connection.request("") return self._to_zones(zones.object)
[docs] def get_zone(self, zone_id): zid = int(zone_id) self.connection.set_context({"zone_id": zone_id}) zone = self.connection.request("", zid) return self._to_zone(zone.object)
[docs] def create_zone(self, domain, type="master", ttl=None, extra=None): params = { "name": domain, } info = self.connection.request("", params) return self._to_zone(info.object)
[docs] def update_zone(self, zone, domain=None, type=None, ttl=None, extra=None): zid = int( params = {"name": domain} self.connection.set_context({"zone_id":}) zone = self.connection.request("", zid, params) return self._to_zone(zone.object)
[docs] def delete_zone(self, zone): zid = int( self.connection.set_context({"zone_id":}) res = self.connection.request("", zid) return res.object
def _to_record(self, record, zone): extra = {"ttl": int(record["ttl"])} value = record["value"] if record["type"] == "MX": # Record is in the following form: # <priority> <value> # e.g. 15 split = record["value"].split(" ") extra["priority"] = int(split[0]) value = split[1] return Record( id="{}:{}".format(record["type"], record["name"]), name=record["name"], type=self._string_to_record_type(record["type"]), data=value, zone=zone, driver=self, ttl=record["ttl"], extra=extra, ) def _to_records(self, records, zone): retval = [] for r in records: retval.append(self._to_record(r, zone)) return retval
[docs] def list_records(self, zone): zid = int( self.connection.set_context({"zone_id":}) records = self.connection.request("", zid, 0) return self._to_records(records.object, zone)
[docs] def get_record(self, zone_id, record_id): zid = int(zone_id) record_type, name = record_id.split(":", 1) filter_opts = {"name": name, "type": record_type} self.connection.set_context({"zone_id": zone_id}) records = self.connection.request("", zid, 0, filter_opts).object if len(records) == 0: raise RecordDoesNotExistError(value="", driver=self, record_id=record_id) return self._to_record(records[0], self.get_zone(zone_id))
def _validate_record(self, record_id, name, record_type, data, extra): if len(data) > 1024: raise RecordError( "Record data must be <= 1024 characters", driver=self, record_id=record_id, ) if extra and "ttl" in extra: if extra["ttl"] < TTL_MIN: raise RecordError( "TTL must be at least 30 seconds", driver=self, record_id=record_id ) if extra["ttl"] > TTL_MAX: raise RecordError("TTL must not excdeed 30 days", driver=self, record_id=record_id)
[docs] def create_record(self, name, zone, type, data, extra=None): self._validate_record(None, name, type, data, extra) zid = int( create = {"name": name, "type": self.RECORD_TYPE_MAP[type], "value": data} if "ttl" in extra: create["ttl"] = extra["ttl"] with NewZoneVersion(self, zone) as vid: con = self.connection con.set_context({"zone_id":}) rec = con.request("", zid, vid, create).object return self._to_record(rec, zone)
[docs] def update_record(self, record, name, type, data, extra): self._validate_record(, name, type, data, extra) filter_opts = {"name":, "type": self.RECORD_TYPE_MAP[record.type]} update = {"name": name, "type": self.RECORD_TYPE_MAP[type], "value": data} if "ttl" in extra: update["ttl"] = extra["ttl"] zid = int( with NewZoneVersion(self, as vid: con = self.connection con.set_context({"zone_id":}) con.request("", zid, vid, filter_opts) res = con.request("", zid, vid, update).object return self._to_record(res,
[docs] def delete_record(self, record): zid = int( filter_opts = {"name":, "type": self.RECORD_TYPE_MAP[record.type]} with NewZoneVersion(self, as vid: con = self.connection con.set_context({"zone_id":}) count = con.request("", zid, vid, filter_opts).object if count == 1: return True raise RecordDoesNotExistError(value="No such record", driver=self,