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

def test_add_operators_to_field_list_field():
    from mongoengine import ListField, SortedListField
    from graphene_mongo.operators import gen_operators_of_field, allowed_operators
    from graphene_mongo.fields.respective import respective_special_fields, respective_fields

    for m_field in [ListField, SortedListField]:
        for f, r_graphene in respective_fields.items():
            field = m_field(f())

            applied_operators = gen_operators_of_field('test', field, respective_special_fields[m_field],
                                                       allowed_operators(field))

            expected = format_fields(['size'])

            assert len(applied_operators.keys()) == len(expected)
            assert sorted(list(applied_operators.keys())) == sorted(expected)

            obj_list_field = applied_operators['test']('listTest', field)
            assert isinstance(obj_list_field, graphene.List)

            # here we test to assert that the type of items of the list is what is suppose to be
            assert isinstance(obj_list_field.of_type, type(r_graphene))

def __get_doc(self, fld, item):
        """Get document as dict or a list of documents."""
        from .fields import FollowReferenceField

        @singledispatch
        def doc(fld, item):
            return item

        @doc.register(db.ListField)
        def doc_list(fld, item):
            return [self.__get_doc(fld.field, el) for el in item]

        @doc.register(FollowReferenceField)
        def doc_frl(fld, item):
            doc = fld.document_type.objects(id=item).get()
            doc.begin_goodjson()
            result = doc.to_mongo()
            doc.end_goodjson()
            return result

        result = doc(fld, item)

        if isinstance(result, dict) and "id" not in result and "_id" in result:
            result["id"] = result.pop("_id")
        return result

def is_to_many_relationship(field):
    """
    Returns True, if the *field* describes a *to many* relationship.

    The field types are:

    *   :class:`mongoengine.ListField`
    *   :class:`mongoengine.SortedListField`

    with a simple reference field as element.
    """
    if isinstance(field, mongoengine.ListField) \
        and is_to_one_relationship(field.field):
        return True

    if isinstance(field, mongoengine.SortedListField)\
        and is_to_one_relationship(field.field):
        return True
    return False

def test_list_field_no_field():
    """ Assert that raises error if a ListField is given without a type, for instance: ListField() """
    from graphene_mongo.fields.special_fields import list_field
    from mongoengine import ListField

    with pytest.raises(Exception) as e_info:
        list_field('test_field', ListField())

    assert str(e_info.value) == str(AttributeError('Error in {} field, have sure that this is defined with a '
                                                   'mongoengine field'.format('test_field')))

def __set_gj_flag_sub_field(self, instance, fld, cur_depth):
        """Set $$good_json$$ flag to subfield."""
        from .fields import FollowReferenceField

        def set_good_json(fld):
            setattr(fld, "$$good_json$$", True)
            setattr(fld, "$$cur_depth$$", cur_depth)

        @singledispatch
        def set_flag_recursive(fld, instance):
            set_good_json(fld)

        @set_flag_recursive.register(db.ListField)
        def set_flag_list(fld, instance):
            set_good_json(fld.field)

        @set_flag_recursive.register(db.EmbeddedDocumentField)
        def set_flag_emb(fld, instance):
            if isinstance(instance, Helper):
                instance.begin_goodjson(cur_depth)

        @set_flag_recursive.register(FollowReferenceField)
        def set_flag_self(fld, instance):
            set_good_json(fld)

        set_flag_recursive(fld, instance)

def __unset_gj_flag_sub_field(self, instance, fld, cur_depth):
        """Unset $$good_json$$ to subfield."""
        from .fields import FollowReferenceField

        def unset_flag(fld):
            setattr(fld, "$$good_json$$", None)
            setattr(fld, "$$cur_depth$$", None)
            delattr(fld, "$$good_json$$")
            delattr(fld, "$$cur_depth$$")

        @singledispatch
        def unset_flag_recursive(fld, instance):
            unset_flag(fld)

        @unset_flag_recursive.register(db.ListField)
        def unset_flag_list(fld, instance):
            unset_flag(fld.field)

        @unset_flag_recursive.register(db.EmbeddedDocumentField)
        def unset_flag_emb(fld, instance):
            if isinstance(instance, Helper):
                instance.end_goodjson(cur_depth)

        @unset_flag_recursive.register(FollowReferenceField)
        def unset_flag_self(fld, instance):
            unset_flag(fld)

        unset_flag_recursive(fld, instance)

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 _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 _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