Created
September 4, 2013 00:49
-
-
Save B-Con/6431500 to your computer and use it in GitHub Desktop.
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…
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""A Python3 wrapper for the Digital Ocean API. | |
Brad Conte ([email protected]) | |
""" | |
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