Python mongoengine 模块，Document() 实例源码

def test_add_operators_to_field_reference_field():
    from mongoengine import ReferenceField, Document, StringField
    from graphene_mongo.operators import gen_operators_of_field, allowed_operators
    from graphene_mongo.fields import respective_special_fields

    class Other(Document):
        name = StringField()

    class Test(Document):
        test = ReferenceField(Other)

    field = Test.test
    r_graphene = respective_special_fields[type(field)]
    applied_operators = gen_operators_of_field('test', field, r_graphene('test', field), allowed_operators(field))

    assert sorted(list(applied_operators.keys())) == format_fields(['in', 'nin', 'ne'])

    assert isinstance(applied_operators['test__in'], graphene.List)
    assert isinstance(applied_operators['test__nin'], graphene.List)
    assert isinstance(applied_operators['test__ne'], graphene.ID)

def test_mutate_wrong_number_arguments():
    from mongoengine import Document

    class Test(Document):
        pass

    def mutate():
        pass

    with pytest.raises(Exception) as e_info:
        Options('TestSchema', {'model': Test, 'mutate': staticmethod(mutate)})

    assert str(e_info.value) == 'Failed to generate schema {}, mutate method must accept two params. ' \
                                'The first is the arguments passed to mutate in query, for instance: ' \
                                'username:"NewObjName". Second is the context of the application, if it is flask, ' \
                                'will be flask global request.'.format('TestSchema')

def test_validator_wrong():
    from mongoengine import Document, StringField

    class Test(Document):
        parent = StringField()

    with pytest.raises(Exception) as e_info:
        Options('TestSchema', {'model': Test, 'validator': True})

    assert str(e_info.value) == "'validator' attribute must be callable."

    with pytest.raises(Exception) as e_info:
        Options('TestSchema', {'model': Test, 'validator': lambda x: x})

    assert str(e_info.value) == ("The 'validator' attribute must be a callable that accepts four arguments: "
                                 "model, fields, query, special_params")

def setUp(self):
        """Setup class."""
        from bson import DBRef, ObjectId

        class Source(db.Document):
            pass

        class Model(db.Document):
            src = db.ReferenceField(Source, dbref=True)

        self.src_cls = Source
        self.model_cls = Model
        self.src_id = ObjectId()
        self.data = json.dumps({
            "src": {"collection": "source", "id": str(self.src_id)}
        })
        self.expected_data = {"src": DBRef("source", self.src_id)}
        self.hook = generate_object_hook(self.model_cls)

def setUp(self):
        """Setup class."""
        from bson import ObjectId, DBRef

        class Source(db.Document):
            pass

        class Model(db.Document):
            src = db.ReferenceField(Source)

        self.src_cls = Source
        self.model_cls = Model
        self.src_id = ObjectId()
        self.data = json.dumps({
            "src": str(self.src_id)
        })
        self.expected_data = {
            "src": DBRef("source", self.src_id)
        }
        self.hook = generate_object_hook(self.model_cls)

def setUp(self):
        """Setup test."""
        from datetime import datetime, timedelta
        from calendar import timegm

        class DateTime(db.Document):
            date = db.DateTimeField()

        self.model_cls = DateTime
        now = datetime.utcnow()
        epoch_mil = int(timegm(now.timetuple())*1000 + now.microsecond / 1000)
        self.data = json.dumps({"date": epoch_mil})
        self.expected_data = {
            "date": datetime.utcfromtimestamp(
                        int(epoch_mil / 1000)
                    ) + timedelta(milliseconds=int(epoch_mil % 1000))
        }
        self.hook = generate_object_hook(self.model_cls)

def setUp(self):
        """Setup test."""
        from datetime import datetime

        class DateTime(db.Document):
            date = db.DateTimeField()

        self.model_cls = DateTime
        now = datetime.utcnow()

        # This format shouldn't be supported.
        self.data = json.dumps({"date": {
            "year": now.year,
            "month": now.month,
            "date": now.day
        }})
        self.hook = generate_object_hook(self.model_cls)

def init_mappings(self):
        """Initialize RBAC Mappings.

        RBAC Mappings always refer to a (Organization, Team) combination.

        In order to reference the Organization instance of this Team, we
        are using `self._instance`, which is a proxy object to the upper
        level Document.

        """
        if not RBACMapping:
            return
        if self.name == 'Owners':
            return
        if RBACMapping.objects(org=self._instance.id, team=self.id).only('id'):
            raise me.ValidationError(
                'RBAC Mappings already initialized for Team %s' % self
            )
        for perm in ('read', 'read_logs'):
            RBACMapping(
                org=self._instance.id, team=self.id, permission=perm
            ).save()

def test_not_mongoengine_document():
    with pytest.raises(Exception) as e_info:
        Options('TestSchema', {'model': False})

    assert str(e_info.value) == 'Failed to generate schema {}, model must be ' \
                                'a subclass of mongoengine.Document.'.format('TestSchema')

def test_mutate_not_static():
    from mongoengine import Document

    class Test(Document):
        pass

    with pytest.raises(Exception) as e_info:
        Options('TestSchema', {'model': Test, 'mutate': 'To fails'})

    assert str(e_info.value) == 'Failed to generate schema {}, mutate method must ' \
                                'be a method with the decorator staticmethod.'.format('TestSchema')

def test_mongoengine_field_not_implemented():
    from mongoengine import FileField, Document

    class Test(Document):
        field_error = FileField()

    with pytest.raises(Exception) as e_info:
        Options('TestSchema', {'model': Test})

    assert str(e_info.value) == "It was not possible to generate schema for {} because the " \
                                "field {} is of the type {}, and that field is not supported yet." \
                                .format("Test", 'field_error', FileField)

