Python 基于pymongo操作Mongodb学习总结

实践环境

Python 3.6.4

pymongo 4.1.1

pymongo-3.12.3-cp36-cp36m-win_amd64.whl

代码实践

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import datetime
import random
import pymongo
from pymongo import MongoClient
from bson.objectid import ObjectId



# # # # # # # # # 建立连接
# 方式1
# client = MongoClient() # 使用默认主机和端口连接本地Mongodb服务器

# 方式2：
# client = MongoClient("localhost", 27017)  # 也可以手动指定服务器和端口

# 方式3：采用url
user_name = 'tcems'
password = 'Password123456'

uri = f'mongodb://{user_name}:{password}@polartcems-mrs1.dbsit.sfcloud.local:24000,polartcems-mrs2.dbsit.sfcloud.local:24000,polartcems-mrs3.dbsit.sfcloud.local:24000'
client = MongoClient(uri)


# # # # # # # # # 获取数据库信息
databases = client.list_databases()
for database in databases:
    print(database) # 输出字典，形如：{'name': 'custom_db_name', 'sizeOnDisk': 5001216.0, 'empty': False}

# # # # # # # # # 获取数据库
my_test_db = client.tcems
# my_test_db = client['tcems'] # 如果不支持.属性方式访问(比如test-db)，可以考虑使用字典方式访问


# # # # # # # # # 获取当前数据库拥有的集合名称列表
collections = my_test_db.list_collection_names()
print(collections) # 输出名称列表，形如['YiLiuTemHumLog', 'test_collection', ...]

# # # # # # # # # 获取集合
collection = my_test_db.test_collection
# collection = my_test_db['test_collection'] # 如果不支持.属性方式访问(比如test-db)，可以考虑使用字典方式访问


# # # # # # # # # 索引
# 创建索引
# collection.create_index([("date", pymongo.ASCENDING)])
# collection.create_index([("field_name", pymongo.ASCENDING)], unique=True) # 创建唯一索引

# # # # # # # # #  集合文档操作
# # # # # # # # # 插入文档
# 逐条插入文档
post = {
    'author': 'Mike',
    'visitor_num': random.randint(0,100),
    'text': 'blog post of Mike!',
    'tags': ['mongodb', 'python', 'pymongo'],
    'date': datetime.datetime.now(tz=datetime.timezone.utc)
}

post_id = collection.insert_one(post).inserted_id  # 注意：insert_one函数返回 pymongo.results.InsertOneResult对象
print(post_id, type(post_id)) # 输出形如：65a881fffa04b0dc0e7a74bc <class 'bson.objectid.ObjectId'>    # 注意，获取的insert_id为 bson.objectid.ObjectId类型

try:
    post['_id'] = post_id
    collection.insert_one(post)  # 运行报错： pymongo.errors.DuplicateKeyError: E11000 duplicate key error collection:...
except pymongo.errors.DuplicateKeyError:
    print('id重复')
    # do something

# 批量插入文档
posts = [
    {
        'author': 'Mike',
        'visitor_num': random.randint(0,100),
        'text': 'blog post of Mike!',
        'tags': ['mongodb', 'python', 'pymongo'],
        'date': datetime.datetime.now(tz=datetime.timezone.utc)
    },
   {
        'author': 'Jack',
        'visitor_num': random.randint(0,100),
        'text': 'blog post of Jack!',
        'tags': ['mongodb', 'python', 'pymongo'],
        'date': datetime.datetime.now(tz=datetime.timezone.utc)
    }
]
res = collection.insert_many(posts) # insert_many函数返回 pymongo.results.InsertManyResult 对象
print(res.inserted_ids) # 获取插入记录的id，形如 [ObjectId('65a9423b782fc7838d729033'), ObjectId('65a9423b782fc7838d729034')]
inserted_ids = res.inserted_ids

# # # # # # # # # 查询文档
# 查询单条文档
res = collection.find_one() # 注意：如果集合中存在记录，则find_one返回字典对象，否则返回None
print(res, type(res))
print(res.get('_id'), str(res.get('_id')))  # 获取插入时自动生成的文档ID

