Python 操作 MongoDB

MongoDB 教程、高级教程：https://www.runoob.com/mongodb/mongodb-tutorial.html

官网：https://www.mongodb.com/

Pymongo 多线程安全。多进程不保证安全

Pymongo 多线程是安全的。多进程不保证是安全的。如果想要多进程同时写入同一个 collection中，需要在每个进程中创建自己的MongoClient对象。

官方文档：Frequently Asked Questions — PyMongo 3.12.1 documentation

简单封装

# -*- coding: utf-8 -*-import json
import pymongoconfig_dict = {'redis': {'host': '','port': '','db': '',},'mongodb': {'host': '127.0.0.1','port': 27017,'username': 'test','password': 'test','auth_db': 'admin'}
}class MongoDBOperate(object):def __init__(self, host=None, port=None, username=None, password=None, auth_db=None):super(MongoDBOperate, self).__init__()self.mongodb_host = hostself.mongodb_port = portself.mongodb_username = usernameself.mongodb_password = passwordself.auth_db = auth_dbself.__client = pymongo.MongoClient(host=self.mongodb_host, port=self.mongodb_port)# self.mongodb = self.__client.get_database('admin')self.mongodb_db = self.__client.get_database(self.auth_db)self.mongodb_db.authenticate(name=self.mongodb_username, password=self.mongodb_password)def __del__(self):"""析构函数，释放资源或者关闭连接:return:"""self.__client.close()passdef insert_one_or_many(self, db_name=None, tb_name=None, data=None):"""插入一条:param db_name:   数据库名:param tb_name:   集合名（表名）:param data:      插入的数据（文档）: dict 类型 或者 list 类型"""if db_name and tb_name and data:mongodb_tb = self.__client.get_database(db_name).get_collection(tb_name)# insert 时如存在会报错，可以使用 save 方法： self.__get_conn(db_name, tb_name).save(item)# self.__client.get_database(db_name).get_collection(tb_name).insert_one(item)func_write = mongodb_tb.insert_many if isinstance(data, list) else mongodb_tb.savefunc_write(data)else:raise Exception('db_name or tb_name is None')def find_one_or_many(self, db_name=None, tb_name=None, query_condition=None, field_list: list = None,query_one=False):""":param db_name: 数据库名:param tb_name: 数据表名:param query_condition: 查询条件:param field_list: 要查询的字段列表:param query_one: 是否只查询一条数据:return:"""if db_name and tb_name:mongodb_tb = self.__client.get_database(db_name).get_collection(tb_name)if query_condition:func_find = mongodb_tb.find_one if query_one else mongodb_tb.findquery_field = {field: 1 for field in field_list} if field_list else Noneresult_list = func_find(query_condition, query_field) if field_list else func_find(query_condition)else:# 没有查询条件，默认返回所有query_field = {field: 1 for field in field_list} if field_list else Noneresult_list = mongodb_tb.find({}, query_field) if field_list else mongodb_tb.find()for result in result_list:yield resultelse:raise Exception('db_name or tb_name is None')def update_one_or_many(self, db_name=None, tb_name=None, update_condition=None, update_one=True, update_id=None, ):"""Update a single document matching the filter ( update_condition ):param db_name::param tb_name::param update_condition:  filter 条件:param update_one::param update_id: 根据 _id 进行更新:return:"""if db_name and tb_name and update_condition:mongodb_tb = self.__client.get_database(db_name).get_collection(tb_name)if update_id:result = mongodb_tb.update_one({'_id': update_id}, update_condition)return resultfunc_update = mongodb_tb.update_one if update_one else mongodb_tb.update_manyresult = func_update(update_condition)else:raise Exception('db_name or tb_name or update_condition is None')return resultdef remove_one_or_many(self, db_name=None, tb_name=None, remove_condition=None, remove_one=True):if db_name and tb_name and remove_condition:mongodb_tb = self.__client.get_database(db_name).get_collection(tb_name)# remove_condition = { "name": "test_test" }func_delete = mongodb_tb.delete_one if remove_one else mongodb_tb.delete_manyfunc_delete(filter=remove_condition)else:raise Exception('db_name or tb_name or remove_condition is None')if __name__ == '__main__':pass

Python 连接 MongoDB 集群

Python 连接 MongoDB 的驱动为 pymongo, 基于 pymongo 有封装的 mongoengine，以及基于mongoengine 的 flask_mongoengie。

方法 1：使用 pymongo 连接 MongoDB 集群

示例代码：

