Python 基于pymongo操作Mongodb学习总结

王才志

实践环境

Python 3.6.4
pymongo 4.1.1
pymongo-3.12.3-cp36-cp36m-win_amd64.whl
下载地址：https://pypi.org/simple/pymongo/
代码实践

1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
4. import datetime
5. import random
6. import pymongo
7. from pymongo import MongoClient
8. from bson.objectid import ObjectId
12. # # # # # # # # # 建立连接
13. # 方式1
14. # client = MongoClient() # 使用默认主机和端口连接本地Mongodb服务器
16. # 方式2：
17. # client = MongoClient("localhost", 27017)  # 也可以手动指定服务器和端口
19. # 方式3：采用url
20. user_name = 'tcems'
21. password = 'Password123456'
23. uri = f'mongodb://{user_name}:{password}@polartcems-mrs1.dbsit.sfcloud.local:24000,polartcems-mrs2.dbsit.sfcloud.local:24000,polartcems-mrs3.dbsit.sfcloud.local:24000'
24. client = MongoClient(uri)
27. # # # # # # # # # 获取数据库信息
28. databases = client.list_databases()
29. for database in databases:
30.     print(database) # 输出字典，形如：{'name': 'custom_db_name', 'sizeOnDisk': 5001216.0, 'empty': False}
32. # # # # # # # # # 获取数据库
33. my_test_db = client.tcems
34. # my_test_db = client['tcems'] # 如果不支持.属性方式访问(比如test-db)，可以考虑使用字典方式访问
37. # # # # # # # # # 获取当前数据库拥有的集合名称列表
38. collections = my_test_db.list_collection_names()
39. print(collections) # 输出名称列表，形如['YiLiuTemHumLog', 'test_collection', ...]
41. # # # # # # # # # 获取集合
42. collection = my_test_db.test_collection
43. # collection = my_test_db['test_collection'] # 如果不支持.属性方式访问(比如test-db)，可以考虑使用字典方式访问
46. # # # # # # # # # 索引
47. # 创建索引
48. # collection.create_index([("date", pymongo.ASCENDING)])
49. # collection.create_index([("field_name", pymongo.ASCENDING)], unique=True) # 创建唯一索引
51. # # # # # # # # #  集合文档操作
52. # # # # # # # # # 插入文档
53. # 逐条插入文档
54. post = {
55.     'author': 'Mike',
56.     'visitor_num': random.randint(0,100),
57.     'text': 'blog post of Mike!',
58.     'tags': ['mongodb', 'python', 'pymongo'],
59.     'date': datetime.datetime.now(tz=datetime.timezone.utc)
60. }
62. post_id = collection.insert_one(post).inserted_id  # 注意：insert_one函数返回 pymongo.results.InsertOneResult对象
63. print(post_id, type(post_id)) # 输出形如：65a881fffa04b0dc0e7a74bc <class 'bson.objectid.ObjectId'>    # 注意，获取的insert_id为 bson.objectid.ObjectId类型
65. try:
66.     post['_id'] = post_id
67.     collection.insert_one(post)  # 运行报错： pymongo.errors.DuplicateKeyError: E11000 duplicate key error collection:...
68. except pymongo.errors.DuplicateKeyError:
69.     print('id重复')
70.     # do something
72. # 批量插入文档
73. posts = [
74.     {
75.         'author': 'Mike',
76.         'visitor_num': random.randint(0,100),
77.         'text': 'blog post of Mike!',
78.         'tags': ['mongodb', 'python', 'pymongo'],
79.         'date': datetime.datetime.now(tz=datetime.timezone.utc)
80.     },
81.    {
82.         'author': 'Jack',
83.         'visitor_num': random.randint(0,100),
84.         'text': 'blog post of Jack!',
85.         'tags': ['mongodb', 'python', 'pymongo'],
86.         'date': datetime.datetime.now(tz=datetime.timezone.utc)
87.     }
88. ]
89. res = collection.insert_many(posts) # insert_many函数返回 pymongo.results.InsertManyResult 对象
90. print(res.inserted_ids) # 获取插入记录的id，形如 [ObjectId('65a9423b782fc7838d729033'), ObjectId('65a9423b782fc7838d729034')]
91. inserted_ids = res.inserted_ids
93. # # # # # # # # # 查询文档
94. # 查询单条文档
95. res = collection.find_one() # 注意：如果集合中存在记录，则find_one返回字典对象，否则返回None
96. print(res, type(res))
97. print(res.get('_id'), str(res.get('_id')))  # 获取插入时自动生成的文档ID
99. # 查询时指定查询条件
100. print(collection.find_one({'author': 'Mike'})) # 文档author必须为Mike
101. print(collection.find_one({'_id': res.get('_id')})) # 按_id查询
102. print(collection.find_one({'_id': post_id}))  # 注意：_id值类型必须为ObjectId
105. # 批量查询
106. # 遍历所有记录
107. for post in collection.find():
108.     print(post)   # 此处，post为字典类型
109.     break
111. # 只查询满足条件的记录(注意：字典中逗号分隔的多个条件，默认的and关系
112. for post in collection.find({'visitor_num': {'$gte': 12, '$lte': 18}}): # 查找visitor_num大于等于12小于等于18的文档
113.     print(post)
115. for post in collection.find({'author': {'$ne': 'Mike'}}): # 查找 author 不等于 Mike的文档
116.     print(post)
118. for post in collection.find({'visitor_num': {'$ne': None}}): # 查找 visitor_num不为null的文档记录
119.     print(post)
121. for post in collection.find({'$or':[{'visitor_num':{'$gte':18}},{'author': 'Mike'}]}): # 查找 visitor_num 大于等于18，或者 author 等于 Mike 的文档
122.     print(post)
124. for post in collection.find({'visitor_num':{'$in': [58, 90, 41]}}): # 查找 visitor_num 值在数组 [58，90，41]中的文档
125.     print(post)
127. for post in collection.find({'author':{'$nin': ['Mike','Jack']}}): # 查找 visitor_num 不在数组 ['Mike','Jack'] 中的文档
128.     print(post)
130.   
131. # 限制返回文档数
132. for post in collection.find().limit(10): # 仅返回10条文档
133.     print(post)
135. # 查询排序
136. # 单个字段排序
137. # for post in collection.find().sort('visitor_num'): # 按 visitor_num 升序排序 # 注意：不存在排序字段的文档在有排序字段文档之上
138. for post in collection.find().sort('visitor_num', pymongo.ASCENDING): # 按 visitor_num 升序排序 pymongo.ASCENDING = 1
139.     print(post)
141. # 多字段排序
142. print('多字段排序1: 按 _id 升序，再按 visitor_num 降序')
143. for post in collection.find().sort([('_id', 1), ('visitor_num', pymongo.DESCENDING)]):
144.     print(post)
146. print('多字段排序2: 按 _id 降序，再按 visitor_num 升序')
147. for post in collection.find().sort([('_id', -1), ('visitor_num', 1)]):
148.     print(post)
152. # 聚合查询
153. start_time = datetime.datetime(2024, 1, 15, 6, 37, 37, 246000)
154. end_time = datetime.datetime(2024, 1, 15, 6, 44, 7, 239000)
156. # 查询当前集合中最大,最小文档ID
157. for record in collection.aggregate([
158.     {
159.         "$group": {
160.             "_id": None,
161.             "min_id": {"$min": '$_id'},
162.             "max_id": {"$max": '$_id'}
163.         }
164.     }
165. ]):
166.     if record: # record为字典类型
167.         print(record.get('min_id'))
168.         print(record.get('max_id'))
170. # 查询当前集合中指定时间范围内最大,最小文档ID
171. for record in collection.aggregate([
172.     {
173.         '$match': {
174.             'date': {'$gte': start_time, '$lte': end_time}},
175.     },
176.     {
177.         "$group": {
178.             "_id": None,
179.             "min_id": {"$min": '$_id'},
180.             "max_id": {"$max": '$_id'}
181.         }
182.     }
183. ]):
184.     if record: # record为字典类型
185.         print(record.get('min_id'))
186.         print(record.get('max_id'))
190. # 聚合管道
191. # 查询文档，按visitor_num降序排序，限制返回文档数为10--获取visitor_num top 10的记录
192. for post in collection.find().sort('visitor_num', pymongo.DESCENDING).limit(10):
193.     print(post)
194. # 或者
195. for post in collection.find().limit(10).sort('visitor_num', pymongo.ASCENDING):
196.     print(post)
198. # 分页查询
199. for post in collection.find().skip(5).limit(10): # skip(N) 跳过前N个文档，等价于mysql查询中的offset 根据limit参数值，返回第N+1条及往后文档
200.     print(post)
203. # 统计
204. # 获取文档总数
205. print(collection.count_documents({})) # 获取文档总数 # 注意：查询条件 {} 不能少，否则会报错
206. print(collection.count_documents({'author': 'Jack'})) # 获取author值为Jack的文档总数
208. # 获取最小值
209. min_visitor_num = collection.find_one(sort=[('visitor_num', pymongo.ASCENDING)]).get('visitor_num') # 获取最小 visitor_num
210. print('min_visitor_num', min_visitor_num)
212. # 获取最大值
213. max_visitor_num = collection.find_one(sort=[('visitor_num', pymongo.DESCENDING)]).get('visitor_num') # 获取最大 visitor_num
214. print(collection.find_one(sort=[('visitor_num', pymongo.DESCENDING)]))
218. # # # # # # # # # 更新文档
219. # 逐条更新
220. new_content = {
221.     'author': 'Json',
222.     'visitor_num': random.randint(0,100),
223.     'text': 'blog post of Json!',
224.     'tags': ['mongodb', 'python', 'pymongo'],
225.     'date': datetime.datetime.now(tz=datetime.timezone.utc)
226. }
227. query_condition = {'_id':  ObjectId('65a4d2b165b14a57a38a1504')}
228. collection.update_one(query_condition, {'$set': new_content})
229. print(collection.find_one({'_id': ObjectId('65a4d2b165b14a57a38a1504')})) # 查看更新后的文档
231. # 批量更新
232. query_condition = {'visitor_num': None}
233. collection.update_many(query_condition, {'$set': new_content}) # 批量更新visitor_num值为null的的文档内容为 new_content变量值
236. # # # # # # # # #  删除文档
237. # 逐条删除文档
238. res = collection.delete_one({'_id': post_id}) # 删除指定_id等于 post_id变量值的文档 # 注意：delete_one函数返回 pymongo.results.DeleteResult对象
239. # print(res.deleted_count, res.raw_result)
240. if res.deleted_count == 1:
241.     print('删除成功')
243. # 批量删除
244. res = collection.delete_many({'_id': {'$in': inserted_ids}}) # delete_many函数返回 pymongo.results.DeleteResult  对象
245. print(res.deleted_count) # 输出被删除文档数量
247. # 删除全部文档
248. collection.delete_many({})
252. # # # # # # # # # 删除集合
253. collection.drop()

