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

def convert_fields(attrs_mongo_doc):
    """ Return a tuple od dicts,
    @param attrs_mongo_doc: dict of attributes of the mongoengine.Document, that is accessible by Document._fields     
    fields: is the pure fields, key is the name of field, and values is the repective graphene field
    fields_mutation: is the fields that is used in mutation, that will go direct in args of mutation
    operators_mutation: is the fields that user set values to be saved on document
    operators_single: is the fields with operators that will be in the query for a single result
    operators_list: same of operators single, but these are avaliable only for query of lists
    """

    fields, fields_mutation, operators_mutation, operators_single = {}, {}, {}, {}

    for f_name, mongo_field in attrs_mongo_doc.items():
        field = RelationMongoGraphene(name=f_name, mongo_field=mongo_field)

        fields[f_name] = field.graphene
        operators_single.update(field.operators)
        fields_mutation[f_name] = field.mutation
        if not isinstance(mongo_field, ObjectIdField):
            operators_mutation[f_name] = field.mutation

    operators_list = operators_single.copy()
    operators_single['skip'] = graphene.Int()

    operators_list.update({p: graphene.Int() for p in special_query_parameters})

    return fields, fields_mutation, operators_mutation, operators_single, operators_list

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 find_by_ids(cls, ids, params=FindParams()):
        if isinstance(cls.id, ObjectIdField):
            ids = [ObjectId(i) for i in ids]
        return cls.find(_id={'$in': ids}, params=params)

def find_by_ids_ordered(cls, ids, params=FindParams(), strict=True):
        # pylint: disable=no-member
        if isinstance(cls.id, ObjectIdField):
            ids = [ObjectId(i) for i in ids]
        docs = {doc['_id']: doc for doc in cls.find(_id={'$in': ids}, params=params)}
        docs = [docs[i] for i in ids if i in docs]
        if strict and len(docs) != len(ids):
            raise Exception('Some docs not found')
        return docs