# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, unicode_literals
import json
from io import IOBase
from wolframclient.utils import six
from wolframclient.utils.api import oauth, urllib
__all__ = [
"SecuredAuthenticationKey",
"UserIDPassword",
"OAuthSessionBase",
"OAuthAsyncSessionBase",
]
[docs]class SecuredAuthenticationKey(object):
""" Represents a Secured Authentication Key generated using the Wolfram Language
function `GenerateSecuredAuthenticationKey[]`
It is used as an input when authenticating a cloud session.
"""
is_xauth = False
def __init__(self, consumer_key, consumer_secret):
self.consumer_key = consumer_key
self.consumer_secret = consumer_secret
[docs]class UserIDPassword(object):
""" Represents user credentials used to login to a cloud.
It is used as an input when authenticating a cloud session.
"""
is_xauth = True
def __init__(self, user, password):
self.user = user
self.password = password
[docs]class OAuthSessionBase(object):
""" A family of classes dealing with authentication with OAuth method."""
DEFAULT_CONTENT_TYPE = {
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": "WolframClientForPython/1.0",
}
def __init__(
self,
server,
consumer_key,
consumer_secret,
signature_method=None,
client_class=oauth.Client,
):
self.consumer_key = consumer_key
self.consumer_secret = consumer_secret
self.signature_method = signature_method or oauth.SIGNATURE_HMAC
self.client_class = client_class
self._client = None
self._oauth_token = None
self._oauth_token_secret = None
self.server = server
[docs] def authenticate(self):
""" Authenticate with a given server using the user credentials"""
raise NotImplementedError
[docs] def signed_request(self, uri, headers={}, data=None, method="POST"):
""" Sign a given request and issue it."""
raise NotImplementedError
[docs] def authorized(self):
"""Return a reasonably accurate state of the authentication status."""
return (
self._client is not None
and bool(self._client.client_secret)
and bool(self._client.resource_owner_key)
and bool(self._client.resource_owner_secret)
)
def _update_client(self):
self._client = self.client_class(
self.consumer_key,
client_secret=self.consumer_secret,
resource_owner_key=self._oauth_token,
resource_owner_secret=self._oauth_token_secret,
signature_type=oauth.SIGNATURE_TYPE_AUTH_HEADER,
realm=self.server.cloudbase,
encoding="iso-8859-1",
)
def _update_token_from_request_body(self, body):
try:
token = json.loads(body)
self._oauth_token = token["oauth_token"]
self._oauth_token_secret = token["oauth_token_secret"]
except:
token = urllib.parse_qs(body)
self._oauth_token = token[b"oauth_token"][0]
self._oauth_token_secret = token[b"oauth_token_secret"][0]
[docs]class OAuthAsyncSessionBase(OAuthSessionBase):
[docs] async def authenticate(self):
""" Asynchronous OAuth authentication class dealing with various tokens and signing requests. """
raise NotImplementedError
[docs] async def signed_request(self, uri, headers={}, data=None, method="POST"):
""" Sign a given request and issue it asynchronously."""
raise NotImplementedError
class WolframAPICallBase(object):
"""Perform an API call to a given target.
The API call is actually performed when :func:`~wolframclient.evaluation.WolframAPICall.perform`
is called.
Parameters can be added using one of the various functions that this class exposes. They
can be of many types including: string, files, WL serializable python objects, binary data with arbitrary
content-type (e.g: *image/png*).
"""
def __init__(self, target, api, permission_key=None):
self.target = target
self.api = api
self.parameters = {}
self.files = {}
self.permission_key = permission_key
def set_parameter(self, name, value):
"""Add a new API input parameter from a serializable python object."""
self.parameters[name] = value
return self
def add_file_parameter(self, name, fp, filename=None, content_type=None):
"""Add a new API input parameter from a file pointer `fp`"""
if content_type is None:
self.files[name] = fp
else:
fname = filename or "tmp_%s" % name
self.files[name] = (fname, fp, content_type)
return self
def add_binary_parameter(
self, name, data, filename=None, content_type="application/octet-stream"
):
"""Add a new API input parameter as a blob of binary data."""
if isinstance(data, IOBase):
return self.add_file_parameter(
name, data, filename=filename, content_type=content_type
)
if isinstance(data, six.binary_type):
fname = filename or "tmp_%s" % name
self.files[name] = (fname, data, content_type)
return self
else:
raise TypeError("Input data must be bytes or IOBase.")
def add_image_data_parameter(
self, name, image_data, filename=None, content_type="image/png"
):
"""Add a new API image input parameter from binary data.
If the data in `image_data` does not represent an image in the `PNG` format, the optional parameter
`content_type` must be set accordingly to the appropriate content type, e.g. *image/jpeg*, *image/gif*, etc.
"""
return self.add_binary_parameter(name, image_data, filename, content_type)
def perform(self, **kwargs):
"""Make the API call and return the result."""
raise NotImplementedError
def perform_future(self, **kwargs):
"""Make the API call asynchronously and return a future object."""
raise NotImplementedError
def __repr__(self):
return "<%s api=%s, parameters=%s>" % (
self.__class__.__name__,
self.api,
set().union(self.parameters.keys(), self.files.keys()) or None,
)
def __str__(self):
return repr(self)
