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

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 __init__(self, *args, **kwargs):
        """
        Initialize the class.

        Parameters:
            *args, **kwsrgs: Any arguments to be passed to ReferenceField.
        Keyword Arguments:
            id_check: Set False to disable id check. By default, this value is
                True
            autosave: Set True to save/update the referenced document when
                to_python is called.
            max_depth: Set natural value to set depath limit for
                loop-reference. By default this value is set to 3.
        """
        self.id_check = kwargs.pop("id_check", True)
        self.autosave = kwargs.pop("autosave", False)
        self.max_depth = kwargs.pop("max_depth", 3)
        super(FollowReferenceField, self).__init__(*args, **kwargs)

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 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 setUp(self):
        """Setup the class."""
        class SelfReferenceDocument(Document):
            name = db.StringField()
            reference = db.ReferenceField("self")

        class TestDocument(Document):
            title = db.StringField()
            references = db.ListField(
                db.ReferenceField(SelfReferenceDocument)
            )

        self.references = [
            SelfReferenceDocument(
                pk=ObjectId(), name=("test {}").format(counter)
            ) for counter in range(3)
        ]
        for (index, srd) in enumerate(self.references):
            srd.reference = self.references[
                (index + 1) % len(self.references)
            ]
            srd.to_json = MagicMock(side_effect=srd.to_json)
        self.model_cls = TestDocument
        self.model = TestDocument(
            pk=ObjectId(), title="Test", references=self.references
        )
        self.model.to_mongo = MagicMock(
            return_value={
                "id": self.model.id, "title": self.model.title,
                "references": self.references
            }
        )

        self.model.to_mongo = lambda x: {
            "id": self.model.pk,
            "title": "Test",
            "references": [str(srd.pk) for srd in self.references]
        }

def owner(self):
        """Return the Organization (instance) owning self.

        We refrain from storing the owner as a me.ReferenceField in order to
        avoid automatic/unwanted dereferencing.

        """
        return Organization.objects.get(id=self.owner_id)

def conv_List(self, model, field, kwargs):
        if isinstance(field.field, ReferenceField):
            return ModelSelectMultipleField(model=field.field.document_type, **kwargs)
        if field.field.choices:
            kwargs['multiple'] = True
            return self.convert(model, field.field, kwargs)
        field_args = kwargs.pop("field_args", {})
        unbound_field = self.convert(model, field.field, field_args)
        unacceptable = {
            'validators': [],
            'filters': [],
            'min_entries': kwargs.get('min_entries', 0)
        }
        kwargs.update(unacceptable)
        return f.FieldList(unbound_field, **kwargs)

def _document_typeof(doc_cls, field_name):
    try:
        orm_field = doc_cls._fields[field_name]
    except (KeyError, AttributeError):
        return None

    if isinstance(orm_field, ListField):
        orm_field = orm_field.field
    if isinstance(orm_field, (ReferenceField, EmbeddedDocumentField)):
        return orm_field.document_type

    return None

def is_to_one_relationship(field):
    """
    Returns True, if the *field* is a reference field:

    *   :class:`mongoengine.ReferenceField`
    *   :class:`mongoengine.CachedReferenceField`
    *   :class:`mongoengine.GenericReferenceField`

    All of these fields describe a *to-one* relationship.
    """
    return isinstance(field, mongoengine.ReferenceField)\
        or isinstance(field, mongoengine.CachedReferenceField)\
        or isinstance(field, mongoengine.GenericReferenceField)

def _follow_reference(self, max_depth, current_depth,
                          use_db_field, *args, **kwargs):
        from .fields import FollowReferenceField
        ret = {}
        for fldname in self:
            fld = self._fields.get(fldname)
            is_list = isinstance(fld, db.ListField)
            target = fld.field if is_list else fld

            if all([
                isinstance(
                    target, (db.ReferenceField, db.EmbeddedDocumentField)
                ), not isinstance(target, FollowReferenceField)
            ]):
                value = None
                if is_list:
                    value = []
                    for doc in getattr(self, fldname, []):
                        value.append(json.loads((
                            target.document_type.objects(
                                id=doc.id
                            ).get() if isinstance(doc, DBRef) else doc
                        ).to_json(
                            follow_reference=True,
                            max_depth=max_depth,
                            current_depth=current_depth + 1,
                            use_db_field=use_db_field,
                            *args, **kwargs
                        )))
                else:
                    doc = getattr(self, fldname, None)
                    value = json.loads(
                        (
                            target.document_type.objects(
                                id=doc.id
                            ).get() if isinstance(doc, DBRef) else doc
                        ).to_json(
                            follow_reference=True,
                            max_depth=max_depth,
                            current_depth=current_depth + 1,
                            use_db_field=use_db_field,
                            *args, **kwargs
                        )
                    ) if doc else doc
                if value is not None:
                    ret.update({fldname: value})
        return ret

def from_json(cls, json_str, created=False, *args, **kwargs):
        """
        Decode from human-readable json.

        Parameters:
            json_str: JSON string that should be passed to the serialized
            created: a parameter that is passed to cls._from_son.
            *args, **kwargs: Any additional arguments that is passed to
                json.loads.
        """
        from .fields import FollowReferenceField
        hook = generate_object_hook(cls)
        if "object_hook" not in kwargs:
            kwargs["object_hook"] = hook
        dct = json.loads(json_str, *args, **kwargs)
        for name, fld in cls._fields.items():
            if any([
                getattr(fld, "exclude_from_json", None),
                getattr(fld, "exclude_json", None)
            ]):
                dct.pop(name, None)
        from_son_result = cls._from_son(SON(dct), created=created)

        @singledispatch
        def normalize_reference(ref_id, fld):
            """Normalize Reference."""
            return ref_id and fld.to_python(ref_id) or None

        @normalize_reference.register(dict)
        def normalize_reference_dict(ref_id, fld):
            """Normalize Reference for dict."""
            return fld.to_python(ref_id["id"])

        @normalize_reference.register(list)
        def normalize_reference_list(ref_id, fld):
            """Normalize Reference for list."""
            return [
                normalize_reference(ref.id, fld) for ref in ref_id
            ]

        for fldname, fld in cls._fields.items():
            target = fld.field if isinstance(fld, db.ListField) else fld

            if not isinstance(target, db.ReferenceField) or \
                    isinstance(target, FollowReferenceField):
                continue

            value = dct.get(fldname)
            setattr(
                from_son_result, fldname,
                normalize_reference(getattr(value, "id", value), target)
            )
        return from_son_result