python 之pymongo的CURD

本文主要是介绍python 之pymongo的CURD，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

pymongo的基本操作

前言

前面有了 mongodb命令操作学习 以及 pymongo单例连接池的学习，
接下来就学习pymongo 的增删查改操作，具体源码地址：https://gitee.com/allen-huang/python/blob/master/python-code/do-mongodb/test_curd.py

新增

1、新增一条记录

• 格式：

集合.insert_one(<字典结构>)

• 测试用例代码

def test_insert_one(self):"""单条插入数据@return:"""insert_data = {"uid": 1,"name": "0001","real_name": "黄飞鸿","email": "000@sina.cn","age": 30,"sex": 1,}res = MongoPool().test.user.insert_one(insert_data)print(res.inserted_id)pass

2、新增多条记录

• 格式：

集合.insert_one(<列表结构>)

• 测试用例代码：

def test_insert_many(self):"""批量插入数据@return:"""user_list = [{"uid": 2, "name": "0002", "real_name": "张三", "email": "001@sina.cn", "age": 29, "sex": 1},{"uid": 3, "name": "0003", "real_name": "魏延", "email": "002@sina.cn", "age": 28, "sex": 1},{"uid": 4, "name": "0004", "real_name": "刘备", "email": "003@sina.cn", "age": 38, "sex": 1},{"uid": 5, "name": "0005", "real_name": "关羽", "email": "004@sina.cn", "age": 32, "sex": 1},{"uid": 6, "name": "0006", "real_name": "张飞", "email": "005@sina.cn", "age": 33, "sex": 1},{"uid": 7, "name": "0007", "real_name": "赵云", "email": "006@sina.cn", "age": 21, "sex": 1},{"uid": 8, "name": "0008", "real_name": "马超", "email": "007@sina.cn", "age": 22, "sex": 1},{"uid": 9, "name": "0009", "real_name": "黄忠", "email": "008@sina.cn", "age": 24, "sex": 1},{"uid": 10, "name": "0010", "real_name": "孙尚香", "email": "009@sina.cn", "age": 28, "sex": 2},{"uid": 11, "name": "0011", "real_name": "黄月英", "email": "0010@sina.cn", "age": 27, "sex": 2},{"uid": 12, "name": "0012", "real_name": "诸葛亮", "email": "0011@sina.cn", "age": 29, "sex": 1},]res = MongoPool().test.user.insert_many(user_list)print(res.inserted_ids)pass

3、自定义_id 的新增

def test_insert_custom_ids(self):"""批量插入数据，自定义_id@return: """ids_list = [{"_id": 1, "name": "曹操"},{"_id": 2, "name": "曹丕"},{"_id": 3, "name": "典韦"},{"_id": 4, "name": "许诸"}]res = MongoPool().test.user_ids.insert_many(ids_list)print(res.inserted_ids)pass

更新

1、更新一条记录

• 格式：
  第三个参数 upsert=True 表示存在更新，不存在就新增

集合.update_one(filter=<更新条件>,
update=<更新的数据属于字典结构>,
upsert=True|False
)

• 不带 upsert测试用例代码：

def test_update_one(self):"""更新一条记录@return: """# 更新条件condition = {"uid": 5}# 更新的数据update_data = {"email": "0004@163.com"}res = MongoPool().test.user.update_one(filter=condition, update={"$set": update_data})print(res.modified_count)pass

• 带上 upsert测试用例代码：

def test_update_upsert(self):"""更新一条记录，存在更新，不存在新增@return:"""# 更新的数据update_data = {"uid": 13,"name": "0013","real_name": "铁桥三","email": "013sina.cn","age": 27,"sex": 1,}# 更新条件condition = {"uid": 13}# 更新的选项，存在更新，不存在新增upsert_option = Trueres = MongoPool().test.user.update_one(filter=condition, update={"$set": update_data}, upsert=upsert_option)print(res)pass

2、更新多条记录

• 格式：

格式和更新一条数据差不多

集合.update_many(filter=<更新条件>,
update=<更新的数据属于字典结构>,
upsert=True|False
)

• 测试用例1：

def test_update_many(self):"""更新多条记录，更新年龄大于30岁的人，武力值更新为95@return:"""# 更新的数据update_data = {"$set": {"force_value": 95}}# 更新条件condition = {"age": {"$gt": 30}}# 更新选项，存在更新，不存在就新增upsert_option = Trueres = MongoPool().test.user.update_many(filter=condition, update=update_data, upsert=upsert_option)print(res)pass

• 测试用例2：

def test_update_many_init(self):"""更新多条记录，筛选出name包含000的人员，武力值更新为90，初始化登录次数@return:"""# 更新的数据,$setOnInsert类似新增数据后进行初始化操作update_data = {"$set": {"force_value": 90}, "$setOnInsert": {"login_nums": 1}}# 更新条件condition = {"name": {"$regex": "^000"}}# 更新选项，存在更新，不存在就新增upsert_option = Trueres = MongoPool().test.user.update_many(filter=condition, update=update_data, upsert=upsert_option)print(res)pass

