Create documents from meta classes

Documents.py

class Field:

    def __init__(self, name=None, db_name=None, default=None, required=None, unique=None):
        self.name = name
        self.db_name = db_name
        self.default = default
        self.required = required
        self.unique = unique

    def to_mongo(self, value):
        return value

    def from_mongo(self, value):
        return value

    def is_empty(self, value):
        return value is None

    def validate(self, value):
        return not (self.required and self.is_empty(value))

    def __get__(self, instance, owner):
        if instance is None:
            return self

        if self.name not in instance._values:
            if callable(self.default):
                instance._values[self.name] = self.default()
            else:
                instance._values[self.name] = self.default

        return instance._values[self.name]

    def __set__(self, instance, value):
        instance._values[self.name] = value


class StringField(Field):

    def __init__(self, max_length=None, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_length = max_length

    def validate(self, value):
        if value is not None:
            if not isinstance(value, str):
                return False

            if self.max_length is not None and len(value) > self.max_length:
                return False

        return super().validate(value)

    def is_empty(self, value):
        return value is None or value == ""


class ListField(Field):

    def __init__(self, item_field=None, *args, **kwargs):

        if not isinstance(item_field, Field):
            raise ValueError("The item_field for must be a Field")

        super().__init__(*args, **kwargs)

        self.item_field = item_field

    def validate(self, value):
        if value is not None:
            if not isinstance(value, list):
                return False

            for item in value:
                if not self.item_field.validate(item):
                    return False

        return super().validate(value)

    def is_empty(self, value):
        return value is None or len(value) == 0

    def to_mongo(self, value):
        return list(map(self.item_field.to_mongo, value))

    def from_mongo(self, value):
        if value is None:
            return []
        return list(map(self.item_field.from_mongo, value))


class MetaDocument(type):
    def __new__(cls, name, bases, dct):

        if '_root' in dct and dct['_root']:
            return super().__new__(cls, name, bases, dct)

        dct['_fields'] = {}
        dct['_db_name_map'] = {}

        for base in bases:
            for field_name, field in filter(lambda x: isinstance(x[1], Field), base.__dict__.items()):
                MetaDocument.add_field(dct, field_name, field)

        for field_name, field in filter(lambda x: isinstance(x[1], Field), dct.items()):
            field.name = field_name
            MetaDocument.add_field(dct, field_name, field)

        dct['_values'] = {}

        if '__collection__' not in dct:
            dct['__collection__'] = name

        klass = super().__new__(cls, name, bases, dct)

        return klass

    @classmethod
    def add_field(cls, dct, field_name, field):
        if field_name in dct['_fields']:
            raise KeyError(f"Field '{field_name}' already exists")
        if not field.db_name:
            field.db_name = field_name
        if field.db_name in dct['_db_name_map']:
            raise KeyError(f"Field '{field_name}' already exists")
        field.name = field_name
        dct['_fields'][field_name] = field
        dct['_db_name_map'][field.db_name] = field_name


class Document(metaclass=MetaDocument):
    _root = True

    def __init__(self, **kwargs):
        for name, value in kwargs.items():
            if name in self._fields:
                setattr(self, name, value)

    def to_mongo(self):
        data = {}
        for field in self._fields.values():
            data[field.db_name] = field.to_mongo(
                getattr(self, field.name, None))
        return data

    @classmethod
    def from_mongo(cls, dct):

        kwargs = {}
        for db_name, value in dct.items():
            field_name = cls._db_name_map[db_name]
            field = cls._fields[field_name]
            kwargs[field_name] = field.from_mongo(value)

        return cls(**kwargs)

    @property
    def is_valid(self):
        for field in self._fields.values():
            if not field.validate(getattr(self, field.name, None)):
                return False
        return True

    def __eq__(self, other):
        for name in self._fields.keys():
            if getattr(self, name, None) != getattr(other, name, None):
                return False
        return True


class User(Document):
    first_name = StringField(db_name='firstName')
    last_name = StringField(db_name='lastName')


class PermissionedUser(User):
    roles = ListField(item_field=StringField())


assert PermissionedUser.first_name.name == "first_name"
assert PermissionedUser.first_name.db_name == "firstName"
assert PermissionedUser.last_name.name == "last_name"
assert PermissionedUser.last_name.db_name == "lastName"
assert PermissionedUser.roles.name == "roles"
assert PermissionedUser.roles.db_name == "roles"

user = PermissionedUser(
    first_name='Rob',
    last_name='Blackbourn',
    roles=['a', 'b', 'c'])

assert user.first_name == 'Rob'
assert user.last_name == 'Blackbourn'
assert user.roles == ['a', 'b', 'c']
assert user.is_valid

user.first_name = 'Thomas'
assert user.first_name == 'Thomas'

user2 = PermissionedUser.from_mongo(user.to_mongo())
assert user2 == user
assert user2.is_valid

print('Done')