First basic functionnal version
This commit is contained in:
Hugo Levy-Falk
commit
6353a70fbe
8 changed files with 695 additions and 0 deletions
349
gandi.py
Normal file
349
gandi.py
Normal file
|
|
@ -0,0 +1,349 @@
|
|||
import logging
|
||||
|
||||
import requests
|
||||
|
||||
API_URL = "https://api.gandi.net/v5/"
|
||||
|
||||
|
||||
class GandiAPIClient:
|
||||
"""Wrapper for the Gandi API client.
|
||||
|
||||
It handles Gandi API over https for LiveDNS. Logging is done through the
|
||||
`gandiAPI` logger. See `logging` package documentation for how to configure
|
||||
it.
|
||||
"""
|
||||
|
||||
def __init__(self, api_key):
|
||||
"""Initializes the API client.
|
||||
|
||||
Args:
|
||||
api_key: the API key as provided by Gandi.
|
||||
log_level: logging level (see `logging` package documentation)
|
||||
"""
|
||||
self.logger = logging.getLogger("gandiAPI")
|
||||
self.api_key = api_key
|
||||
|
||||
def _request(self, method, url, headers={}, params={}, *args, **kwargs):
|
||||
self.logger.debug(
|
||||
"Requesting %s: %s, headers=%r, params=%r, args=%r, kwargs=%r", method, url, headers, params, args, kwargs
|
||||
)
|
||||
|
||||
headers.update({"Authorization": "Apikey %s" % self.api_key})
|
||||
response = getattr(requests, method)(
|
||||
url, headers=headers, params=params, allow_redirects=False, *args, **kwargs
|
||||
)
|
||||
|
||||
self.logger.debug("Got response %r.", response)
|
||||
response.raise_for_status()
|
||||
try:
|
||||
ret = response.json()
|
||||
except json.decoder.JSONDecodeError as e:
|
||||
ret = None
|
||||
self.logger.debug("No valid JSON in the response.")
|
||||
self.logger.debug("Request {} {} successful.".format(method, response.url))
|
||||
return ret
|
||||
|
||||
def delete(self, *args, **kwargs):
|
||||
"""Performs a DELETE request.
|
||||
|
||||
DELETE request on a given URL that acts like `requests.delete` except
|
||||
that authentication to the API is automatically done and JSON response
|
||||
is decoded.
|
||||
|
||||
Args:
|
||||
url: The URL of the requests.
|
||||
*args: See `requests.delete` parameters.
|
||||
**kwargs: See `requests.delete` parameters.
|
||||
|
||||
Returns:
|
||||
The JSON-decoded result of the request.
|
||||
|
||||
Raises:
|
||||
requests.exceptions.RequestException: An error occured while
|
||||
performing the request.
|
||||
exceptions.PermissionDenied: The user does not have the right
|
||||
to perform this request.
|
||||
"""
|
||||
return self._request("delete", *args, **kwargs)
|
||||
|
||||
def get(self, *args, **kwargs):
|
||||
"""Performs a GET request.
|
||||
|
||||
GET request on a given URL that acts like `requests.get` except that
|
||||
authentication to the API is automatically done and JSON response is
|
||||
decoded.
|
||||
|
||||
Args:
|
||||
url: The URL of the requests.
|
||||
*args: See `requests.get` parameters.
|
||||
**kwargs: See `requests.get` parameters.
|
||||
|
||||
Returns:
|
||||
The JSON-decoded result of the request.
|
||||
|
||||
Raises:
|
||||
requests.exceptions.RequestException: An error occured while
|
||||
performing the request.
|
||||
exceptions.PermissionDenied: The user does not have the right
|
||||
to perform this request.
|
||||
"""
|
||||
return self._request("get", *args, **kwargs)
|
||||
|
||||
def head(self, *args, **kwargs):
|
||||
"""Performs a HEAD request.
|
||||
|
||||
HEAD request on a given URL that acts like `requests.head` except that
|
||||
authentication to the API is automatically done and JSON response is
|
||||
decoded.
|
||||
|
||||
Args:
|
||||
url: The URL of the requests.
|
||||
*args: See `requests.head` parameters.
|
||||
**kwargs: See `requests.head` parameters.
|
||||
|
||||
Returns:
|
||||
The JSON-decoded result of the request.
|
||||
|
||||
Raises:
|
||||
requests.exceptions.RequestException: An error occured while
|
||||
performing the request.
|
||||
exceptions.PermissionDenied: The user does not have the right
|
||||
to perform this request.
|
||||
"""
|
||||
return self._request("get", *args, **kwargs)
|
||||
|
||||
def option(self, *args, **kwargs):
|
||||
"""Performs a OPTION request.
|
||||
|
||||
OPTION request on a given URL that acts like `requests.option` except
|
||||
that authentication to the API is automatically done and JSON response
|
||||
is decoded.
|
||||
|
||||
Args:
|
||||
url: The URL of the requests.
|
||||
*args: See `requests.option` parameters.
|
||||
**kwargs: See `requests.option` parameters.
|
||||
|
||||
Returns:
|
||||
The JSON-decoded result of the request.
|
||||
|
||||
Raises:
|
||||
requests.exceptions.RequestException: An error occured while
|
||||
performing the request.
|
||||
exceptions.PermissionDenied: The user does not have the right
|
||||
to perform this request.
|
||||
"""
|
||||
return self._request("get", *args, **kwargs)
|
||||
|
||||
def patch(self, *args, **kwargs):
|
||||
"""Performs a PATCH request.
|
||||
|
||||
PATCH request on a given URL that acts like `requests.patch` except
|
||||
that authentication to the API is automatically done and JSON response
|
||||
is decoded.
|
||||
|
||||
Args:
|
||||
url: The URL of the requests.
|
||||
*args: See `requests.patch` parameters.
|
||||
**kwargs: See `requests.patch` parameters.
|
||||
|
||||
Returns:
|
||||
The JSON-decoded result of the request.
|
||||
|
||||
Raises:
|
||||
requests.exceptions.RequestException: An error occured while
|
||||
performing the request.
|
||||
exceptions.PermissionDenied: The user does not have the right
|
||||
to perform this request.
|
||||
"""
|
||||
return self._request("patch", *args, **kwargs)
|
||||
|
||||
def post(self, *args, **kwargs):
|
||||
"""Performs a POST request.
|
||||
|
||||
POST request on a given URL that acts like `requests.post` except that
|
||||
authentication to the API is automatically done and JSON response is
|
||||
decoded.
|
||||
|
||||
Args:
|
||||
url: The URL of the requests.
|
||||
*args: See `requests.post` parameters.
|
||||
**kwargs: See `requests.post` parameters.
|
||||
|
||||
Returns:
|
||||
The JSON-decoded result of the request.
|
||||
|
||||
Raises:
|
||||
requests.exceptions.RequestException: An error occured while
|
||||
performing the request.
|
||||
exceptions.PermissionDenied: The user does not have the right
|
||||
to perform this request.
|
||||
"""
|
||||
return self._request("post", *args, **kwargs)
|
||||
|
||||
def put(self, *args, **kwargs):
|
||||
"""Performs a PUT request.
|
||||
|
||||
PUT request on a given URL that acts like `requests.put` except that
|
||||
authentication to the API is automatically done and JSON response is
|
||||
decoded.
|
||||
|
||||
Args:
|
||||
url: The URL of the requests.
|
||||
*args: See `requests.put` parameters.
|
||||
**kwargs: See `requests.put` parameters.
|
||||
|
||||
Returns:
|
||||
The JSON-decoded result of the request.
|
||||
|
||||
Raises:
|
||||
requests.exceptions.RequestException: An error occured while
|
||||
performing the request.
|
||||
exceptions.PermissionDenied: The user does not have the right
|
||||
to perform this request.
|
||||
"""
|
||||
return self._request("put", *args, **kwargs)
|
||||
|
||||
|
||||
class APIElement:
|
||||
def __init__(self, client, api_endpoint):
|
||||
self.client = client
|
||||
self.api_endpoint = api_endpoint
|
||||
|
||||
def save(self):
|
||||
data = self.as_api()
|
||||
if self.exists:
|
||||
self.client.put(API_URL + self.api_endpoint, json=data)
|
||||
else:
|
||||
self.client.post(API_URL + self.api_endpoint, json=data)
|
||||
|
||||
@property
|
||||
def exists(self):
|
||||
try:
|
||||
r = self.client.get(API_URL + self.api_endpoint)
|
||||
return True
|
||||
except requests.exceptions.HTTPError:
|
||||
return False
|
||||
|
||||
def fetch(self):
|
||||
r = self.client.get(API_URL + self.api_endpoint)
|
||||
self.from_dict(r)
|
||||
|
||||
def delete(self):
|
||||
self.client.delete(API_URL + self.api_endpoint)
|
||||
|
||||
|
||||
class Record(APIElement):
|
||||
class Types:
|
||||
A = "A"
|
||||
AAAA = "AAAA"
|
||||
ALIAS = "ALIAS"
|
||||
CAA = "CAA"
|
||||
CDS = "CDS"
|
||||
CNAME = "CNAME"
|
||||
DNAME = "DNAME"
|
||||
DS = "DS"
|
||||
KEY = "KEY"
|
||||
LOC = "LOC"
|
||||
MX = "MX"
|
||||
NS = "NS"
|
||||
OPENPGPKEY = "OPENPGPKEY"
|
||||
PTR = "PTR"
|
||||
SPF = "SPF"
|
||||
SRV = "SRV"
|
||||
SSHFP = "SSHFP"
|
||||
TLSA = "TLSA"
|
||||
TXT = "TXT"
|
||||
WKS = "WKS"
|
||||
|
||||
ENDPOINT = "livedns/domains/{fqdn}/records/{rrset_name}/{rrset_type}"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
client,
|
||||
domain,
|
||||
rrset_name="",
|
||||
rrset_type=None,
|
||||
rrset_values=None,
|
||||
**kwargs
|
||||
):
|
||||
endpoint = self.ENDPOINT.format(
|
||||
fqdn=domain, rrset_name=rrset_name, rrset_type=rrset_type
|
||||
)
|
||||
super().__init__(client, endpoint)
|
||||
self.rrset_name = rrset_name
|
||||
self.rrset_type = rrset_type
|
||||
self.rrset_values = rrset_values
|
||||
if self.rrset_values is None or self.rrset_type is None:
|
||||
self.fetch()
|
||||
|
||||
@classmethod
|
||||
def from_name(cls, client, domain, name):
|
||||
cls(client, domain, rrset_name=name)
|
||||
|
||||
def as_dict(self):
|
||||
return {
|
||||
"rrset_name": self.rrset_name,
|
||||
"rrset_type": self.rrset_type,
|
||||
"rrset_values": self.rrset_values,
|
||||
}
|
||||
|
||||
def as_api(self):
|
||||
return {
|
||||
"rrset_values": self.rrset_values
|
||||
}
|
||||
|
||||
def from_dict(self, d):
|
||||
self.rrset_name = d["rrset_name"]
|
||||
self.rrset_type = d["rrset_type"]
|
||||
self.rrset_values = d["rrset_values"]
|
||||
|
||||
def __repr__(self):
|
||||
return "<Record name=%s, type=%s, values=%r>" % (
|
||||
self.rrset_name,
|
||||
self.rrset_type,
|
||||
self.rrset_values,
|
||||
)
|
||||
|
||||
def __eq__(self, other):
|
||||
return (
|
||||
self.rrset_name == other.rrset_name
|
||||
and self.rrset_type == other.rrset_type
|
||||
and self.rrset_values == other.rrset_values
|
||||
)
|
||||
|
||||
def __hash__(self):
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
class DomainsRecords(APIElement):
|
||||
ENDPOINT = "livedns/domains/{fqdn}/records"
|
||||
|
||||
def __init__(self, client, fqdn, records=None, fetch=True):
|
||||
endpoint = self.ENDPOINT.format(fqdn=fqdn)
|
||||
super().__init__(client, endpoint)
|
||||
self.fqdn = fqdn
|
||||
self.records = records
|
||||
if self.records is None and fetch:
|
||||
self.fetch()
|
||||
else:
|
||||
self.records = []
|
||||
|
||||
def save(self):
|
||||
for r in self.records:
|
||||
r.save()
|
||||
|
||||
def from_dict(self, d):
|
||||
# it's actually an array
|
||||
if isinstance(d, dict):
|
||||
l = d["records"]
|
||||
else:
|
||||
l = d
|
||||
logging.getLogger("gandiAPI").debug(l)
|
||||
self.records = [Record(self.client, self.fqdn, **r) for r in l]
|
||||
|
||||
def as_dict(self):
|
||||
return [r.as_dict() for r in self.records]
|
||||
|
||||
def __repr__(self):
|
||||
return "<DomainsRecords domain=%s, records=%r>" % (self.fqdn, self.records)
|
||||
Loading…
Add table
Add a link
Reference in a new issue