您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

python3操作MongoDB的crud以及聚合案例，代碼可直接運行（python經典編程案例）

編輯：Python

目錄：基礎篇（能解決工作中80的問題）
01 MongoDB的概述、應用場景、下載方式、連接方式和發展歷史等

02 MongoDB數據類型、重要概念以及shell常用指令

03 MongoDB文檔的各種增加、更新、刪除操作總結

04 MongoDB各種查詢操作以及聚合操作總結

05 MongoDB對列的各種操作總結

python3操作MongoDB的各種案例

一. 插入數據案例

# -*- encoding: utf-8 -*-
import time
import pymongo
import datetime
# 創建對象
client = pymongo.MongoClient('mongodb://賬號:密碼@主機:端口號/?authSource=admin')
# 連接DB數據庫
db = client['數據庫名']
def insert_one():
# 連接集合user，集合類似於關系數據庫的數據表; 如果集合不存在，就會新建集合user
user_collection = db.user_demo
# 設置文檔格式（文檔即我們常說的數據）
user_info = {

"_id": 105,
"author": "小綠",
"text": "Python開發",
"tags": ["mongodb", "pymongo"],
"date": datetime.datetime.now()}
# 使用insert_one單條添加文檔，inserted_id獲取寫入後的id
# 添加文檔時，如果文檔尚未包含"_id"鍵，就會自動添加"_id"。"_id"的值在集合中必須是唯一的
# inserted_id用於獲取添加後的id，若不需要，則可以去掉
user_id = user_collection.insert_one(user_info).inserted_id
print("user id is ", user_id)
def insert_many():
#批量添加
user_infos = [{

"_id": 101,
"author": "小黃",
"text": "Python開發",
"tags": ["mongodb", "python", "pymongo"],
"date": datetime.datetime.utcnow()},
{

"_id": 102,
"author": "小黃_A",
"text": "Python開發_A",
"tags": {
"db":"Mongodb","lan":"Python","modle":"Pymongo"},
"date": datetime.datetime.utcnow()},
]
user_collection = db.user_insert_many
# inserted_ids用於獲取添加後的id，若不需要，則可以直接去掉
user_id = user_collection.insert_many(user_infos).inserted_ids
print("user id is ", user_id)
def bulk_insert_data():
from pymongo import UpdateOne
data_list = [{
'user_id': 5, 'name': '張三1', 'age': 27, 'email': '[email protected]'},
{
'user_id': 6, 'name': '李四1', 'age': 26, 'email': '[email protected]'},
{
'user_id': 7, 'name': '王五1', 'age': 29, 'email': '[email protected]'},
{
'user_id': 8, 'name': '趙六1', 'age': 26, 'email': '[email protected]'}]
bulk_data_list = []
for data in data_list:
one = UpdateOne({
"_id": data['user_id']}, {

"$set": {
"name": data['name'],
"age": data['age'],
"email": data['email'],
"date": datetime.datetime.now()}}, upsert=True)
bulk_data_list.append(one)
try:
collection_item = db.bulk_insert_demo
collection_item.bulk_write(bulk_data_list)
except Exception as e:
print(f'e: {
e}')
print(f"{
time.strftime('%Y-%m-%d %H:%M:%S')}, 已存mongo: {
len(bulk_data_list)}條")
if __name__ == '__main__':
# 插入單條數據
insert_one()
# 插入多條數據
# insert_many()
# 批量插入
# bulk_insert_data()

二. 查詢數據案例

