aiomongodel 0.2.0

建模

要创建模型，只需创建一个新的模型类，从 aiomongodel.Document类，列出所有模型字段并放置 带有模型元选项的Meta类。要创建子文档，请创建 一个带有字段的类，并从aiomongodel.EmbeddedDocument继承它。

# models.pyfromdatetimeimportdatetimefrompymongoimportIndexModel,DESCENDINGfromaiomongodelimportDocument,EmbeddedDocumentfromaiomongodel.fieldsimport(StrField,BoolField,ListField,EmbDocField,RefField,SynonymField,IntField,FloatField,DateTimeField,ObjectIdField)classUser(Document):_id=StrField(regex=r'[a-zA-Z0-9_]{3, 20}')is_active=BoolField(default=True)posts=ListField(RefField('models.Post'),default=lambda:list())quote=StrField(required=False)# create a synonym fieldname=SynonymField(_id)classMeta:collection='users'classPost(Document):# _id field will be added automatically as# _id = ObjectIdField(defalut=lambda: ObjectId())title=StrField(allow_blank=False,max_length=50)body=StrField()created=DateTimeField(default=lambda:datetime.utcnow())views=IntField(default=0)rate=FloatField(default=0.0)author=RefField(User,mongo_name='user')comments=ListField(EmbDocField('models.Comment'),default=lambda:list())classMeta:collection='posts'indexes=[IndexModel([('created',DESCENDING)])]default_sort=[('created',DESCENDING)]classComment(EmbeddedDocument):_id=ObjectIdField(default=lambda:ObjectId())author=RefField(User)body=StrField()# `s` property of the fields can be used to get a mongodb string name# to use in queriesassertUser._id.s=='_id'assertUser.name.s=='_id'# name is synonymassertPost.title.s=='title'assertPost.author.s=='user'# field has mongo_nameassertPost.comments.body.s=='comments.body'# compound name

积垢

frommotor.motor_asyncioimportAsyncIOMotorClientasyncdefgo(db):# create model's indexesawaitUser.q(db).create_indexes()# CREATE# create using save# Note: if do_insert=False (default) save performs a replace# with upsert=True, so it does not raise if _id already exists# in db but replace document with that _id.u=awaitUser(name='Alexandro').save(db,do_insert=True)assertu.name=='Alexandro'assertu._id=='Alexandro'assertu.is_activeisTrueassertu.posts==[]assertu.quoteisNone# using queryu=awaitUser.q(db).create(name='Ihor',is_active=False)# READ# get by idu=awaitUser.q(db).get('Alexandro')assertu.name=='Alexandro'# findusers=awaitUser.q(db).find({User.is_active.s:True}).to_list(10)assertlen(users)==2# using for loopusers=[]asyncforuserinUser.q(db).find({User.is_active.s:False}):users.append(user)assertlen(users)==1# in Python 3.6 an up use async comprehensionsusers=[userasyncforuserinUser.q(db).find({})]assertlen(users)==3# UPDATEu=awaitUser.q(db).get('Ihor')u.is_active=Trueawaitu.save(db)assert(awaitUser.q(db).get('Ihor')).is_activeisTrue# using update (without data validation)# object is reloaded from db after update.awaitu.update(db,{'$push':{User.posts.s:ObjectId()}})# DELETEu=awaitUser.q(db).get('Ihor')awaitu.delete(db)loop=asyncio.get_event_loop()client=AsyncIOMotorClient(io_loop=loop)db=client.aiomongodel_testloop.run_until_complete(go(db))

验证

使用模型的validate方法验证模型的数据。如果 有任何无效的数据aiomongodel.errors.ValidationError 会提高。