from pymongo import MongoClient, ReadPreferencedef db_conn():client = MongoClient(['192.168.2.188:20000', '192.168.2.190:20000', '192.168.2.191:20000'])db_auth = client.admindb_auth.authenticate('lty', 'lty@100729')db = client.get_database('kafka_test', read_preference=ReadPreference.SECONDARY_PREFERRED)collection = db.ITS_Topic_CommandACKfor x in collection.find():print(x)if __name__ == '__main__':db_conn()pass

说明：

创建 client 客户端，通常写法为 MongoClient("mongo://host:port")，由于此处是连接集群，所以书写方式有所不同；
第二、第三行为登录；
获取数据库，通常写法为 client['db']，其中 db 为数据库名称，此处采用 get_database 方法，也只是写法不同；
获取集合，使用方法 db['collection']，其中 collection 为集群的名称，另外就是使用上面示例中的方法，直接带上集群名称即可，上面示例中 ITS_Topic_CommandACK 为集群的名称；
之后就可以进行增删改查了，上面示例中为轮询所有的数据；

示例代码 2：

建立了一个MongoDB集群，集群名称 replSet=rs0，包含以下 3 个节点：

172.16.250.233:27017 # SECONDARY
172.16.250.234:27017 # PRIMARY
172.16.250.237:27017 # SECONDARY

集群的 test 用户的密码为 “123456” ，只能访问 test 数据库。另外还有一个 MongoDB 数据库单独部署在 172.16.250.238，其 admin 用户的密码为 “123456” ，只能访问 admin 数据库。MongoDB 集群没有提供类似MySQL集群和Redis集群中的虚拟IP。在使用的时候需要客户端连接时指定MongoDB 集群的所有节点。

# coding=utf-8from pymongo import MongoClient# ################################ single mongo ##########################################
# ret : ['admin', 'config', 'local']
c = MongoClient(host="172.16.250.238", port=27017)
# c = MongoClient(host="mongodb://172.16.250.238:27017")
# c = MongoClient(host="mongodb://admin:123456@172.16.250.238:27017")
# c = MongoClient(host="mongodb://admin:123456@172.16.250.238:27017/admin")# ################################ mongo cluster ##########################################
# ret : ['test']
mongodb_cluster_ip = '172.16.250.233:27017,172.16.250.234:27017,172.16.250.237:27017'
c = MongoClient(f'mongodb://test:123456@{mongodb_cluster_ip}/test')
# c = MongoClient(f'mongodb://test:123456@{mongodb_cluster_ip}/test?replicaSet=rs0')
# c = MongoClient(f'mongodb://test:123456@{mongodb_cluster_ip}/test?readPreference=secondary')print(c.list_database_names())

方法 2：使用 mongoengine 连接 MongoDB 集群

示例代码：

# coding=utf-8from mongoengine import connect
from mongoengine import Document
from mongoengine import StringField# single mongo
connect('admin', host='mongodb://admin:123456@172.16.250.238:27017')# mongo cluster
cluster_ip = '172.16.250.233:27017,172.16.250.234:27017,172.16.250.237:27017'
# connect('test', host=f'mongodb://test:123456@{cluster_ip}/test?readPreference=secondary')class User(Document):  # 类名需要与数据库中的集合名称一致name = StringField(required=True, max_length=200)users = User.objects.all()  # 返回所有的文档对象列表for u in users:print("name:", u.name)
pass

方法 3：使用 flask_mongoengine 连接 MongoDB 集群

示例代码：

# coding=utf-8
from flask_mongoengine import MongoEngine
from flask import Flaskapp = Flask(__name__)# app.config['MONGODB_SETTINGS'] = {
#         'host': 'mongodb://admin:123456@172.16.250.238:27017/admin',
#         }# app.config.from_pyfile('mongo.cfg')# app.config['MONGODB_SETTINGS'] = {
#        'db': 'admin',
#        'username': 'admin',
#        'password': '123456',
#        'host': '172.16.250.238',
#        'port': 27017
#        }app.config['MONGODB_SETTINGS'] = [{'db': 'test','username': 'test','password': '123456','host': '172.16.250.233','port': 27017},{'db': 'test','username': 'test','password': '123456','host': '172.16.250.234','port': 27017}, {'db': 'test','username': 'test','password': '123456','host': '172.16.250.237','port': 27017}
]db = MongoEngine(app)class User(db.Document):name = db.StringField(required=True, max_length=50)users = User.objects.all()  # 返回所有的文档对象列表for u in users:print("name:", u.name)

配置文件 mongo.cfg 格式如下：

MONGODB_SETTINGS = {'host': 'mongodb://admin:123456@172.16.250.238:27017/admin',
}