Source code for wolframclient.evaluation.result

# -*- coding: utf-8 -*-

from __future__ import absolute_import, print_function, unicode_literals

import logging

from wolframclient.deserializers import binary_deserialize
from wolframclient.evaluation.cloud.request_adapter import wrap_response
from wolframclient.exception import (
    RequestException,
    WolframEvaluationException,
    WolframLanguageException,
    WolframParserException,
)
from wolframclient.utils import six
from wolframclient.utils.api import json
from wolframclient.utils.decorators import cached_property
from wolframclient.utils.logger import str_trim

logger = logging.getLogger(__name__)

__all__ = [
    "WolframResult",
    "WolframAPIResponseBuilder",
    "WolframAPIResponse",
    "WolframCloudEvaluationResponse",
    "WolframCloudEvaluationWXFResponse",
    "WolframCloudEvaluationJSONResponse",
    "WolframKernelEvaluationResult",
    "WolframAPIResponseAsync",
    "WolframEvaluationJSONResponseAsync",
    "WolframEvaluationWXFResponseAsync",
]


class WolframResultBase(object):
    pass


[docs]class WolframResult(WolframResultBase): """ The most generic result object. The actual result is returned via the method :func:`~wolframclient.evaluation.result.WolframResult.get`. If the result is a `success`, the field `result` is returned; otherwise, `failure` is returned and most likely contains an error message. """ def __init__(self, result=None, failure=None): self.success = failure is None self.failure = failure self.result = result
[docs] def get(self): """Return the result or raise an exception based on the success status.""" if self.success: return self.result else: raise WolframLanguageException(self.failure)
def __repr__(self): if self.success: return "{}<success={}, result={}>".format( self.__class__.__name__, self.success, self.result ) else: return "{}<success={}, failure={}>".format( self.__class__.__name__, self.success, self.failure )
class WolframAsyncResult(WolframResultBase): async def get(self): raise NotImplementedError class WolframEvaluationResultBase(WolframResultBase): def __init__(self): self._built = False self._success = False self._failure = None self._messages = None self._messages_name = None self._output = None self._result = None self._is_message_failure = False self.parsed_response = None @property def success(self): """ Evaluations succeed when it returns a result and no message is issued. """ if not self._built: self.build() return self._success @property def failure(self): if not self._built: self.build() return self._failure @property def messages(self): """ A list of the messages issued during the evaluation. """ if not self._built: self.build() return self._messages @property def messages_name(self): """ A list of the name of all the messages issued during the evaluation. """ if not self._built: self.build() return self._messages_name def iter_messages(self): """ Iterator over all text messages issued during the evaluation. :return: message text as a string. """ if self.messages: yield from self.messages def iter_messages_name(self): if self.messages_name: yield from self.messages_name def iter_messages_tuple(self): """ Iterator over all messages returned as a tuple: (message name, message text)""" if self.messages and self.messages_name: yield from zip(self.iter_messages_name(), self.iter_messages()) @property def output(self): """ A list of all content that got printed during evaluation e.g. using :wl:`Print`. """ if not self._built: self.build() return self._output def iter_output(self): """ Iterator over all printed output.""" if self.output: yield from self.output @property def result(self): if not self._built: self.build() return self._result @property def is_message_failure(self): if not self._built: self.build() return self._is_message_failure def build(self): self.parse_response() if self.parsed_response: self.build_from_parsed_response() # make sure to always build a response, here a generic one. elif not self._built: self.build_invalid_format() def get(self, silent=True): """Return the result or raise an exception. `silent` can be set to False to log all messages with warning severity. """ if self.success: return self.result elif self.is_message_failure: if not silent: for msg in self.iter_messages(): logger.warning(msg) return self.result else: raise WolframEvaluationException("Evaluation failed.", messages=self.failure) def parse_response(self): """ Parse the result input and set the attribute `parsed_response`. The result input can be encoded in various formats such as WXF, JSON, etc. The `parsed_response` dict is expected to have keys corresponding to those of the association returned by :wl:`EvaluationData`. """ raise NotImplementedError( "%s does not implement parse_response method." % (self.__class__.__name__,) ) def build_invalid_format(self, response_format_name=None): """ Build a result object for invalid format cases. """ logger.fatal("Invalid format. Failed to parse result.") self._success = False if response_format_name: self._failure = "Failed to decode response encoded with %s" % response_format_name else: self._failure = "Failed to decode response." self._built = True def build_from_parsed_response(self): self._success = self.parsed_response["Success"] self._result = self.parsed_response["Result"] self._output = self.parsed_response.get("Output", []) if not self._success: self._failure = self.parsed_response["FailureType"] if self._failure == "MessageFailure": self._is_message_failure = True self._messages_name = self.parsed_response["Messages"] self._messages = self.parsed_response["MessagesText"] else: logger.warning("Evaluation failed.") self._is_message_failure = False self._built = True def __repr__(self): if self.success: return "{}<expression={}>".format(self.__class__.__name__, self.result) elif self.is_message_failure: return "{}<success={}, result={}, messages={}>".format( self.__class__.__name__, self.success, self.result, self.messages ) else: return "{}<failure={}>".format(self.__class__.__name__, self.failure)
[docs]class WolframKernelEvaluationResult(WolframEvaluationResultBase): """A Wolfram result with WXF encoded data. Messages can be issued during a kernel evaluation. Those are stored as `messages`. If any message was returned by the kernel then the success status is `False`. The evaluation result is lazily computed when accessing the field `result`. The WXF bytes holding the evaluation result are stored in `wxf` and thus can be later parsed with a customized parser if necessary. All strings printed during the evaluation (e.g. Print["something"]) are stored in property `output` as a list. The dict holding evaluation data is available in `evaluation_data`. """ def __init__(self, wxf_eval_data, consumer=None): super().__init__() self.wxf_evaluation_data = wxf_eval_data # store the expression result serialized. self.wxf = None self.consumer = consumer
[docs] def parse_response(self): self.parsed_response = binary_deserialize(self.wxf_evaluation_data) self.wxf = self.parsed_response["Result"]
@cached_property def result(self): # Kernel evaluation encode the result as WXF. Lazily decoding it using the user consumer. return binary_deserialize(super().result, consumer=self.consumer)
[docs]class WolframCloudEvaluationResponse(WolframEvaluationResultBase): """Result object associated with cloud kernel evaluation. The response body associated to this type of result is encoded. Other fields provide additional information. The HTTP response object is stored as `http_response` and when HTTP error occurred it is stored in `request_error`. """ def __init__(self, response): super().__init__() self.http_response = wrap_response(response) self.request_error = self.http_response.status() != 200 def __repr__(self): if self._built and not self.request_error: return super().__repr__() elif self.request_error: return "{}<request error {}>".format( self.__class__.__name__, self.http_response.status() ) else: return "{}<successful request, request body not yet parsed>".format( self.__class__.__name__ )
[docs] def get(self, silent=False): return super().get(silent)
[docs] def build(self): if not self.request_error: super().build() else: logger.fatal( "Server invalid response %i: %s", self.http_response.status(), self.http_response.text(), ) raise RequestException(self.http_response)
[docs]class WolframCloudEvaluationWXFResponse(WolframCloudEvaluationResponse): """ Result object associated with cloud evaluation request WXF encoded. """
[docs] def parse_response(self): wxf = self.http_response.content() try: self.parsed_response = binary_deserialize(wxf) except WolframLanguageException as e: self.build_invalid_format(response_format_name="WXF")
[docs]class WolframCloudEvaluationJSONResponse(WolframCloudEvaluationResponse): """ Result object associated with cloud evaluation request JSON encoded. """
[docs] def parse_response(self): try: self.parsed_response = self.http_response.json() except json.JSONDecodeError as e: self.build_invalid_format(response_format_name="JSON")
class WolframCloudEvaluationResponseAsync(WolframCloudEvaluationResponse): """Asynchronous result object associated with cloud evaluation request. """ async def build(self): if not self.request_error: await self.parse_response() if self.parsed_response: self.build_from_parsed_response() elif not self._built: self.build_invalid_format() else: msg = await self.http_response.text() logger.fatal("Server invalid response %i: %s", self.http_response.status(), msg) self._built = True raise RequestException(self.http_response, msg=msg) async def parse_response(self): raise NotImplementedError( "%s does not implement parse_response asynchronous method." % (self.__class__.__name__,) ) @property async def success(self): if not self._built: await self.build() return self._success @property async def failure(self): if not self._built: await self.build() return self._failure @property async def result(self): if not self._built: await self.build() return self._result @property async def is_message_failure(self): if not self._built: await self.build() return self.is_message_failure @property async def messages(self): """ A list of the messages issued during the evaluation. """ if not self._built: await self.build() return self._messages @property async def messages_name(self): """ A list of the name of all the messages issued during the evaluation. """ if not self._built: await self.build() return self._messages_name async def iter_messages(self): """ Iterator over all text messages issued during the evaluation. :return: message text as a string. """ msgs = await self.messages if msgs: for msg in msgs: yield msg async def iter_messages_name(self): names = await self.messages_name if names: for name in names: yield name async def iter_messages_tuple(self): """ Iterator over all messages returned as a tuple: (message name, message text)""" msg = await self.messages names = await self.messages_name if msg and names: for tuple_msg in zip(names, msg): yield tuple_msg @property async def output(self): """ A list of all content that got printed during evaluation e.g. using :wl:`Print`. """ if not self._built: await self.build() return self._output async def iter_output(self): """ Iterator over all printed output.""" output = await self.output if output: for line in self.output: yield line @property async def is_message_failure(self): if not self._built: await self.build() return self._is_message_failure async def get(self, silent=False): """Return the result or raise an exception based on the success status.""" if await self.success: return await self.result elif await self.is_message_failure: if not silent: for msg in self.iter_messages(): logger.warning(msg) return self.result else: raise WolframEvaluationException("Evaluation failed.", messages=await self.failure)
[docs]class WolframEvaluationJSONResponseAsync(WolframCloudEvaluationResponseAsync): """Asynchronous result object associated with cloud evaluation request encoded with JSON. """
[docs] async def parse_response(self): try: self.parsed_response = await self.http_response.json() except json.JSONDecodeError as e: self.build_invalid_format(response_format_name="JSON")
[docs]class WolframEvaluationWXFResponseAsync(WolframCloudEvaluationResponseAsync): """Asynchronous result object associated with cloud evaluation request encoded with WXF. """
[docs] async def parse_response(self): wxf = await self.http_response.content() try: self.parsed_response = binary_deserialize(wxf) except WolframLanguageException as e: self.build_invalid_format(response_format_name="WXF")
_DEFAULT_DECODERS = { "application/vnd.wolfram.wxf": binary_deserialize, "application/json": json.loads, }
[docs]class WolframAPIResponse(WolframResult): """ A generic API response. This class is lazily constructed when the response body becomes available. A decoder is inferred from the content type. Currently JSON and WXF formats are supported. """ def __init__(self, response, decoder=None): self.response = response self.content_type = response.headers().get("Content-Type", None) if decoder: self.decoder = decoder else: self.decoder = _DEFAULT_DECODERS.get(self.content_type) self.status = response.status() self.parsed_response = None self.success = None self._failure = None self._built = False
[docs] def build(self): raise NotImplementedError
[docs] def get(self): if not self._built: self.build() return self._get()
def _get(self): if self.success: return self.result else: raise WolframLanguageException(self._failure)
[docs] def failure(self): if not self._built: self.build() return self._failure
def __repr__(self): return "<%s:success=%s>" % (self.__class__.__name__, self.success)
[docs]class WolframAPIResponseAsync(WolframAPIResponse): """ Asynchronous counterpart of :class:`~wolframclient.evaluation.result.WolframAPIResponse`, awaiting for the response body. Most of the class logic is implemented in :data:`WolframAPIResponse`, except the build method which has to be a coroutine. """
[docs] async def get(self): """ Return the result or raise an exception based on the success status. This is a coroutine.""" if not self._built: await self.build() return self._get()
[docs] async def build(self): raise NotImplementedError
class WolframAPIFailureResponse(WolframAPIResponse): def __init__(self, response, decoder=None): super().__init__(response, decoder=decoder) self.success = False class WolframAPIFailureResponseAsync(WolframAPIResponseAsync, WolframAPIFailureResponse): pass class WolframAPIResponse200(WolframAPIResponse): def __init__(self, response, decoder=None): super().__init__(response, decoder=decoder) self.success = True def build(self): self._built = True if self.decoder is not None: try: self.result = self.decoder(self.response.content()) except Exception as e: self.success = False self._failure = "Decoder error: {}".format(e) self.exception = e else: self.result = self.response.content() class WolframAPIResponse200Async(WolframAPIResponseAsync, WolframAPIResponse200): async def build(self): if self.decoder is not None: try: self.result = self.decoder(await self.response.content()) except Exception as e: self.success = False self._failure = "Decoder error: {}".format(e) self.exception = e else: self.result = await self.response.content() self._built = True class WolframAPIResponseRedirect(WolframAPIFailureResponse): def __init__(self, response, decoder=None): super().__init__(response, decoder) self.location = None def build(self): self.location = self.response.headers().get("location", None) logger.warning("Redirected to %s.", self.location) self._specific_failure() self._built = True def _specific_failure(self): raise NotImplementedError class WolframAPIResponse301(WolframAPIResponseRedirect): def _specific_failure(self): """ should not happen since we follow redirection """ self._failure = "Resource permanently moved to new location {}".format(self.location) class WolframAPIResponse301Async(WolframAPIResponse301, WolframAPIResponseAsync): pass class WolframAPIResponse302(WolframAPIResponseRedirect): def _specific_failure(self): # hack because the server is not returning 403. cf. CLOUD-12946 if self.location is not None and "j_spring_oauth_security_check" in self.location: self._failure = "Not allowed to access requested resource." else: self._failure = "Resource moved to new location {}".format(self.location) class WolframAPIResponse302Async(WolframAPIResponse302, WolframAPIResponseAsync): pass class WolframAPIResponse400(WolframAPIFailureResponse): def __init__(self, response, decoder=None): super().__init__(response, decoder=decoder) self._fields_in_error = None def build(self): try: if self.decoder: self.parsed_response = self.decoder(self.response.content()) else: self.parsed_response = self._unexpected_content_type() except (json.JSONDecodeError, WolframParserException): logger.fatal( "Failed to parse server response as %s:\n%s", self.content_type, str_trim(self.response.content(), max_char=200), ) raise self._failed_to_parse() self._update_from_response() self._built = True def fields_in_error(self): """Return all the fields in error with their message as a tuple of tuples""" if not self._built: self.build() return self._fields_in_error def _failed_to_parse(self): return RequestException(self.response, msg="Failed to parse server response.") def _unexpected_content_type(self): logger.warning( "Response content-type: %s is not supported. Cannot decode content: %s", self.content_type, str_trim(self.response.content()), ) return { "Failure": "Cannot decode server response. No decoder found for content-type: %s." % self.content_type } def _update_from_response(self): self._failure = self.parsed_response.get("Failure", None) fields = self.parsed_response.get("Fields", None) logger.warning("Wolfram API error response: %s", self._failure) if fields is not None: self._fields = set(fields.keys()) logger.warning("Fields in error: %s", self._fields) self._fields_in_error = [] for field, err in fields.items(): failure = err.get("Failure", None) if failure is not None: self._fields_in_error.append((field, failure)) class WolframAPIResponse400Async(WolframAPIResponse400, WolframAPIResponseAsync): async def build(self): # ignoring content-type. Must be JSON. Make sure it's robust enough. try: if self.decoder: self.parsed_response = self.decoder(await self.response.content()) else: self.parsed_response = await self._unexpected_content_type() except json.JSONDecodeError as e: logger.fatal( "Failed to parse server response as %s:\n%s", self.content_type, await self.response.content(), ) raise self._failed_to_parse() self._update_from_response() self._built = True async def _unexpected_content_type(self): logger.warning( "Response content-type: %s is not supported. Cannot decode content: %s", self.content_type, str_trim(await self.response.content()), ) return { "Failure": "Cannot decode server response. No decoder found for content-type: %s." % self.content_type } async def fields_in_error(self): """Return all the fields in error with their message as a list of tuples""" if not self._built: await self.build() return self._fields_in_error class WolframAPIResponse401(WolframAPIFailureResponse): def build(self): self._failure = self.response.text() logger.warning("Authentication missing or failed. Server response: %s", self._failure) self._built = True class WolframAPIResponse401Async(WolframAPIResponse401, WolframAPIResponseAsync): async def build(self): # ignoring content-type. Must be JSON. Make sure it's robust enough. self._failure = await self.response.text() self._built = True logger.warning("Authentication missing or failed. Server response: %s", self._failure) class WolframAPIResponse404(WolframAPIFailureResponse): def build(self): self._failure = "The resource %s can't not be found." % self.response.url() logger.warning("Wolfram API error response: %s", self._failure) self._built = True class WolframAPIResponse404Async(WolframAPIResponse404, WolframAPIResponseAsync): async def build(self): WolframAPIResponse404.build(self) class WolframAPIResponseGeneric(WolframAPIFailureResponse): def build(self): self._failure = self.response.text() self._built = True class WolframAPIResponseGenericAsync(WolframAPIResponseAsync): async def build(self): self._failure = await self.response.text() self._built = True class WolframAPIResponse500(WolframAPIResponseGeneric): def __init__(self, response, decoder=None): super().__init__(response, decoder) logger.fatal("Internal server error occurred.") class WolframAPIResponse500Async(WolframAPIResponseGenericAsync): def __init__(self, response, decoder=None): super().__init__(response, decoder) logger.fatal("Internal server error occurred.")
[docs]class WolframAPIResponseBuilder(object): """Map error code to handler building the appropriate :class:`~wolframclient.evaluation.result.WolframAPIResponse` """ response_mapper = { 200: WolframAPIResponse200, 301: WolframAPIResponse301, 302: WolframAPIResponse302, 400: WolframAPIResponse400, 401: WolframAPIResponse401, 404: WolframAPIResponse404, 500: WolframAPIResponse500, } async_response_mapper = { 200: WolframAPIResponse200Async, 301: WolframAPIResponse301Async, 302: WolframAPIResponse302Async, 400: WolframAPIResponse400Async, 401: WolframAPIResponse401Async, 404: WolframAPIResponse404Async, 500: WolframAPIResponse500Async, }
[docs] @staticmethod def build(response, decoder=None): adapter = wrap_response(response) if adapter.asynchronous: return WolframAPIResponseBuilder.async_response_mapper.get( adapter.status(), WolframAPIResponseGenericAsync )(adapter, decoder=decoder) else: return WolframAPIResponseBuilder.response_mapper.get( adapter.status(), WolframAPIResponseGeneric )(adapter, decoder=decoder)
[docs] @staticmethod def map(status_code, response_class): if not isinstance(response_class, WolframAPIResponse): raise ValueError( "Response class %s is not a subclass of %s" % (response_class.__class__.__name__, WolframAPIResponse.__class__.__name__) ) if not isinstance(status_code, six.integer_types): logger.warning("Invalid status code: %s", status_code) raise ValueError("HTTP status code must be string.") logger.debug( "Mapping http response status %i to function %s", status_code, response_class.__name__, ) WolframAPIResponseBuilder.response_mapper[status_code] = response_class
def __init__(self): raise NotImplementedError("Cannot initialize. Use static 'method' build.")