classModel(Document):name=StrField(max_length=7)value=IntField(gt=5,lte=13)data=FloatField()defgo():m=Model(name='xxx',value=10,data=1.6)# validate data# should not raise any errorm.validate()# invalid data# note that there are no errors while creating# model with invalid datainvalid=Model(name='too long string',value=0)try:invalid.validate()exceptaiomongodel.errors.ValidationErrorase:asserte.as_dict()=={'name':'length is greater than 7','value':'value should be greater than 5','data':'field is required'}# using translation - you can translate messages# to your language or modify themtranslation={"field is required":"This field is required","length is greater than {constraint}":("Length of the field ""is greater than ""{constraint} characters"),# see all error messages in ValidationError docs# for missed messages default messages will be used}asserte.as_dict(translation=translation)=={'name':'Length of the field is greater than 7 characters','value':'value should be greater than 5','data':'This field is required'}

查询

asyncdefgo(db):# find returns a cursorcursor=User.q(db).find({},{'_id':1}).skip(1).limit(2)asyncforuserincursor:print(user.name)assertuser.is_activeisNone# we used projection# find oneuser=awaitUser.q(db).find_one({User.name.s:'Alexandro'})assertuser.name=='Alexandro'# updateawaitUser.q(db).update_many({User.is_active.s:True},{'$set':{User.is_active.s:False}})# deleteawaitUser.q(db).delete_many({})

模型继承

模型的层次结构可以通过从一个模型继承另一个模型来构建。 aiomongodel.Document类应该位于模型的层次结构中的某个位置 adnaiomongodel.EmbeddedDocument用于子文档。 请注意，字段是继承的，但元选项不是。

classMixin:value=IntField()classParent(Document):name=StrField()classChild(Mixin,Parent):# also has value and name fieldsrate=FloatField()classOtherChild(Child):# also has rate and name fieldsvalue=FloatField()# overwrite value field from MixinclassSubDoc(Mixin,EmbeddedDocument):# has value fieldpass

具有相同集合的模型继承

classMixin:is_active=BoolField(default=True)classUser(Mixin,Document):_id=StrField()role=StrField()name=SynonymField(_id)classMeta:collection='users'@classmethoddeffrom_mongo(cls,data):# create appropriate model when loading from dbifdata['role']=='customer':returnsuper(User,Customer).from_mongo(data)ifdata['role']=='admin':returnsuper(User,Admin).from_mongo(data)classCustomer(User):role=StrField(default='customer',choices=['customer'])# overwrite role fieldaddress=StrField()classMeta:collection='users'default_query={User.role.s:'customer'}classAdmin(User):role=StrField(default='admin',choices=['admin'])# overwrite role fieldrights=ListField(StrField(),default=lambda:list())classMeta:collection='users'default_query={User.role.s:'admin'}

交易

frommotor.motor_asyncioimportAsyncIOMotorClientasyncdefgo(db):# create collection before using transactionawaitUser.create_collection(db)asyncwithawaitdb.client.start_session()assession:try:asyncwiths.start_transaction():# all statements that use session inside this block# will be executed in one transaction# pass session to QuerySetawaitUser.q(db,session=session).create(name='user')# note session param# pass session to QuerySet methodawaitUser.q(db).update_one({User.name.s:'user'},{'$set':{User.is_active.s:False}},session=session)# note session usageassertawaitUser.q(db,session).count_documents({User.name.s:'user'})==1# session could be used in document crud methodsu=awaitUser(name='user2').save(db,session=session)awaitu.delete(db,session=session)raiseException()# simulate error in transaction blockexceptException:# transaction was not committedassertawaitUser.q(db).count_documents({User.name.s:'user'})==0loop=asyncio.get_event_loop()client=AsyncIOMotorClient(io_loop=loop)db=client.aiomongodel_testloop.run_until_complete(go(db))

0.2.0（2018-09-12）

将“要求”移到“电机”>；=2.0。

从MotorQuerySetCursor中删除count方法。

将会话支持添加到MotorQuerySet和Document。

将create_collection方法添加到Document。

修复python 3.7的MotorQuerySetCursor中的__aiter__。

不赞成MotorQuerySet的count方法。

不赞成Document的create方法。

0.1.0（2017-05-19）

第一个aiomongodel版本。