复制代码

说明：

• 如果连接用户名和密码包含诸如':', '/', '+' 及'@'保留字符，则使用前应该先进行编码，如下：
  1. from urllib.parse import quote_plus
  3. user_name = quote_plus('@username')
  4. password = quote_plus('test_password+')
  5. host = 'project.example.local'
  6. port = 27017
  7. uri = f'mongodb://{user_name}:{password}@{host}:{port}'
  8. client = MongoClient(uri)

  复制代码
• 使用URI建立连接时，URI书写格式分这么几种情况：
  
  
  
  • 需要验证密码
    1. uri = 'mongodb://user_name:password@host:port/authentication_database'

    复制代码
    说明：authentication_database ：授权数据库，可选配，默认admin，如果不为admin时，必须显示指明，否则会报类似如下错误：
    1. pymongo.errors.OperationFailure: Authentication failed., full error: {'ok': 0.0, 'errmsg': 'Authentication failed.', 'code': 18, 'codeName': 'AuthenticationFailed'}

    复制代码
  • 不需要密码验证
    1. uri = 'mongodb://host:port'

    复制代码
  • 集群模式
    1. uri = 'mongodb://user_name:password@host1:port1,host2:port2,host3:port3,...hostN:portN/authentication_database'

    复制代码
  
  