def test_mongoengine_field_references_self():
    from mongoengine import Document, ReferenceField

    class Test(Document):
        parent = ReferenceField('self')

    with pytest.raises(Exception) as e_info:
        Options('TestSchema', {'model': Test})

    assert str(e_info.value) == "It was not possible to generate schema for {} because the field {} is a " \
                                "ReferenceField to self and this is not supported yet."\
                                .format("TestSchema", 'parent')

def to_mongo(self, document, **kwargs):
        """
        Convert to python-typed dict.

        Parameters:
            document: The document.
        """
        cur_depth = getattr(self, "$$cur_depth$$", self.max_depth)
        if not getattr(self, "$$good_json$$", None) or \
                cur_depth >= self.max_depth:
            return super(FollowReferenceField, self).to_mongo(
                document, **kwargs
            )
        doc = document
        if isinstance(document, db.Document):
            if document.pk is None and self.id_check:
                self.error("The referenced document needs ID.")
        else:
            doc = self.document_type.objects(
                pk=super(
                    FollowReferenceField, self
                ).to_mongo(document, **kwargs)
            ).get()
        if isinstance(doc, Document):
            doc.begin_goodjson(cur_depth + 1)
        ret = doc.to_mongo(**kwargs)
        if isinstance(doc, Document):
            doc.end_goodjson(cur_depth + 1)
        if "_id" in ret and issubclass(self.document_type, Document):
            ret["id"] = ret.pop("_id", None)
        return ret

def setUp(self):
        """Setup class."""
        from bson import ObjectId

        class TestModel(db.Document):
            user = db.ObjectIdField()

        self.model_cls = TestModel
        self.hook = generate_object_hook(self.model_cls)
        self.data = json.dumps({"user": "56f63a716a8dec7705f36409"})
        self.expected_data = {"user": ObjectId("56f63a716a8dec7705f36409")}

def setUp(self):
        """Setup test."""
        from base64 import b64encode, b64decode
        from bson.binary import Binary, BINARY_SUBTYPE

        class BinaryTest(db.Document):
            text = db.BinaryField()

        self.model_cls = BinaryTest
        self.data = {
            "text": {
                "data": b64encode(
                    ("This is a test").encode("utf-8")
                ).decode("utf-8"),
                "type": BINARY_SUBTYPE
            }
        }
        self.expected_data = {
            "text": Binary(
                b64decode(self.data["text"]["data"]),
                self.data["text"]["type"]
            )
        }
        self.data = json.dumps(self.data)

        self.hook = generate_object_hook(BinaryTest)

def setUp(self):
        """Setup test."""
        from uuid import uuid5, NAMESPACE_DNS

        class UUIDModel(db.Document):
            uuid = db.UUIDField()

        self.model_cls = UUIDModel
        uuid = uuid5(NAMESPACE_DNS, "This is a test")
        self.expected_data = {
            "uuid": uuid
        }
        self.data = json.dumps({"uuid": str(uuid)})
        self.hook = generate_object_hook(UUIDModel)

def __init__(self, app=None, config=None):
        _include_mongoengine(self)

        self.app = None
        self.Document = Document
        self.DynamicDocument = DynamicDocument

        if app is not None:
            self.init_app(app, config)

def _is_foreign_key(field_cls, field_name):
    return field_cls and issubclass(field_cls, Document) and not _is_dereference_disabled(field_name)

def verified_attrs(self, attrs):
        """ Function to verify if the attributes is of the right type """
        model = attrs.get('model')
        mutate = attrs.get('mutate')
        validator = attrs.get('validator')
        validator = validator.__func__ if isinstance(validator, staticmethod) else validator

        if model is None:
            raise AttributeError('Failed to generate schema {},'
                                 ' model attribute was not given.'.format(self.class_name))

        if not model or not issubclass(model, Document):
            raise TypeError('Failed to generate schema {}, model must be '
                            'a subclass of mongoengine.Document.'.format(self.class_name))

        if mutate and not isinstance(mutate, staticmethod):
            raise TypeError('Failed to generate schema {}, mutate method must '
                            'be a method with the decorator staticmethod.'.format(self.class_name))

        if mutate and len(inspect.signature(mutate.__func__).parameters) != 2:
            raise TypeError('Failed to generate schema {}, mutate method must accept two params. '
                            'The first is the arguments passed to mutate in query, for instance: username:"NewObjName".'
                            ' Second is the context of the application, if it is flask, will be flask global request.'
                            .format(self.class_name))

        # verify if all fields of document is supported
        for f_name, m_field in model._fields.items():
            if type(m_field) not in respective_fields and type(m_field) not in respective_special_fields:
                raise NotImplementedError("It was not possible to generate schema for {} because the "
                                          "field {} is of the type {}, and that field is not supported yet."
                                          .format(model.__name__, f_name, type(m_field)))

            if isinstance(m_field, list_fields):
                if type(m_field.field) not in respective_fields and type(m_field.field) not in respective_special_fields:
                    raise NotImplementedError("It was not possible to generate schema for {} because the "
                                              "field {} is a List of the type {}, and that field is not supported yet."
                                              .format(model.__name__, f_name, type(m_field.field)))

            if isinstance(m_field, reference_fields):
                if isinstance(m_field.document_type_obj, str) and m_field.document_type_obj == 'self':
                    raise NotImplementedError("It was not possible to generate schema for {} because the "
                                              "field {} is a ReferenceField to self and this is not supported yet."
                                              .format(self.class_name, f_name))

        if validator and not callable(validator):
            raise AttributeError("'validator' attribute must be callable.")
        elif validator and len(inspect.signature(validator).parameters) != 4:
            raise AttributeError("The 'validator' attribute must be a callable that accepts four arguments: "
                                 "model, fields, query, special_params")

        return model, mutate.__func__ if mutate else None, model._fields, validator