Source

tutagx / tutagx / meta / from_yaml.py

from collections import namedtuple
import yaml

from tutagx.meta import validate, decode
from tutagx.meta.decode import ObjectCreationStrategy

_ReaderNamespace = namedtuple('_ReaderNamespace', 'seen unfinished extra')


#TODO replace this API with something better
class YAMLReader:
    def __init__(self, model):
        self._mcls = type(model)
        self._cls = model
        self._decoder = decode.make_decoder(model)

    def decode(self, val, namespace=None, validator_ns=None, source=None):
        results, errors = self.decode_many(
            [val], namespace, validator_ns, source
        )
        if errors:
            return None, errors
        return results[0], errors

    def decode_many(self, raw_values,
                    namespace=None, validator_ns=None,
                    source=None):
        if namespace is None:
            namespace = type(self).new_namespace()
        if validator_ns is None:
            validator_ns = validate.Validator.new_namespace()
        args = (namespace.seen, namespace.unfinished) + namespace.extra
        validator = validate.Validator(self._cls)
        results = []
        for raw_value in raw_values:
            validator.validate(raw_value, validator_ns, source)
            # Note that we continue validating, but stop decoding,
            # as soon as ANY error is encountered.
            # This allows producing additional useful errors without
            # trying to decode possibly invalid data.
            if validator_ns.errors:
                continue
            # If decoding raises an exception, it's a bug in the
            # validation and SHOULD lead to a crash!
            results.append(self._decoder(raw_value, *args))
        # If any error occured, the whole data set is likely invalid.
        # Thus, we do not attempt to return partial data.
        if validator_ns.errors:
            return None, validator_ns.errors
        return results, validator_ns.errors

    def load(self, f, namespace=None, validator_ns=None):
        results, errors = self.load_many(f, namespace, validator_ns)
        if errors:
            return None, errors
        return results[0], errors

    def load_many(self, f, namespace=None, validator_ns=None):
        try:
            source = f.name
        except AttributeError:
            source = '<unknown file>'
        raw_data = yaml.safe_load_all(f)
        return self.decode_many(raw_data, namespace, validator_ns, source)

    def loads(self, s, namespace=None):
        return self.decode(yaml.safe_load(s), namespace)

    @classmethod
    def new_namespace(cls):
        """
        Return a namespace object, which controls the scope of object IDs
        encountered in YAML documents.
        Reader namespaces have two members:

        * ``seen``, a dictionary with object IDs encountered while decoding as
          keys and the objects instanciated for these IDs as values.
        * ``unfinished``, a set of object IDs that some loaded objects referred
          to for which no object has been loaded (yet).
          If this set is non-empty after all data was loaded, there were
          bogus IDs in the input.

        Namespaces are only relevant if you plan on using one namespace across
        multiple documents, or need access to the information exposed.
        """
        return _ReaderNamespace({}, set(), cls._extra_decoder_args())

    @classmethod
    def _extra_decoder_args(cls):
        # Dummy implementation, this is only useful for subclasses
        return ()