• MongoDB中的集合和数据库，都是懒惰地创建的——在第一个文档插入其中时创建的。
  备注：笔者实践时发现，无法自动创建数据库和集合，会提示授权认证失败。
  
• MongoDB中的数据使用JSON样式的文档表示（和存储）。在PyMongo中，使用字典来表示文档。例如，以下字典可能用于表示博客文章：
  1. import datetime
  2. post = {
  3.     "author": "Mike",
  4.     "text": "My first blog post!",
  5.     "tags": ["mongodb", "python", "pymongo"],
  6.     "date": datetime.datetime.now(tz=datetime.timezone.utc),
  7. }

  复制代码
  注意，文档可以包含本地Python类型（如datetime.datetime实例），这些类型将自动转换为相应的BSON类型或从相应的BSON类型转换。
  
• 实践时遇到类似如下错误：
  1. pymongo.errors.ConfigurationError: Server at * reports wire version 5, but this version of PyMongo requires at least 6 (MongoDB 3.6).

  复制代码
  原因分析：
  错误信息提示来看，使用的PyMongo版本与MongoDB服务器的Wire版本不兼容。服务器wire版本为5，而该PyMongo要求至少wire版本6(MongoDB 3.6)
  解决方法：
  
  
  
  • 升级MongoDB服务器：将MongoDB服务器升级到PyMongo所需的版本，即MongoDB 3.6或更高版本。
    
  • 降级PyMongo版本：如果不能升级MongoDB服务器，可以尝试降级PyMongo版本，以匹配MongoDB服务器版本。
    笔者实践时选择了降低PyMongo版本为 pymongo-3.12.3-cp36-cp36m-win_amd64.whl
    
  
  

参考链接

https://pymongo.readthedocs.io/en/stable/tutorial.html
https://pymongo.readthedocs.io/en/stable/api/pymongo/cursor.html#pymongo.cursor.Cursor.sort
https://pymongo.readthedocs.io/en/stable/api/pymongo/collection.html#pymongo.collection.Collection.count_documents
https://www.mongodb.com/docs/manual/reference/operator/query/
https://www.mongodb.com/docs/manual/reference/operator/aggregation/
https://www.mongodb.com/docs/manual/reference/operator/aggregation-pipeline/
https://www.mongodb.com/docs/manual/reference/operator/aggregation/match/#mongodb-pipeline-pipe.-match

来源:https://www.cnblogs.com/shouke/p/17993189
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！