使用Python操作MongoDB数据库 - ToB企服应用市场:ToB评测及商务社交产业平台

mongod --dbpath D:\mongoDB\data\db

复制代码

use admin
db.createUser({
user: 'yourUsername',
pwd: 'yourPassword',
roles: [ { role: 'userAdminAnyDatabase', db: 'admin' } ]
})

复制代码

security:
authorization: enabled

复制代码

mongod --config D:\mongoDB\bin\mongod.cfg --dbpath=D:\mongoDB\data --logpath=D:\mongoDB\data\mongodb.log

复制代码

mongo -u user -p password123 --authenticationDatabase admin

复制代码

mongod --config D:\mongoDB\bin\mongod.cfg --dbpath=D:\mongoDB\data --logpath=D:\mongoDB\data\mongodb.log -install -serviceName "MongoDB"

复制代码

pip install pymongo

复制代码

from pymongo import MongoClient
# 连接到本地 MongoDB 服务
# 默认端口是 27017
client = MongoClient('localhost', 27017)
# 选择或创建数据库
db = client['mydatabase'] # 如果数据库不存在，将会自动创建
# 选择或创建集合
collection = db['mycollection'] # 集合类似于关系数据库中的表

复制代码

from pymongo import MongoClient
# 连接到本地 MongoDB 服务
# 默认端口是 27017
client = MongoClient(host='127.0.0.1', port=27017,username="admin",password="*****")
# 选择或创建数据库
db = client['mydatabase'] # 如果数据库不存在，将会自动创建
# 选择或创建集合
collection = db['mycollection'] # 集合类似于关系数据库中的表

复制代码

# 插入单个文档
post = {"author": "John", "text": "My first blog post!", "tags": ["mongodb", "python", "pymongo"]}
post_id = collection.insert_one(post).inserted_id
print("Inserted post with ID:", post_id)
# 插入多个文档
posts = [
{"author": "Alice", "text": "Another post", "tags": ["web", "databases"]},
{"author": "Bob", "text": "Yet another post", "tags": ["developers", "blogging"]},
]
collection.insert_many(posts)

复制代码

# 查询所有文档
for post in collection.find():
print(post)
# 使用查询条件
for post in collection.find({"author": "John"}):
print(post)
# 投影（只返回指定字段）
for post in collection.find({}, {"_id": 0, "text": 1}):
print(post)

复制代码

# 更新单个文档
collection.update_one({"author": "John"}, {"$set": {"text": "Updated text"}})
# 更新多个文档
collection.update_many({"tags": "mongodb"},{"$set": {"text": "MongoDB post updated"}})

复制代码

collection.delete_one({"author": "Alice"})
# 删除多个文档
collection.delete_many({"tags": "blogging"})

复制代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/7/19 下午3:19
@Author : Bill Fang
@File : MongoDBUtil.py
@Desc :
"""
import pymongo
class MongoDBUtil:
"""
MongoDB工具类
"""
def __init__(self, ip="127.0.0.1", db_name=None, port=27017):
"""构造函数"""
self.client = pymongo.MongoClient(host=ip, port=port,username="admin",password="****")
self.database = self.client[db_name]
print(self.database)
def create_database(self, db_name):
"""创建数据库"""
return self.client.get_database(db_name)
def drop_database(self, db_name):
"""删除数据库"""
return self.client.drop_database(db_name)
def select_database(self, db_name):
"""使用数据库"""
self.database = self.client[db_name]
return self.database
def get_database(self, db_name):
"""使用数据库"""
# return self.client[db_name]
return self.client.get_database(db_name)
def list_database_names(self):
"""获取所有数据库列表"""
return self.client.list_database_names()
def create_collection(self, collect_name):
"""创建集合"""
collect = self.database.get_collection(collect_name)
if(collect is not None):
print("collection %s already exists" % collect_name)
return collect
return self.database.create_collection(collect_name)
def drop_collection(self, collect_name):
"""获取所有集合名称"""
return self.database.drop_collection(collect_name)
def get_collection(self, collect_name):
"""获取集合"""
return self.database.get_collection(collect_name)
def list_collection_names(self):
"""获取所有集合名称"""
return self.database.list_collection_names()
def insert(self, collect_name, documents):
"""插入单条或多条数据"""
return self.database.get_collection(collect_name).insert(documents)
def insert_one(self, collect_name, document):
"""插入一条数据"""
return self.database.get_collection(collect_name).insert_one(document)
def insert_many(self, collect_name, documents):
"""插入多条数据"""
return self.database.get_collection(collect_name).insert_many(documents)
def delete_one(self, collect_name, filter, collation=None, hint=None, session=None):
"""删除一条记录"""
return self.database.get_collection(collect_name).delete_one(filter, collation, hint, session)
def delete_many(self, collect_name, filter, collation=None, hint=None, session=None):
"""删除所有记录"""
return self.database.get_collection(collect_name).delete_many(filter, collation, hint, session)
def find_one_and_delete(self, collect_name, filter, projection=None, sort=None, hint=None, session=None, **kwargs):
"""查询并删除一条记录"""
return self.database.get_collection(collect_name).find_one_and_delete(filter, projection, sort, hint, session, **kwargs)
def count_documents(self, collect_name, filter, session=None, **kwargs):
"""查询文档数目"""
return self.database.get_collection(collect_name).count_documents(filter, session, **kwargs)
def find_one(self, collect_name, filter=None, *args, **kwargs):
"""查询一条记录"""
return self.database.get_collection(collect_name).find_one(filter, *args, **kwargs)
def find(self, collect_name, *args, **kwargs):
"""查询所有记录"""
return self.database.get_collection(collect_name).find(*args, **kwargs)
def update(self, collect_name, spec, document, upsert=False, manipulate=False,
multi=False, check_keys=True, **kwargs):
"""更新所有记录"""
return self.database.get_collection(collect_name).update(spec, document,
upsert, manipulate, multi, check_keys, **kwargs)
def update_one(self, collect_name, filter, update, upsert=False, bypass_document_validation=False,
collation=None, array_filters=None, hint=None, session=None):
"""更新一条记录"""
return self.database.get_collection(collect_name).update_one(filter, update,
upsert, bypass_document_validation, collation, array_filters, hint, session)
def update_many(self, collect_name, filter, update, upsert=False, array_filters=None,
bypass_document_validation=False, collation=None, hint=None, session=None):
"""更新所有记录"""
return self.database.get_collection(collect_name).update_many(filter, update,
upsert, array_filters, bypass_document_validation, collation, hint, session)
def find_one_and_update(self, collect_name, filter, update, projection=None, sort=None, upsert=False,
return_document=False, array_filters=None, hint=None, session=None, **kwargs):
"""查询并更新一条记录"""
return self.database.get_collection(collect_name).find_one_and_update(filter, update, projection,
sort, upsert, return_document, array_filters, hint, session, **kwargs)

复制代码