# 查询时指定查询条件
print(collection.find_one({'author': 'Mike'})) # 文档author必须为Mike
print(collection.find_one({'_id': res.get('_id')})) # 按_id查询
print(collection.find_one({'_id': post_id}))  # 注意：_id值类型必须为ObjectId


# 批量查询
# 遍历所有记录
for post in collection.find():
    print(post)   # 此处，post为字典类型
    break

# 只查询满足条件的记录(注意：字典中逗号分隔的多个条件，默认的and关系
for post in collection.find({'visitor_num': {'$gte': 12, '$lte': 18}}): # 查找visitor_num大于等于12小于等于18的文档
    print(post)

for post in collection.find({'author': {'$ne': 'Mike'}}): # 查找 author 不等于 Mike的文档
    print(post)

for post in collection.find({'visitor_num': {'$ne': None}}): # 查找 visitor_num不为null的文档记录
    print(post)

for post in collection.find({'$or':[{'visitor_num':{'$gte':18}},{'author': 'Mike'}]}): # 查找 visitor_num 大于等于18，或者 author 等于 Mike 的文档
    print(post)

for post in collection.find({'visitor_num':{'$in': [58, 90, 41]}}): # 查找 visitor_num 值在数组 [58，90，41]中的文档
    print(post)

for post in collection.find({'author':{'$nin': ['Mike','Jack']}}): # 查找 visitor_num 不在数组 ['Mike','Jack'] 中的文档
    print(post)

  
# 限制返回文档数
for post in collection.find().limit(10): # 仅返回10条文档
    print(post)

# 查询排序
# 单个字段排序
# for post in collection.find().sort('visitor_num'): # 按 visitor_num 升序排序 # 注意：不存在排序字段的文档在有排序字段文档之上
for post in collection.find().sort('visitor_num', pymongo.ASCENDING): # 按 visitor_num 升序排序 pymongo.ASCENDING = 1
    print(post)

# 多字段排序
print('多字段排序1: 按 _id 升序，再按 visitor_num 降序')
for post in collection.find().sort([('_id', 1), ('visitor_num', pymongo.DESCENDING)]):
    print(post)

print('多字段排序2: 按 _id 降序，再按 visitor_num 升序')
for post in collection.find().sort([('_id', -1), ('visitor_num', 1)]):
    print(post)



# 聚合查询
start_time = datetime.datetime(2024, 1, 15, 6, 37, 37, 246000)
end_time = datetime.datetime(2024, 1, 15, 6, 44, 7, 239000)

# 查询当前集合中最大,最小文档ID
for record in collection.aggregate([
    {
        "$group": {
            "_id": None,
            "min_id": {"$min": '$_id'},
            "max_id": {"$max": '$_id'}
        }
    }
]):
    if record: # record为字典类型
        print(record.get('min_id'))
        print(record.get('max_id'))

# 查询当前集合中指定时间范围内最大,最小文档ID
for record in collection.aggregate([
    {
        '$match': {
            'date': {'$gte': start_time, '$lte': end_time}},
    },
    {
        "$group": {
            "_id": None,
            "min_id": {"$min": '$_id'},
            "max_id": {"$max": '$_id'}
        }
    }
]):
    if record: # record为字典类型
        print(record.get('min_id'))
        print(record.get('max_id'))



# 聚合管道
# 查询文档，按visitor_num降序排序，限制返回文档数为10--获取visitor_num top 10的记录
for post in collection.find().sort('visitor_num', pymongo.DESCENDING).limit(10):
    print(post)
# 或者
for post in collection.find().limit(10).sort('visitor_num', pymongo.ASCENDING):
    print(post)

# 分页查询
for post in collection.find().skip(5).limit(10): # skip(N) 跳过前N个文档，等价于mysql查询中的offset 根据limit参数值，返回第N+1条及往后文档
    print(post)


# 统计
# 获取文档总数
print(collection.count_documents({})) # 获取文档总数 # 注意：查询条件 {} 不能少，否则会报错
print(collection.count_documents({'author': 'Jack'})) # 获取author值为Jack的文档总数

# 获取最小值
min_visitor_num = collection.find_one(sort=[('visitor_num', pymongo.ASCENDING)]).get('visitor_num') # 获取最小 visitor_num
print('min_visitor_num', min_visitor_num)

# 获取最大值
max_visitor_num = collection.find_one(sort=[('visitor_num', pymongo.DESCENDING)]).get('visitor_num') # 获取最大 visitor_num
print(collection.find_one(sort=[('visitor_num', pymongo.DESCENDING)]))



# # # # # # # # # 更新文档
# 逐条更新
new_content = {
    'author': 'Json',
    'visitor_num': random.randint(0,100),
    'text': 'blog post of Json!',
    'tags': ['mongodb', 'python', 'pymongo'],
    'date': datetime.datetime.now(tz=datetime.timezone.utc)
}
query_condition = {'_id':  ObjectId('65a4d2b165b14a57a38a1504')}
collection.update_one(query_condition, {'$set': new_content})
print(collection.find_one({'_id': ObjectId('65a4d2b165b14a57a38a1504')})) # 查看更新后的文档

# 批量更新
query_condition = {'visitor_num': None}
collection.update_many(query_condition, {'$set': new_content}) # 批量更新visitor_num值为null的的文档内容为 new_content变量值


# # # # # # # # #  删除文档
# 逐条删除文档
res = collection.delete_one({'_id': post_id}) # 删除指定_id等于 post_id变量值的文档 # 注意：delete_one函数返回 pymongo.results.DeleteResult对象
# print(res.deleted_count, res.raw_result)
if res.deleted_count == 1:
    print('删除成功')

# 批量删除
res = collection.delete_many({'_id': {'$in': inserted_ids}}) # delete_many函数返回 pymongo.results.DeleteResult  对象
print(res.deleted_count) # 输出被删除文档数量

# 删除全部文档
collection.delete_many({})



# # # # # # # # # 删除集合
collection.drop()

说明：

如果连接用户名和密码包含诸如':', '/', '+' 及'@'保留字符，则使用前应该先进行编码，如下：

from urllib.parse import quote_plus

user_name = quote_plus('@username')
password = quote_plus('test_password+')
host = 'project.example.local'
port = 27017
uri = f'mongodb://{user_name}:{password}@{host}:{port}'
client = MongoClient(uri)

使用URI建立连接时，URI书写格式分这么几种情况：

需要验证密码

uri = 'mongodb://user_name:password@host:port/authentication_database'

说明：authentication_database ：授权数据库，可选配，默认admin，如果不为admin时，必须显示指明，否则会报类似如下错误：

pymongo.errors.OperationFailure: Authentication failed., full error: {'ok': 0.0, 'errmsg': 'Authentication failed.', 'code': 18, 'codeName': 'AuthenticationFailed'}

不需要密码验证

uri = 'mongodb://host:port'

集群模式

uri = 'mongodb://user_name:password@host1:port1,host2:port2,host3:port3,...hostN:portN/authentication_database'

MongoDB中的集合和数据库，都是懒惰地创建的——在第一个文档插入其中时创建的。

备注：笔者实践时发现，无法自动创建数据库和集合，会提示授权认证失败。

MongoDB中的数据使用JSON样式的文档表示（和存储）。在PyMongo中，使用字典来表示文档。例如，以下字典可能用于表示博客文章：

import datetime
post = {
    "author": "Mike",
    "text": "My first blog post!",
    "tags": ["mongodb", "python", "pymongo"],
    "date": datetime.datetime.now(tz=datetime.timezone.utc),
}

注意，文档可以包含本地Python类型（如datetime.datetime实例），这些类型将自动转换为相应的BSON类型或从相应的BSON类型转换。

实践时遇到类似如下错误：

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