Skip to content

Instantly share code, notes, and snippets.

@cloudnull
Last active December 27, 2015 12:19
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save cloudnull/7325348 to your computer and use it in GitHub Desktop.
Save cloudnull/7325348 to your computer and use it in GitHub Desktop.
Using Openstack Clients as Libraries. This is a simple gist with the means to allow you to leverage the various openstack clients as libraries.
import logging
from novaclient.v1_1 import client as nova_client
from cinderclient.v1 import client as cinder_client
from swiftclient import client as swift_client
LOG = logging.getLogger(__name__)
class auth_plugin(object):
def __init__(self):
"""Craetes an authentication plugin for use with Rackspace."""
self.auth_url = self.global_auth()
def global_auth(self):
"""Return the Rackspace Cloud US Auth URL."""
return "https://identity.api.rackspacecloud.com/v2.0/"
def get_auth_url(self):
"""Return the Rackspace Cloud US Auth URL."""
return "https://identity.api.rackspacecloud.com/v2.0/"
def _authenticate(self, cls, auth_url):
"""Authenticate against the Rackspace auth service."""
body = {"auth": {
"RAX-KSKEY:apiKeyCredentials": {
"username": cls.user,
"apiKey": cls.password,
"tenantName": cls.projectid}}}
return cls._authenticate(auth_url, body)
def authenticate(self, cls, auth_url):
"""Authenticate against the Rackspace US auth service."""
return self._authenticate(cls, auth_url)
class rax_creds(object):
"""Load Rackspace Credentials."""
def __init__(self, user, region, key=None, tenant=None, password=None,
system='rackspace'):
self.user = user
self.tenant = tenant
self.apikey = key
self.password = password
self.region = region.upper()
self.system = system
self.plugin = auth_plugin()
class Clients(object):
"""Load in a Openstack client.
Usage:
>>> my_creds = rax_creds(
... user='USERNAME',
... region='RAX_REGION',
... key='RAX_API_KEY',
... password='RAX_PASSWROD' # Required for Swift Client
... )
>>> auth = Clients(creds=my_creds)
>>> novaclient = auth.get_client('novaclient')
>>> servers = novaclient.servers.list()
"""
def __init__(self, creds):
self.creds = creds
self.creds_dict = dict(
username=self.creds.user,
api_key=self.creds.apikey,
project_id=self.creds.user,
tenant_name=self.creds.tenant,
password=self.creds.password,
region_name=self.creds.region,
insecure=False,
cacert=None,
auth_url=self.creds.plugin.auth_url,
)
def novaclient(self):
"""Load the nova client."""
LOG.debug(
'novaclient connection created using token "%s" and url "%s"'
% (self.creds_dict['username'], self.creds_dict['auth_url'])
)
creds_dict = {
'username': self.creds_dict.get('username'),
'api_key': self.creds_dict.get('api_key'),
'project_id': self.creds_dict.get('project_id'),
'tenant_id': self.creds_dict.get('tenant_name'),
'region_name': self.creds_dict.get('region_name'),
'insecure': self.creds_dict.get('insecure', False),
'auth_system': self.creds.system,
'auth_plugin': self.creds.plugin
}
return nova_client.Client(**creds_dict)
def cinderclient(self):
"""Load the cinder client."""
LOG.debug(
'cinderclient connection created using token "%s" and url "%s"'
% (self.creds_dict['username'], self.creds_dict['auth_url'])
)
creds_dict = {
'username': self.creds_dict.get('username'),
'api_key': self.creds_dict.get('api_key'),
'project_id': self.creds_dict.get('project_id'),
'tenant_id': self.creds_dict.get('tenant_name'),
'region_name': self.creds_dict.get('region_name'),
'insecure': self.creds_dict.get('insecure', False),
}
return cinder_client.Client(**creds_dict)
def swiftclient(self):
"""Load the swift client."""
LOG.debug(
'swiftclient connection created using token "%s" and url "%s"'
% (self.creds_dict['username'], self.creds_dict['auth_url'])
)
# If the tenant is None make it an empty string
if self.creds_dict.get('tenant') is None:
self.creds_dict['tenant'] = ' '
creds_dict = {
'user': self.creds_dict.get('username'),
'key': self.creds_dict.get('password'),
'authurl': self.creds_dict.get('auth_url'),
'tenant_name': ' ',
'os_options': {
'region': self.creds_dict.get('region_name')
},
'auth_version': 2.0
}
return swift_client.Connection(**creds_dict)
def get_client(self, client):
"""Get the Client that we need.
:param client: str
:return client_type: function
"""
client_type = getattr(self, client)
if client_type is None:
raise SystemExit('No Client Type Found')
else:
return client_type()
@traverseda
Copy link

:/

Why are things terrible. Not your things, but the amount of monkey patching I need to do to get either django-swift-storage or django-cumulus working. Someone clearly isn't doing their job, and I bet I wouldn't be having this kind of problem if I was using AWS.

Thanks for the code.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment