Source code for wolframclient.serializers.wxfencoder.wxfexprprovider

from __future__ import absolute_import, print_function, unicode_literals

from wolframclient.serializers.wxfencoder.wxfencoder import (
    DefaultWXFEncoder,
    NotEncodedException,
    WXFEncoder,
)


[docs]class WXFExprProvider(object): """ Expression provider pull instances of WXFExpr from instances of `WXFEncoder`. `WXFExprProvider` can be initialized with an encoder. If none is provided the default class `DefaultWXFEncoder` is used to instantiate one. It is possible to add extra encoder using `add_encoder`. The order in which encoders are called is the one in which they were added. Note that for performance reasons, it is recommended to have the default encoder be first, as such one should not initialize a provider with an encoder except if the default one is not suited for ones needs. An optional `default` function can be passed to the provider at initialization. It is used in last resort if no encoder handled the object. It is applied to the object and is expected to transform it to something serializable e.g: a string using `default=repr`. One must carefully design the `default` function to avoid stack overflow. """ def __init__(self, encoder=None, default=None): self.encoders = [] if encoder is not None: self.add_encoder(encoder) else: self.add_encoder(DefaultWXFEncoder()) self.default = default
[docs] def add_encoder(self, *encoders): """ Add a new encoder to be called last if all others failed to encode an object. """ for encoder in encoders: if isinstance(encoder, WXFEncoder): self.encoders.append(encoder) encoder._provider = self else: raise TypeError("Invalid encoder.") return self
[docs] def provide_wxfexpr(self, o): """ Main function, a generator of wxf expr.""" for wxfexpr in self._iter(o): yield wxfexpr
def _iter(self, o, use_default=True): """ Try to encode a given expr using encoders, if none was able to process the expr and a `default` function was provided, applies it and try again. Otherwise fail. """ for encoder in self.encoders: try: for sub in encoder._encode(o): yield sub return # the expression was not encoded. Moving on to the next encoder. except NotEncodedException: pass if use_default and self.default is not None: for sub in self._iter(self.default(o), use_default=False): yield sub else: raise TypeError( "Object of type %s is not serializable to WXF." % o.__class__.__name__ )