# -*- encoding: utf-8 -*-
import re
import pymongo
# 創建對象
# client = pymongo.MongoClient()
client = pymongo.MongoClient('mongodb://賬號:密碼@主機:端口號/?authSource=admin')
# 連接DB數據庫
db = client['數據庫名']
def find_by_condition():
# 連接集合user，集合類似於關系數據庫的數據表, 如果集合不存在，就會新建集合user
user_collection = db.user
# 1. 查詢文檔: find({"_id":101})，其中{"_id":101}為查詢條件, 若查詢條件為空，則默認查詢全部
# find_value = user_collection.find({"_id": 103})
# print(list(find_value))
# 2. 如果要實現多條件查詢，$and和$or，使用方法如下：
# AND條件查詢
# find_value = user_collection.find({"$and": [{"_id": 104}, {"author": "小藍"}]})
# print(list(find_value))
# OR條件查詢
# find_value = user_collection.find({"$or": [{"author": "小黃_A"}, {"author": "小黃"}]})
# print(list(find_value))
# 3. 根據范圍查找: $gt: 大於, $gte: 大於等於, $lt: 小於, $lte: 小於等於, $ne: 不等於,
# 如查找id>102且id<104(_id=101)的文檔
# find_value = user_collection.find({"_id": {"$gt": 102, "$lt": 104}})
# print(list(find_value))
# 查找id在[100,101]的文檔
# find_value = user_collection.find({"_id": {"$in": [100, 101]}})
# print(list(find_value))
# find_value = user_collection.find({"and": [{"_id": {"$gt": 102, "$lt": 105}},
# {"_id": {"$in": [100, 105]}}]})
# print(list(find_value))
# 4. 模糊查詢實際上是加入正則表達式實現的
# # 方法一
# find_value = user_collection.find({"author": {"$regex": ".*小.*"}})
# print(list(find_value))
# #方法二
regex = re.compile(".*小.*")
find_value = user_collection.find({
"author": regex})
print(list(find_value))
# 5. 查詢嵌入/嵌套文檔
# 查詢字段"tags":{"db":"Mongodb","lan":"Python","modle":"Pymongo"}
# 查詢嵌套字段，只需要查詢嵌套裡的某個值即可
find_value = user_collection.find({
"tags.db": "Mongodb"})
print(list(find_value))
# 6. 查詢字段"tags":{"db":
# {"Mongodb":"NoSql","MySql":"Sql"},"lan":"Python","modle":"Pymongo"}
# find_value = user_collection.find({"tags.db.Mongodb": "NoSql"})
# print(list(find_value))
def find_many():
user_collection = db.user
# 1. 查詢文檔數量
# result_data = user_collection.count_documents({})
# print(result_data)
# 2. 限定返回結果
# result_data_limit = user_collection.find({}).limit(2)
# for result in result_data_limit:
# print(result)
# 3. 對查詢結果進行排序: 字段值1表示正序， -1表示倒序
# user_collection = db.bulk_insert_demo
# result_data_sort = user_collection.find({'age': {'$gt': 22}}).sort([('age', -1)])
# print(list(result_data_sort))
# 4. 對數據進行去重
user_collection = db.bulk_insert_demo
# 對age字段去重
result_data_distinct = user_collection.distinct('age')
print(list(result_data_distinct))
# 對滿足特定條件的age字段去重
# result_data_distinct = user_collection.distinct('age', {'age': {'$gte': 22}})
# print(list(result_data_distinct))
# 5.偏移
# results = collection.find().sort('id', pymongo.ASCENDING).skip(1)
# for result in results:
# print(result)
if __name__ == '__main__':
# 根據條件查詢文檔
# find_by_condition()
# 查詢數據
find_many()

三. 更新數據案例

# -*- encoding: utf-8 -*-
import pymongo
# 創建對象
client = pymongo.MongoClient('mongodb://賬號:密碼@主機:端口號/?authSource=admin')
# 連接DB數據庫
db = client['數據庫名']
def update_one():
# update_one(篩選條件,更新內容)，篩選條件為空，默認更新第一條文檔
# 如果查詢有多條數據，就按照排序先後更新第一條數據
# {"author": "小藍"}, {"$set": {"author": "小黃", "text": "數據挖掘"}}
user_collection = db.user
user_collection.update_one({
"author": "小藍"}, {
"$set": {
"author": "小黃", "text": "數據挖掘"}})
def replace_one():
# replace_one(篩選條件,更新內容)用於將整條數據替換
# 如果文檔的部分數據沒有更新，就去除這部分數據
# topic_data.update_one({"_id": ObjectId(mongo_id)}, {"$set": {'tag_field': 0}})
user_collection = db.user
user_collection.replace_one({
"author": "小綠"},
{
"author": "小綠", "text": "Python_django"})
def update_many():
# update_many(篩選條件,更新內容)用於批量更新文檔, 如果查詢有多條數據，就會對全部數據進行更新處理
# topic_data.update_many({"tag_field": {"$exists": False}}, {"$set": {'tag_field': 0}})
user_collection = db.user
user_collection.update_many({
"author": "小黃"},
{
"$set": {
"text": "Python_web開發"}})
if __name__ == '__main__':
# 更新單條文檔
# update_one()
# 替換一條數據
replace_one()
# 更新多條數據
# update_many()

四. 刪除數據案例

# -*- encoding: utf-8 -*-
import pymongo
# 創建對象
# client = pymongo.MongoClient()
client = pymongo.MongoClient('mongodb://賬號:密碼@主機:端口號/?authSource=admin')
# 連接DB數據庫
db = client['數據庫名']
user_collection = db.user
def delete_one():
# 刪除單條文檔
# delete_one(篩選條件)，篩選條件為空，默認刪除第一條文檔
user_collection.delete_one({
"_id": 100})
def delete_many():
# delete_many(篩選條件)用於刪除多條數據
user_collection.delete_many({
"author": "小黃"})
if __name__ == '__main__':
# 刪除單條文檔
delete_one()
# 刪除多條數據
# delete_many()

五. 聚合查詢案例

import pymongo
handler = pymongo.MongoClient().monog_db.example_user
rows = handler.aggregate([
{
'$lookup': {

'from': 'example_post',
'localField': 'id',
'foreignField': 'user_id',
'as': 'weibo_info'
}
},
{
'$unwind': '$weibo_info'},
{
'$project': {

'name': 1,
'work': 1,
'content': '$weibo_info.content',
'post_time': '$weibo_info.post_time'}}
])
for row in rows:
print(row)

️ 如果覺得有用，感謝一鍵三連哦！！！️