Source

tutagx / tutagx / meta / from_yaml.py

import yaml

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


#TODO replace this API with something better
class YAMLReader:
    def __init__(self, context=None):
        if context is None:
            context = DecodingContext()
        self.context = context
        self.validator = validate.Validator()

    def decode(self, cls, val, source=None):
        results, errors = self.decode_many(cls, [val], source)
        if errors:
            return None, errors
        assert len(results) == 1
        return results[0], errors

    def decode_many(self, cls, raw_values, source=None):
        results = []
        valid = True
        try:
            add = cls.add_decoding_extra
        except AttributeError:
            pass
        else:
            add(self.context.extra)
        for raw_value in raw_values:
            valid = self.validator.validate(cls, raw_value)
            # 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 valid:
                # If decoding raises an exception, it's a bug in the
                # validation and SHOULD lead to a crash!
                results.append(decode.decode(cls, raw_value, self.context))
        # If any error occured, the whole data set is likely invalid.
        # Thus, we do not attempt to return partial data.
        if not valid:
            return None, self.validator.context.errors
        return results, self.validator.context.errors

    def load(self, cls, f):
        results, errors = self.load_many(cls, f)
        if errors:
            return None, errors
        assert len(results) == 1
        return results[0], errors

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

    def loads(self, cls, s):
        return self.decode(cls, yaml.safe_load(s))
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.