Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
A simple wrapper around the Digital Ocean API (https://api.digitalocean.com). Initialize it with your credentials, then perform Digital Ocean API requests and it returns the parsed JSON API response as a dict. It throws an exception on API errors and maintains the same HTTPS connection for the lifetime of the object. This minimizing the bookkeep…
"""A Python3 wrapper for the Digital Ocean API.
Brad Conte (brad@bradconte.com)
"""
from http.client import HTTPSConnection
from http.client import HTTPConnection
import ssl
import json
class APIException(Exception):
"""Thrown when an Digital Ocean API request returns an error.
Attributes:
api_msg - the error message returned by the API
api_call - the API call that generated the error
"""
def __init__(self, api_msg, api_call):
self.api_msg = api_msg
self.api_call = api_call
def __str__ (self):
return 'API failed. Error msg = "{}", attempted API = "{}".'.format(
self.api_msg, self.api_call)
class DigitalOceanAPI(object):
"""A general Digital Ocean API wrapper. A single SSL connection is made
when the object is instantiated and kept until it's deleted. API requests
are returned in parsed JSON. API failures throw an exception. SSL
parameters are pass-through to the SSL module. Requires the D.O. Client ID
and API Key for the account.
Keyword Arguments:
check_cert - whether the HTTPS connection's cert should be verified
pemfile - path to PEM file, such as "/etc/ssl/certs/ca-certificates.pem"
capath - path to list of CA certs, such as "/etc/ssl/certs/"
Requires from the STL:
ssl
json
http.client
"""
api_host = "api.digitalocean.com"
# "check_cert" is True by default because we have API keys we're sending.
# By default, this means at least one of the keyword arguments must always
# be supplied since checking the cert requires a cert path, which has no
# working default.
def __init__(self, client_id, api_key, check_cert = True, pemfile = None,
capath = None):
self.connection = None
self.auth_data = ""
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
if check_cert:
ssl_ctx.verify_mode = ssl.CERT_REQUIRED
ssl_ctx.load_verify_locations(pemfile, capath)
else:
ssl_cts.verify_mode = ssl.CERT_NONE
self.connection = HTTPSConnection(
DigitalOceanAPI.api_host, context=ssl_ctx)
self.auth_data = "&client_id={}&api_key={}".format(client_id, api_key)
def __enter__(self):
return self
def request(self, url):
"""Perform an API request and returns the parsed JSON data. Throws
an exception with API description if the response is an error. See
api.digitalocean.com for API usage.
Arguments:
url - API call, minus auth, ex: "/domains/new?data=127.0.0.1&name=home"
"""
# If the URL doesn't contain arguments, add the "?" so we an append
# the self-auth arguments.
final_url = url
if not "?" in final_url:
final_url += "?"
final_url += self.auth_data
self.connection.request("GET", final_url)
response = self.connection.getresponse()
response_data = json.loads(response.read().decode("utf-8"))
response.close()
if response_data["status"] != "OK":
raise APIException(response_data["message"], url)
return response_data
def close(self):
if not self.connection is None:
self.connection.close()
self.connection = None
self.auth_data = ""
def __exit__(self, exc_type, exc_value, traceback):
self.close()
def __del__(self):
self.close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment