<commit_before>from inspect import isclass

from typing import Any, Dict, List, Tuple


# TODO improve handling of non-string keys


def _assert_isinstance(value, _type):
    """Check if a value conforms to a type (recursively for collections and records)."""
    if not isinstance(type(_type), type):
        raise ValueError("_type must be a type, but got: {}".format(_type))
    if not isclass(_type):  # Any and Union are types but aren't classes
        if _type is Any:
            return
        try:
            for union_branch in _type.__args__:
                try:

                    _assert_isinstance(value, union_branch)
                    break
                except TypeError:
                    pass
            else:  # no break means no match
                raise TypeError("{!r} is not in {}".format(value, _type))
        except AttributeError:
            raise ValueError("Unrecognized non-class type {}".format(_type))
    else:
        if isinstance(_type, JsonRecord):
            if not isinstance(value, _type):
                raise TypeError("{!r} is not of type {}".format(value, _type))
            for f, f_type in _type.schema.items():
                _assert_isinstance(value.get(f), f_type)
        elif issubclass(_type, Dict):
            if not isinstance(value, Dict):
                raise TypeError("{!r} is not of type {}".format(value, _type))
            k_type, v_type = _type.__args__
            for k, v in value.items():
                _assert_isinstance(k, k_type)  # kind of meaningless since keys have to be str
                _assert_isinstance(v, v_type)
        elif issubclass(_type, List):
            if not isinstance(value, List):
                raise TypeError("{!r} is not of type {}".format(value, _type))
            e_type, = _type.__args__  # args is a 1-tuple for list, so the trailing comma is needed
            for e in value:
                _assert_isinstance(e, e_type)
        elif issubclass(_type, Tuple):
            if not isinstance(value, (Tuple, List)):
                raise TypeError("{!r} is not of type {}".format(value, _type))
            for item, union_branch in zip(value, _type.__args__):
                _assert_isinstance(item, union_branch)
        else:
            if _type is int and type(value) is bool:  # bool passes isinstance(b, int)
                raise TypeError("{!r} is not of type {}".format(value, _type))
            if _type is float:  # JSON numbers may be parsed as ints when we expect floats
                _type = (int, float)
            if not isinstance(value, _type):
                raise TypeError("{!r} is not of type {}".format(value, _type))


class JsonRecord(type):
    """
    Metaclass for records which can be represented in JSON.

    Enforces types according to the schema provided. Schema is defined as a dictionary of field
    names mapped to types (primitive types, List, Dict, Tuple, Optional, other record types).

    class ExampleRecord(metaclass=JsonRecord):
        schema = {
            "field": str,
            "optional_field": Optional[int],
            "complex_field": Dict[str, List[Union[int, bool, str]]],
        }
    """
    def __new__(meta, name, bases, dct):
        class _JsonRecordSuper(dict):
            def __init__(self, *a, **kw):
                for a_dict in a + (kw,):
                    for k, v in a_dict.items():
                        self[k] = v
                for f, f_type in type(self).schema.items():
                    # TODO find a way to only check for presence of required fields
                    _assert_isinstance(self[f], f_type)

            def _validate_key(self, key):
                if key not in type(self).schema:
                    raise KeyError(key)

            def __getitem__(self, item):
                self._validate_key(item)
                return self.get(item)  # absent Optionals should return None

            def __getattr__(self, item):
                return self[item]

            def __setitem__(self, key, value):
                self._validate_key(key)
                _assert_isinstance(value, type(self).schema[key])
                super(_JsonRecordSuper, self).__setitem__(key, value)

            def __setattr__(self, key, value):
                self[key] = value

            def __repr__(self):
                return "{}({})".format(name, dict(self))<commit_msg>Adds coercion when constructing records to support constructoring nested records from raw json<commit_after>from inspect import isclass

from typing import Any, Dict, List, Tuple


# TODO improve handling of non-string keys


