summaryrefslogtreecommitdiff
path: root/warlock/core.py
blob: eacd7ab934d6eea341cde4feb8202daaa865f1f3 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
"""Core Warlock functionality"""

import copy

import jsonschema


class InvalidOperation(RuntimeError):
    pass


class ValidationError(ValueError):
    pass


def model_factory(schema):
    """Generate a model class based on the provided JSON Schema

    :param schema: dict representing valid JSON schema
    """
    schema = copy.deepcopy(schema)

    def validator(obj):
        """Apply a JSON schema to an object"""
        try:
            jsonschema.validate(obj, schema)
        except jsonschema.ValidationError:
            raise ValidationError()

    class Model(object):
        """Self-validating model for arbitrary objects"""
        def __init__(self, **kwargs):
            self.__dict__['validator'] = validator
            try:
                self.__dict__['validator'](kwargs)
            except ValidationError:
                raise ValueError()
            else:
                self.__dict__['raw'] = kwargs

            self.__dict__['changes'] = {}

        def __getattr__(self, key):
            try:
                return self.__getitem__(key)
            except KeyError:
                raise AttributeError(key)

        def __getitem__(self, key):
            return self.__dict__['raw'][key]

        def __setitem__(self, key, value):
            mutation = copy.deepcopy(self.__dict__['raw'])
            mutation[key] = value
            try:
                self.__dict__['validator'](mutation)
            except ValidationError:
                raise InvalidOperation()
            self.__dict__['raw'] = mutation
            self.__dict__['changes'][key] = value

        def __setattr__(self, key, value):
            self.__setitem__(key, value)

        def iteritems(self):
            return copy.deepcopy(self.__dict__['raw']).iteritems()

        def items(self):
            return copy.deepcopy(self.__dict__['raw']).items()

        @property
        def changes(self):
            return copy.deepcopy(self.__dict__['changes'])

    Model.__name__ = str(schema['name'])
    return Model