"""
SoftLayer.API
~~~~~~~~~~~~~
SoftLayer API bindings
:license: MIT, see LICENSE for more details.
"""
# pylint: disable=invalid-name
import warnings
from SoftLayer import auth as slauth
from SoftLayer import config
from SoftLayer import consts
from SoftLayer import transports
API_PUBLIC_ENDPOINT = consts.API_PUBLIC_ENDPOINT
API_PRIVATE_ENDPOINT = consts.API_PRIVATE_ENDPOINT
__all__ = [
'create_client_from_env',
'Client',
'BaseClient',
'API_PUBLIC_ENDPOINT',
'API_PRIVATE_ENDPOINT',
]
VALID_CALL_ARGS = set((
'id',
'mask',
'filter',
'headers',
'compress',
'raw_headers',
'limit',
'offset',
'verify',
))
[docs]def create_client_from_env(username=None,
api_key=None,
endpoint_url=None,
timeout=None,
auth=None,
config_file=None,
proxy=None,
user_agent=None,
transport=None,
verify=True):
"""Creates a SoftLayer API client using your environment.
Settings are loaded via keyword arguments, environemtal variables and
config file.
:param username: an optional API username if you wish to bypass the
package's built-in username
:param api_key: an optional API key if you wish to bypass the package's
built in API key
:param endpoint_url: the API endpoint base URL you wish to connect to.
Set this to API_PRIVATE_ENDPOINT to connect via SoftLayer's private
network.
:param proxy: proxy to be used to make API calls
:param integer timeout: timeout for API requests
:param auth: an object which responds to get_headers() to be inserted into
the xml-rpc headers. Example: `BasicAuthentication`
:param config_file: A path to a configuration file used to load settings
:param user_agent: an optional User Agent to report when making API
calls if you wish to bypass the packages built in User Agent string
:param transport: An object that's callable with this signature:
transport(SoftLayer.transports.Request)
:param bool verify: decide to verify the server's SSL/TLS cert. DO NOT SET
TO FALSE WITHOUT UNDERSTANDING THE IMPLICATIONS.
Usage:
>>> import SoftLayer
>>> client = SoftLayer.create_client_from_env()
>>> resp = client.call('Account', 'getObject')
>>> resp['companyName']
'Your Company'
"""
settings = config.get_client_settings(username=username,
api_key=api_key,
endpoint_url=endpoint_url,
timeout=timeout,
proxy=proxy,
verify=verify,
config_file=config_file)
if transport is None:
url = settings.get('endpoint_url')
if url is not None and '/rest' in url:
# If this looks like a rest endpoint, use the rest transport
transport = transports.RestTransport(
endpoint_url=settings.get('endpoint_url'),
proxy=settings.get('proxy'),
timeout=settings.get('timeout'),
user_agent=user_agent,
verify=verify,
)
else:
# Default the transport to use XMLRPC
transport = transports.XmlRpcTransport(
endpoint_url=settings.get('endpoint_url'),
proxy=settings.get('proxy'),
timeout=settings.get('timeout'),
user_agent=user_agent,
verify=verify,
)
# If we have enough information to make an auth driver, let's do it
if auth is None and settings.get('username') and settings.get('api_key'):
# NOTE(kmcdonald): some transports mask other transports, so this is
# a way to find the 'real' one
real_transport = getattr(transport, 'transport', transport)
if isinstance(real_transport, transports.XmlRpcTransport):
auth = slauth.BasicAuthentication(
settings.get('username'),
settings.get('api_key'),
)
elif isinstance(real_transport, transports.RestTransport):
auth = slauth.BasicHTTPAuthentication(
settings.get('username'),
settings.get('api_key'),
)
return BaseClient(auth=auth, transport=transport)
[docs]def Client(**kwargs):
"""Get a SoftLayer API Client using environmental settings.
Deprecated in favor of create_client_from_env()
"""
warnings.warn("use SoftLayer.create_client_from_env() instead",
DeprecationWarning)
return create_client_from_env(**kwargs)
[docs]class BaseClient(object):
"""Base SoftLayer API client.
:param auth: auth driver that looks like SoftLayer.auth.AuthenticationBase
:param transport: An object that's callable with this signature:
transport(SoftLayer.transports.Request)
"""
_prefix = "SoftLayer_"
def __init__(self, auth=None, transport=None):
self.auth = auth
self.transport = transport
[docs] def authenticate_with_password(self, username, password,
security_question_id=None,
security_question_answer=None):
"""Performs Username/Password Authentication
:param string username: your SoftLayer username
:param string password: your SoftLayer password
:param int security_question_id: The security question id to answer
:param string security_question_answer: The answer to the security question
"""
self.auth = None
res = self.call('User_Customer', 'getPortalLoginToken',
username,
password,
security_question_id,
security_question_answer)
self.auth = slauth.TokenAuthentication(res['userId'], res['hash'])
return res['userId'], res['hash']
def __getitem__(self, name):
"""Get a SoftLayer Service.
:param name: The name of the service. E.G. Account
Usage:
>>> import SoftLayer
>>> client = SoftLayer.create_client_from_env()
>>> client['Account']
<Service: Account>
"""
return Service(self, name)
[docs] def call(self, service, method, *args, **kwargs):
"""Make a SoftLayer API call.
:param method: the method to call on the service
:param \\*args: (optional) arguments for the remote call
:param id: (optional) id for the resource
:param mask: (optional) object mask
:param dict filter: (optional) filter dict
:param dict headers: (optional) optional XML-RPC headers
:param boolean compress: (optional) Enable/Disable HTTP compression
:param dict raw_headers: (optional) HTTP transport headers
:param int limit: (optional) return at most this many results
:param int offset: (optional) offset results by this many
:param boolean iter: (optional) if True, returns a generator with the results
:param bool verify: verify SSL cert
:param cert: client certificate path
Usage:
>>> import SoftLayer
>>> client = SoftLayer.create_client_from_env()
>>> client.call('Account', 'getVirtualGuests', mask="id", limit=10)
[...]
"""
if kwargs.pop('iter', False):
# Most of the codebase assumes a non-generator will be returned, so casting to list
# keeps those sections working
return list(self.iter_call(service, method, *args, **kwargs))
invalid_kwargs = set(kwargs.keys()) - VALID_CALL_ARGS
if invalid_kwargs:
raise TypeError(
'Invalid keyword arguments: %s' % ','.join(invalid_kwargs))
prefixes = (self._prefix, 'BluePages_Search', 'IntegratedOfferingTeam_Region')
if self._prefix and not service.startswith(prefixes):
service = self._prefix + service
http_headers = {'Accept': '*/*'}
if kwargs.get('compress', True):
http_headers['Accept-Encoding'] = 'gzip, deflate, compress'
else:
http_headers['Accept-Encoding'] = None
if kwargs.get('raw_headers'):
http_headers.update(kwargs.get('raw_headers'))
request = transports.Request()
request.service = service
request.method = method
request.args = args
request.transport_headers = http_headers
request.identifier = kwargs.get('id')
request.mask = kwargs.get('mask')
request.filter = kwargs.get('filter')
request.limit = kwargs.get('limit')
request.offset = kwargs.get('offset')
if kwargs.get('verify') is not None:
request.verify = kwargs.get('verify')
if self.auth:
extra_headers = self.auth.get_headers()
if extra_headers:
warnings.warn("auth.get_headers() is deprecated and will be "
"removed in the next major version",
DeprecationWarning)
request.headers.update(extra_headers)
request = self.auth.get_request(request)
request.headers.update(kwargs.get('headers', {}))
return self.transport(request)
__call__ = call
[docs] def iter_call(self, service, method, *args, **kwargs):
"""A generator that deals with paginating through results.
:param service: the name of the SoftLayer API service
:param method: the method to call on the service
:param integer limit: result size for each API call (defaults to 100)
:param \\*args: same optional arguments that ``Service.call`` takes
:param \\*\\*kwargs: same optional keyword arguments that ``Service.call`` takes
"""
limit = kwargs.pop('limit', 100)
offset = kwargs.pop('offset', 0)
if limit <= 0:
raise AttributeError("Limit size should be greater than zero.")
# Set to make unit tests, which call this function directly, play nice.
kwargs['iter'] = False
result_count = 0
keep_looping = True
while keep_looping:
# Get the next results
results = self.call(service, method, offset=offset, limit=limit, *args, **kwargs)
# Apparently this method doesn't return a list.
# Why are you even iterating over this?
if not isinstance(results, transports.SoftLayerListResult):
if isinstance(results, list):
# Close enough, this makes testing a lot easier
results = transports.SoftLayerListResult(results, len(results))
else:
yield results
return
for item in results:
yield item
result_count += 1
# Got less results than requested, we are at the end
if len(results) < limit:
keep_looping = False
# Got all the needed items
if result_count >= results.total_count:
keep_looping = False
offset += limit
def __repr__(self):
return "Client(transport=%r, auth=%r)" % (self.transport, self.auth)
__str__ = __repr__
def __len__(self):
return 0
class Service(object):
"""A SoftLayer Service.
:param client: A SoftLayer.API.Client instance
:param name str: The service name
"""
def __init__(self, client, name):
self.client = client
self.name = name
def call(self, name, *args, **kwargs):
"""Make a SoftLayer API call
:param service: the name of the SoftLayer API service
:param method: the method to call on the service
:param \\*args: same optional arguments that ``BaseClient.call`` takes
:param \\*\\*kwargs: same optional keyword arguments that
``BaseClient.call`` takes
:param service: the name of the SoftLayer API service
Usage:
>>> import SoftLayer
>>> client = SoftLayer.create_client_from_env()
>>> client['Account'].getVirtualGuests(mask="id", limit=10)
[...]
"""
return self.client.call(self.name, name, *args, **kwargs)
__call__ = call
def iter_call(self, name, *args, **kwargs):
"""A generator that deals with paginating through results.
:param method: the method to call on the service
:param integer chunk: result size for each API call
:param \\*args: same optional arguments that ``Service.call`` takes
:param \\*\\*kwargs: same optional keyword arguments that
``Service.call`` takes
Usage:
>>> import SoftLayer
>>> client = SoftLayer.create_client_from_env()
>>> gen = client.call('Account', 'getVirtualGuests', iter=True)
>>> for virtual_guest in gen:
... virtual_guest['id']
...
1234
4321
"""
return self.client.iter_call(self.name, name, *args, **kwargs)
def __getattr__(self, name):
if name in ["__name__", "__bases__"]:
raise AttributeError("'Obj' object has no attribute '%s'" % name)
def call_handler(*args, **kwargs):
" Handler that actually makes the API call "
return self(name, *args, **kwargs)
return call_handler
def __repr__(self):
return "<Service: %s>" % (self.name,)
__str__ = __repr__