Python?操作?MongoDB數(shù)據(jù)庫的方法(非?ODM)
更新時間：2023年03月22日 09:55:55 作者：Amo?Xiang
這篇文章主要介紹了Python?操作?MongoDB?----非?ODM的方法，本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
一、連接器的安裝和配置
pymongo： MongoDB 官方提供的 Python 工具包。官方文檔： https://pymongo.readthedocs.io/en/stable/ pip安裝，命令如下：
pip install pymongo
來看看小編挑選的這本書對你是否有用：
管理 MongoDB 的主要步驟如下：
- 連接到 MongoDB 數(shù)據(jù)庫系統(tǒng)
- 管理 MongoDB 數(shù)據(jù)庫
- 管理 MongoDB 中的集合
- 管理 MongoDB 中的文檔
第一步，連接 MongoDB：
# 方式一: 使用默認的配置
client = MongoClient()
# 方式二: 指定主機地址和端口號
client = MongoClient('localhost', 27017)
# 方式三: 使用URI連接參數(shù)
client = MongoClient('mongodb://localhost:27017/')第二步，管理數(shù)據(jù)庫：
#通過MongoClient對象來管理多個數(shù)據(jù)庫獲取數(shù)據(jù)庫(邏輯庫)對象 db = client.DATABASE_NAME db = client["DATABASE_NAME"] db = client.get_database(name=None, *args) # 查看所有的數(shù)據(jù)庫(邏輯庫) client.list_databases() # 刪除數(shù)據(jù)庫(邏輯庫) client.drop_database(name_or_database, *args)
第三步，管理集合
# 通過數(shù)據(jù)庫對象db來管理集合 # 獲取集合對象 db = client.DATABASE_NAME db.COLLECTION_NAME client.DATABASE_NAME.COLLECTION_NAME db.get_collection(name, *args) # 查看當前數(shù)據(jù)庫下的集合列表 db.list_collection_names() # 刪除集合 db.drop_collection(name_or_collection, *args)
基礎操作示例：
# -*- coding: utf-8 -*-
# @Time : 2023-03-17 1:47
# @Author : AmoXiang
# @File : 1.數(shù)據(jù)庫連接.py
# @Software: PyCharm
# @Blog : https://blog.csdn.net/xw1680
from pymongo import MongoClient
# 使用默認配置連接到數(shù)據(jù)庫
# client = MongoClient()
# print(client)
# client.close()
# 指定主機地址和端口號連接到數(shù)據(jù)庫
# client = MongoClient('localhost', 27017)
# 使用URI連接參數(shù)連接到數(shù)據(jù)庫
client = MongoClient('mongodb://localhost:27017/')
print(client)
# client.close()
#
# # 訪問數(shù)據(jù)庫
# db = client.test
# db = client["test"]
# print(db)
# db = client.get_database('test')
# client.close()
# print(db)
#
# 查看有哪些數(shù)據(jù)庫
db_list = client.list_databases()
# # db_list = client.list_database_names()
for item in db_list:
print(item)
#
# 查看數(shù)據(jù)庫下有哪些集合
db_test = client["test"]
for item in db_test.list_collection_names():
print(item)
#
# # 集合對象操作
# data = db_test.students.find_one()
# data = client.test.students.find_one()
data = client.test.get_collection('students').find_one()
print(data)
二、新增文檔
說明：pymongo 在插入數(shù)據(jù)時可以將 python 的對象轉換成 BSON
insert_one()：插入一個文檔
# 調(diào)用方法 result = db.COLLECTION_NAME.insert_one(doc) # 返回插入的文檔ID result.inserted _id
insert_many()：批量新增文檔。調(diào)用方法：
doc_list = [doc1,doc2] result = db.COLLECTION_NAME.insert_many(doc_list)
示例：
三、查詢文檔
pymongo 可以將查詢的結果轉換成 python 中的對象
常用方法：
find_one(): 按條件查詢一個文檔 find(): 按條件查詢多個文檔 count_documents(): 統(tǒng)計滿足條件的文檔總數(shù) aggregate(): 聚合統(tǒng)計 .sort(): 排序 .skip().limit(): 分頁
示例代碼：
# -*- coding: utf-8 -*-
# @Time : 2023-03-18 15:03
# @Author : AmoXiang
# @File : 5.查詢文檔.py
# @Software: PyCharm
# @Blog : https://blog.csdn.net/xw1680
from bson.objectid import ObjectId
from pymongo import MongoClient, ASCENDING, DESCENDING
class LearnMongoDBSearch(object):
""" MongoDB查詢練習 """
def __init__(self):
self.client = MongoClient()
def search_one(self):
""" 查詢一個文檔 """
temp_obj = self.client.test.newdb.find_one()
print(temp_obj)
print('喜歡分數(shù)：', temp_obj['likes'])
# print('注冊時間：', temp_obj['reg_date'].date())
print('姓名：', temp_obj['uname'])
def search_user_by_pk(self, pk):
obj_id = ObjectId(pk)
# user_obj = self.client.test.newdb.find_one({'_id': obj_id})
# 面向對象的方法，有代碼提示
db = self.client.get_database('test')
users = db.get_collection('newdb')
user_obj = users.find_one({'_id': obj_id})
print(user_obj)
def search_many(self):
""" 查詢多個文檔 """
db = self.client.get_database('test')
students = db.get_collection('students')
# stu_list = students.find()
# for item in stu_list:
# print(item)
# 查詢年齡大于12歲的學生
stu_list = students.find({'age': {'$gt': 12}}, {'stu_name': 1, 'class_name': 1, 'age': 1})
for item in stu_list:
# 注意: mongo中存儲的整數(shù)轉換成了浮點數(shù)
print(item)
def paginate(self, page=1, page_size=10):
"""
分頁處理
:param page: 當前的頁
:param page_size: 每一頁數(shù)據(jù)大小
:return:
"""
db = self.client.get_database('test')
students = db.get_collection('students')
offset = (page - 1) * page_size
stu_list = students.find().skip(offset).limit(page_size)
return stu_list
def sort_data(self):
""" 排序 """
db = self.client.get_database('test')
grades = db.get_collection('grades')
# // 將學生的語文成績從高到低排序
# db.grades.find({"grade.course_name": "語文"}).sort({"grade.score": -1});
#
# //將學生的語文成績按照年齡和成績排序
# db.grades.find({"grade.course_name": "語文"}).sort({"age": -1, "grade.score": -1});
# db.grades.find({"grade.course_name": "語文"}).sort({"grade.score": -1, "age": -1});
# 按某一列排序
# data_list = grades.find({"grade.course_name": "語文"}).sort("grade.score", DESCENDING);
# for item in data_list:
# print(item)
# 按多列排序
data_list = grades.find({"grade.course_name": "語文"}).sort([('age', DESCENDING),("grade.score", DESCENDING),])
for item in data_list:
print(item)
def counter_students(self):
""" 統(tǒng)計newdb中文檔總數(shù) """
db = self.client.get_database('test')
newdb = db.get_collection('newdb')
# result = newdb.count_documents({})
result = newdb.count_documents({"uname": {"$eq": "張三"}})
print(result)
def test_aggregate(self):
"""
聚合統(tǒng)計：及格的學生成績
"""
db = self.client.get_database('test')
grades = db.get_collection('grades')
result = grades.aggregate([
# //where
{
'$match': {"grade.score": {'$gte': 60}}
},
# //group by
{
'$group': {
'_id': "$stu_no",
'total': {'$sum': 1}
}
},
# // having
{
'$match': {
'total': {'$eq': 3}
}
}
])
for item in result:
print(item)
if __name__ == '__main__':
obj = LearnMongoDBSearch()
# obj.search_one()
# obj.search_user_by_pk('6411ee77b6170000b4003f95')
# obj.search_many()
# stu_list = obj.paginate(page=3)
# for item in stu_list:
# print(item)
# obj.sort_data()
# obj.counter_students()
obj.test_aggregate()
四、更新文檔
回顧，更新數(shù)據(jù)表達式，如下表所示：
修改一個文檔，調(diào)用方法：
update_one(filter, update, *args)
替換一個文檔，調(diào)用方法：
replace_one(filter, replacement, *args)
批量修改文檔，調(diào)用方法：
replace_one(filter, replacement, *args)
快捷方法：
find_one_and_update(filter, update, *args) # 修改一個文檔 find_one_and_replace(filter, replacement, *args) # 替換一個文檔 # 注意返回值的不同
返回結果：
acknowledged：結果是否已經(jīng)被確認
modified_count：修改的文檔數(shù)
matched_count：滿足條件的文檔數(shù)
raw_result：原始數(shù)據(jù)
upserted_id：更新的ID (upsert=True)
示例代碼：
# -*- coding: utf-8 -*-
# @Time : 2023-03-18 14:56
# @Author : AmoXiang
# @File : 4.修改文檔.py
# @Software: PyCharm
# @Blog : https://blog.csdn.net/xw1680
from pymongo import MongoClient
class LearnMongoDBUpdate(object):
""" MongoDB更新練習 """
def __init__(self):
self.client = MongoClient()
def test_update_one(self):
""" 更新一個文檔 """
db = self.client.get_database('test')
newdb = db.get_collection('newdb')
result = newdb.update_one({}, {'$set': {'likes': 70}})
print(result.modified_count)
def test_replace_one(self):
""" 替換一個文檔 """
db = self.client.get_database('test')
newdb = db.get_collection('newdb')
result = newdb.replace_one({}, {'uname': '張三'})
print(result.modified_count)
def test_update_many(self):
""" 批量更新文檔 """
db = self.client.get_database('test')
newdb = db.get_collection('newdb')
result = newdb.update_many({}, {'$set': {'likes': 90}})
print('修改的數(shù)量：', result.modified_count)
print('滿足條件的數(shù)量：', result.matched_count)
def test_update_shortcut(self):
""" 更新文檔的快捷方法 """
db = self.client.get_database('test')
newdb = db.get_collection('newdb')
# 此處返回一個文檔對象
result = newdb.find_one_and_update({}, {'$set': {'sex': '未知'}})
print(result['_id'])
if __name__ == '__main__':
obj = LearnMongoDBUpdate()
# obj.test_update_one()
# obj.test_replace_one()
# obj.test_update_many()
obj.test_update_shortcut()
五、刪除文檔
刪除一個文檔，調(diào)用方法如下：
result = db.COLLECTION_NAME.delete_one(filter, *args) # 返回已經(jīng)刪除的記錄數(shù) result.deleted_count # 刪除時返回文檔對象 find_one_and_delete(filter, *args)
批量刪除文檔，調(diào)用方法：
result = db.COLLECTION_NAME.delete_many(filter, *args) # 返回已經(jīng)刪除的記錄數(shù) result. deleted_count
示例代碼：
# -*- coding: utf-8 -*-
# @Time : 2023-03-18 14:34
# @Author : AmoXiang
# @File : 3.刪除文檔.py
# @Software: PyCharm
# @Blog : https://blog.csdn.net/xw1680
from pymongo import MongoClient
class LearnMongoDBDelete(object):
""" MongoDB刪除練習 """
def __init__(self):
self.client = MongoClient()
def test_delete_one(self):
""" 刪除一個文檔 """
db = self.client.get_database('test')
newdb = db.get_collection('newdb')
result = newdb.delete_one({})
print(result.deleted_count)
def test_delete_many(self):
""" 批量刪除文檔 """
db = self.client.get_database('test')
users = db.get_collection('newdb')
# 刪除所有的數(shù)據(jù)
result = users.delete_many({"likes": {"$lte": 90}})
print(result.deleted_count)
def test_delete_shortcut(self):
""" 刪除文檔的快捷方法 """
db = self.client.get_database('test')
newdb = db.get_collection('newdb')
result = newdb.find_one_and_delete({})
print(result['title'])
if __name__ == '__main__':
obj = LearnMongoDBDelete()
# obj.test_delete_one()
# obj.test_delete_shortcut()
obj.test_delete_many()
至此今天的學習就到此結束了，筆者在這里聲明，筆者寫文章只是為了學習交流，以及讓更多學習數(shù)據(jù)庫的讀者少走一些彎路，節(jié)省時間，并不用做其他用途，如有侵權，聯(lián)系博主刪除即可。感謝您閱讀本篇博文，希望本文能成為您編程路上的領航者。祝您閱讀愉快！
到此這篇關于Python 操作 MongoDB ----非 ODM的文章就介紹到這了,更多相關Python 操作 MongoDB內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
在本篇文章里小編給大家整理的是關于app后端開發(fā)學PHP還是Python的先關問題內(nèi)容，需要的朋友們可以參考下。2020-06-06
這篇文章主要介紹了Selenium 滾動頁面至元素可見的方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-03-03
這篇文章主要介紹了Python切片索引用法,結合實例形式詳細分析了Python切片索引的常見使用方法與操作注意事項,需要的朋友可以參考下2018-05-05
django 實現(xiàn)celery動態(tài)設置周期任務執(zhí)行時間
今天小編就為大家分享一篇django 實現(xiàn)celery動態(tài)設置周期任務執(zhí)行時間，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-11-11
TensorFlow實現(xiàn)checkpoint文件轉換為pb文件
今天小編就為大家分享一篇TensorFlow實現(xiàn)checkpoint文件轉換為pb文件，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-02-02
這篇文章主要為大家詳細介紹了python算法的基礎教程，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-04-04
這篇文章主要為大家詳細介紹了Python實現(xiàn)簡單的2048小游戲，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-03-03
python實現(xiàn)while循環(huán)打印星星的四種形狀
今天小編就為大家分享一篇python實現(xiàn)while循環(huán)打印星星的四種形狀，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-11-11
這篇文章主要介紹了Python語言生成水仙花數(shù)代碼示例，具有一定借鑒價值，需要的朋友可以參考下。2017-12-12
VPN（虛擬私人網(wǎng)絡）是一種有效的解決方案,可以幫助我們在網(wǎng)絡上匿名瀏覽,保護數(shù)據(jù)傳輸?shù)陌踩?雖然市面上有許多商業(yè)VPN服務,但你也可以通過Python自己搭建一個簡單的VPN,本文將介紹如何用Python建立自己的VPN,需要的朋友可以參考下2024-06-06