删除

删除一条记录

• 格式：

集合.delete_one(<删除条件>)

• 测试用例

def test_delete_one(self):"""删除单条数据@return:"""res = MongoPool().test.user.delete_one({'uid': 1})print(res.deleted_count)

删除多条记录

• 格式：

集合.delete_many(<删除条件>)

• 测试用例

def test_delete_many(self):"""批量删除uid<=2的记录@return:"""condition = {"uid": {"$lte": 2}}res = MongoPool().test.user.delete_many(condition)print(res)

查询

查询涉及到条件，运算符，正则表达式查询，排序，投影(查询部分字段)
find_one([查询条件]) 查询一条记录
find([查询条件]) 查询多条记录

条件查询

• 单个条件

def test_find_one(self):"""查询一条数据@return:"""coll = MongoPool().test.userdata = coll.find_one({"uid": 1})print(data)pass

• 多个条件

def test_find_more_filter(self):"""查询sex=1,年龄大于30岁的数据@return: """coll = MongoPool().test.userdata_list = coll.find({"sex": 1, "age": {"$gte": 30}})for data in data_list:print(data)

根据运算符查询

• 大于

def test_find_gt(self):"""查询age大于30岁的数据@return:"""coll = MongoPool().test.userdata_list = coll.find({"age": {"$gt": 30}})for data in data_list:print(data["uid"])

• 大于等于

def test_find_gte(self):"""查询 age 大于等于35岁的数据@return:"""data_list = MongoPool().test.user.find({"age": {"$gte": 35}})for data in data_list:print(data["real_name"])

• 小于

def test_find_lt(self):"""查询 age 小于30岁的数据@return:"""data_list = MongoPool().test.user.find({"age": {"$lt": 30}})for data in data_list:print(data)

• 小于等于

def test_find_lte(self):"""查询 age 小于等于27岁的数据@return:"""data_list = MongoPool().test.user.find({"age": {"$lte": 27}})for data in data_list:print(data)

• 不等于

def test_find_ne(self):"""查询sex!=1的数据@return:"""coll = MongoPool().test.userdata_list = coll.find({"sex": {"$ne": 1}})for data in data_list:print(data)

根据范围查找

• 数字范围start<=num<=end

def test_find_between(self):"""查询 age 在 20到25 之间的数据@return:"""data_list = MongoPool().test.user.find({"age": {"$gte": 20, "$lte": 25}})for data in data_list:print(data)

• 数组范围

def test_find_range(self):"""范围查找,指定 uid来查找@return:"""data_list = MongoPool().test.user.find({"uid": {"$in": [1, 2, 4, 5]}})for data in data_list:print(data)

根据正则表达式查询

• 类似 mysql 的 like 查找

def test_find_like(self):"""查询 name 包含 000的数据，类似mysql like这和mongodb的shell命令行不同，mongodb 的 shell 命令行是/000/"""data_list = MongoPool().test.user.find({"name": {"$regex": "000"}})for data in data_list:print(data)

• 以某个字符串开头

def test_find_pos(self):"""查找邮箱以某个字符串001开头的数据@return:"""data_list = MongoPool().test.user.find({"email": {"$regex": "^001"}})for data in data_list:print(data)

• 以某个字符串结尾

def test_find_rpos(self):"""查找邮箱以某个字符串163.com结尾的数据@return:"""data_list = MongoPool().test.user.find({"email": {"$regex": "163.com$"}})for data in data_list:print(data)

投影

投影是查询若干个字段，不是所有字段的数据

def test_find_projection(self):"""查询数据，只返回uid,real_name,age,sex@return:"""# 筛选条件filter = {"age": {"$lte": 25}}# 字段_id 不显示，uid,real_name,age,sex显示data_list = MongoPool().test.user.find(filter).project({"_id": 0, "uid": 1, "real_name": 1, "age": 1, "sex": 1})for data in data_list:print(data)

排序

在MongoDB中对结果进行排序，可使用sort()方法。
该方法接受一个包含字段列表及其排序顺序的参数，要指定排序顺序，使用1和-1，1用于升序，-1用于降序。

• 格式：

集合.find([查询条件]).sort({KEY:1})

• 示例代码

def test_find_sort(self):"""查找 sex=1,按年龄降序排序@return:"""coll = MongoPool().test.userdata_list = coll.find({"sex": 1}).sort({"age": -1})for data in data_list:print(data)

分页查询

mongodb 分页是使用 skip 和 limit来进行分页

def test_find_page(self):"""分页查找@return:"""coll = MongoPool().test.userpage_size = 10page_num = 1skip_num = (page_num - 1) * page_sizedata_list = coll.find().sort({"age": -1}).skip(skip_num).limit(page_size)for data in data_list:print(data)

管道聚合

待续…

这篇关于python 之pymongo的CURD的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---