def _assert_isinstance(value, _type):
    """Check if a value conforms to a type (recursively for collections and records)."""
    if not isinstance(type(_type), type):
        raise ValueError("_type must be a type, but got: {}".format(_type))
    if not isclass(_type):  # Any and Union are types but aren't classes
        if _type is Any:
            return
        try:
            for union_branch in _type.__args__:
                try:
                    _assert_isinstance(value, union_branch)
                    break
                except TypeError:
                    pass
            else:  # no break means no match
                raise TypeError("{!r} is not in {}".format(value, _type))
        except AttributeError:
            raise ValueError("Unrecognized non-class type {}".format(_type))
    else:
        if isinstance(_type, JsonRecord):
            if not isinstance(value, _type):
                raise TypeError("{!r} is not of type {}".format(value, _type))
            for f, f_type in _type.schema.items():
                _assert_isinstance(value.get(f), f_type)
        elif issubclass(_type, Dict):
            if not isinstance(value, Dict):
                raise TypeError("{!r} is not of type {}".format(value, _type))
            k_type, v_type = _type.__args__
            for k, v in value.items():
                _assert_isinstance(k, k_type)  # kind of meaningless since keys have to be str
                _assert_isinstance(v, v_type)
        elif issubclass(_type, List):
            if not isinstance(value, List):
                raise TypeError("{!r} is not of type {}".format(value, _type))
            e_type, = _type.__args__  # args is a 1-tuple for list, so the trailing comma is needed
            for e in value:
                _assert_isinstance(e, e_type)
        elif issubclass(_type, Tuple):
            if not isinstance(value, (Tuple, List)):
                raise TypeError("{!r} is not of type {}".format(value, _type))
            for item, index_type in zip(value, _type.__args__):
                _assert_isinstance(item, index_type)
        else:
            if _type is int and type(value) is bool:  # bool passes isinstance(b, int)
                raise TypeError("{!r} is not of type {}".format(value, _type))
            if _type is float:  # JSON numbers may be parsed as ints when we expect floats
                _type = (int, float)
            if not isinstance(value, _type):
                raise TypeError("{!r} is not of type {}".format(value, _type))


def _coerce_records(value, _type):
    """Coerce dictionaries into record types where appropriate."""
    if not isinstance(type(_type), type):
        raise ValueError("_type must be a type, but got: {}".format(_type))
    if not isclass(_type):  # Any and Union are types but aren't classes
        if _type is Any:
            return value
        for branch in _type.__args__:  # return the first coercion that sticks
            if isinstance(branch, JsonRecord):
                try:
                    return branch(value)
                except TypeError:
                    pass
            elif issubclass(branch, Dict):
                try:
                    _assert_isinstance(value, branch)
                    return value
                except TypeError:
                    pass
        return value
    else:
        if isinstance(_type, JsonRecord):
            return _type(value)
        elif issubclass(_type, List):
            e_type, = _type.__args__
            return [_coerce_records(e, e_type) for e in value]
        elif issubclass(_type, Tuple):
            return [_coerce_records(e, index_type) for e, index_type in zip(value, _type.__args__)]
        elif issubclass(_type, Dict):
            k_type, v_type = _type.__args__
            return {k: _coerce_records(v, v_type) for k, v in value.items()}
        else:
            return value



class JsonRecord(type):
    """
    Metaclass for records which can be represented in JSON.

    Enforces types according to the schema provided. Schema is defined as a dictionary of field
    names mapped to types (primitive types, List, Dict, Tuple, Optional, other record types).

    class ExampleRecord(metaclass=JsonRecord):
        schema = {
            "field": str,
            "optional_field": Optional[int],
            "complex_field": Dict[str, List[Union[int, bool, str]]],
        }
    """
    def __new__(meta, name, bases, dct):
        class _JsonRecordSuper(dict):
            def __init__(self, *a, **kw):
                for a_dict in a + (kw,):
                    for k, v in a_dict.items():
                        self[k] = v
                for f, f_type in type(self).schema.items():
                    # TODO find a way to only check for presence of required fields
                    _assert_isinstance(self[f], f_type)

            def _validate_key(self, key):
                if key not in type(self).schema:
                    raise KeyError(key)

            def __getitem__(self, item):
                self._validate_key(item)
                return self.get(item)  # absent Optionals should return None

            def __getattr__(self, item):
                return self[item]

            def __setitem__(self, key, value):
                self._validate_key(key)
                _type = type(self).schema[key]
                value = _coerce_records(value, _type)
                _assert_isinstance(value, _type)
                super(_JsonRecordSuper, self).__setitem__(key, value)

            def __setattr__(self, key, value):
                self[key] = value

            def __repr__(self):
                return "{}({})".format(name, dict(self))
{'fraction_non_alphanumeric': 0.09246211755622044, 'fraction_numerical': 0.00019303156066016793, 'mean_word_length': 1.8743411927877947, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '853723', 'n_tokens_mistral': 2876, 'n_tokens_neox': 2746, 'n_words': 942}