Mongo ODM，基于pydantic
mongodantic的Python项目详细描述
Mongodanic公司
设置 在主文件应用程序中
frommongodanticimportinit_db_connection_paramsconnection_str='<your connection url>'db_name='<name of database>'# basicinit_db_connection_params(connection_str,db_name,max_pool_size=100)# if u use sslinit_db_connection_params(connection_str,db_name,max_pool_size=100,ssl=True,ssl_cert_path='<path to cert>')# extra paramsserver_selection_timeout_ms=50000# pymongo serverSelectionTimeoutMSconnect_timeout_ms=50000# pymongo connectTimeoutMSsocket_timeout_ms=50000# pymongo socketTimeoutMS
申报模型
^{pr2}$查询
banner=Banner.querybuilder.find_one()# return a banner model obj# skip and limitbanner_with_skip_and_limit=Banner.querybuilder.find(skip_rows=10,limit_rows=10)banner_data=Banner.querybuilder.find_one().data# return a dictbanners_queryset=Banner.querybuilder.find()# return QuerySet objectbanners_dict=Banner.querybuilder.find().datalist_of_banners=Banner.querybuilder.find().listbanners_generator=Banner.querybuilder.find().generator# generator of Banner objectsbanners_generator_of_dicts=Banner.querybuilder.find().data_generator# generator of Banner objectscount,banners=Banner.querybuilder.find_with_count()# return tuple(int, QuerySet)serializeble_fields=Banner.querybuilder.find().serialize(['utm','banner_id','name'])# return list with dict like {'utm':..., 'banner_id': ..,'name': ...}generator_serializeble_fields=Banner.querybuilder.find().serialize_generator(['utm','banner_id','name'])# return generatorjson_serializeble_fields=Banner.querybuilder.find().serialize_json(['utm','banner_id','name'])# returnn json str serializeble# countcount=Banner.querybuilder.count(name='test')# insert queriesBanner.querybuilder.insert_one(banner_id=1,name='test',utm={'utm_source':'yandex','utm_medium':'cpc'})banners=[Banner(banner_id=2,name='test2',utm={}),Banner(banner_id=3,name='test3',utm={})]Banner.querybuilder.insert_many(banners)# list off models obj, or dictsBanner.querybuilder.bulk_create(banners,batch_size=1000)# insert_many with batch# update queriesBanner.querybuilder.update_one(banner_id=1,name__set='updated')# parameters that end __set - been updated Banner.querybuilder.update_many(name__set='update all names')# delete queriesBanner.querybuilder.delete_one(banner_id=1)# delete one rowBanner.querybuilder.delete_many(banner_id=1)# delete many rows# extra queriesBanner.querybuilder.find(banner_id__in=[1,2])# get data in listBanner.querybuilder.find(banner_id__range=[1,10])# get date from 1 to 10Banner.querybuilder.find(name__regex='^test')# regex queryBanner.querybuilder.find(name__startswith='t')# startswith queryBanner.querybuilder.find(name__endswith='t')# endswith queryBanner.querybuilder.find(name__not_startswith='t')# not startswith queryBanner.querybuilder.find(name__not_endswith='t')# not endswith queryBanner.querybuilder.find(name__nin=[1,2])# not in listBanner.querybuilder.find(name__ne='test')# != testBanner.querybuilder.find(banner_id__gte=1,banner_id__lte=10)# id >=1 and id <=10Banner.querybuilder.find(banner_id__gt=1,banner_id__lt=10)# id >1 and id <10Banner.querybuilder.find_one(banner_id=1,utm__utm_medium='cpm')# find banner where banner_id=1, and utm['utm_medium'] == 'cpm'Banner.querybuilder.update_one(banner_id=1,utm__utm_source__set='google')# update utms['utm_source'] in Banner# find and updateBanner.querybuilder.find_and_update(banner_id=1,name__set='updated',projection_fields=['name':True])# return {'name': 'updated}Banner.querybuilder.find_and_update(banner_id=1,name__set='updated')# return Banner obj# find and replaceBanner.querybuilder.find_and_update(banner_id=1,Banner(banner_id=1,name='uptated'),projection={'name':True})# return {'name': 'updated}Banner.querybuilder.find_and_update(banner_id=1,Banner(banner_id=1,name='uptated'))# return Banner obj# bulk operationsfromrandomimportrandintbanners=Banner.querybuilder.find()to_update=[]forbannerinbanners:banner.banner_id=randint(1,100)to_update.append(banner)Banner.querybuilder.bulk_update(banners,updated_fields=['banner_id'])# bulk update or createbanners=[Banner(banner_id=23,name='new',utms={}),Banner(banner_id=1,name='test',utms={})]Banner.querybuilder.bulk_update_or_create(banners,query_fields=['banner_id'])# aggregate with sum, min, maxclassStats(MongoModel):id:intcost:floatclicks:intshows:intdate:strfrommongodantic.aggregationimportSum,Min,MaxStats.querybuilder.aggregate(date='2020-01-20',aggregation=Sum('cost'))Stats.querybuilder.aggregate(date='2020-01-20',aggregation=Min('clicks'))Stats.querybuilder.aggregate(date='2020-01-20',aggregation=Max('shows'))# sessionsfrommongodantic.sessionimportSessionwithSession()assession:Banner.querybuilder.find(skip_rows=1,limit_rows=1,session=session).data# logicalfrommongodantic.logicalimportQuerydata=Banner.querybuilder.find_one(Query(name='test')|Query(name__regex='testerino'))
- 项目
标签：
欢迎加入QQ群-->： 979659372
