Python 操作 MongoDB數據庫的方法(非 ODM)
一、連接器的安裝和配置
pymongo: MongoDB 官方提供的 Python 工具包。官方文檔: https://pymongo.readthedocs.io/en/stable/ pip安裝,命令如下:
pip install pymongo
管理 MongoDB 的主要步驟如下:
- 連接到 MongoDB 數據庫系統
- 管理 MongoDB 數據庫
- 管理 MongoDB 中的集合
- 管理 MongoDB 中的文檔
第一步,連接 MongoDB:
# 方式一: 使用默認的配置 client = MongoClient() # 方式二: 指定主機地址和端口號 client = MongoClient('localhost', 27017) # 方式三: 使用URI連接參數 client = MongoClient('mongodb://localhost:27017/')
第二步,管理數據庫:
#通過MongoClient對象來管理多個數據庫獲取數據庫(邏輯庫)對象 db = client.DATABASE_NAME db = client["DATABASE_NAME"] db = client.get_database(name=None, *args) # 查看所有的數據庫(邏輯庫) client.list_databases() # 刪除數據庫(邏輯庫) client.drop_database(name_or_database, *args)
第三步,管理集合
# 通過數據庫對象db來管理集合 # 獲取集合對象 db = client.DATABASE_NAME db.COLLECTION_NAME client.DATABASE_NAME.COLLECTION_NAME db.get_collection(name, *args) # 查看當前數據庫下的集合列表 db.list_collection_names() # 刪除集合 db.drop_collection(name_or_collection, *args)
基礎操作示例:
# -*- coding: utf-8 -*- # @Time : 2023-03-17 1:47 # @Author : AmoXiang # @File : 1.數據庫連接.py # @Software: PyCharm # @Blog : https://blog.csdn.net/xw1680 from pymongo import MongoClient # 使用默認配置連接到數據庫 # client = MongoClient() # print(client) # client.close() # 指定主機地址和端口號連接到數據庫 # client = MongoClient('localhost', 27017) # 使用URI連接參數連接到數據庫 client = MongoClient('mongodb://localhost:27017/') print(client) # client.close() # # # 訪問數據庫 # db = client.test # db = client["test"] # print(db) # db = client.get_database('test') # client.close() # print(db) # # 查看有哪些數據庫 db_list = client.list_databases() # # db_list = client.list_database_names() for item in db_list: print(item) # # 查看數據庫下有哪些集合 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 在插入數據時可以將 python 的對象轉換成 BSON
insert_one():插入一個文檔
# 調用方法 result = db.COLLECTION_NAME.insert_one(doc) # 返回插入的文檔ID result.inserted _id
insert_many():批量新增文檔。調用方法:
doc_list = [doc1,doc2] result = db.COLLECTION_NAME.insert_many(doc_list)
示例:
三、查詢文檔
pymongo 可以將查詢的結果轉換成 python 中的對象
常用方法:
find_one(): 按條件查詢一個文檔 find(): 按條件查詢多個文檔 count_documents(): 統計滿足條件的文檔總數 aggregate(): 聚合統計 .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('喜歡分數:', 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中存儲的整數轉換成瞭浮點數 print(item) def paginate(self, page=1, page_size=10): """ 分頁處理 :param page: 當前的頁 :param page_size: 每一頁數據大小 :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): """ 統計newdb中文檔總數 """ 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): """ 聚合統計:及格的學生成績 """ 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()
四、更新文檔
回顧,更新數據表達式,如下表所示:
修改一個文檔,調用方法:
update_one(filter, update, *args)
替換一個文檔,調用方法:
replace_one(filter, replacement, *args)
批量修改文檔,調用方法:
replace_one(filter, replacement, *args)
快捷方法:
find_one_and_update(filter, update, *args) # 修改一個文檔 find_one_and_replace(filter, replacement, *args) # 替換一個文檔 # 註意返回值的不同
返回結果:
acknowledged:結果是否已經被確認
modified_count:修改的文檔數
matched_count:滿足條件的文檔數
raw_result:原始數據
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('修改的數量:', result.modified_count) print('滿足條件的數量:', 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()
五、刪除文檔
刪除一個文檔,調用方法如下:
result = db.COLLECTION_NAME.delete_one(filter, *args) # 返回已經刪除的記錄數 result.deleted_count # 刪除時返回文檔對象 find_one_and_delete(filter, *args)
批量刪除文檔,調用方法:
result = db.COLLECTION_NAME.delete_many(filter, *args) # 返回已經刪除的記錄數 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') # 刪除所有的數據 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()
至此今天的學習就到此結束瞭,筆者在這裡聲明,筆者寫文章隻是為瞭學習交流,以及讓更多學習數據庫的讀者少走一些彎路,節省時間,並不用做其他用途,如有侵權,聯系博主刪除即可。感謝您閱讀本篇博文,希望本文能成為您編程路上的領航者。祝您閱讀愉快!
到此這篇關於Python 操作 MongoDB —-非 ODM的文章就介紹到這瞭,更多相關Python 操作 MongoDB內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- 深入瞭解Python如何操作MongoDB
- Python 操作 MongoDB 講解詳細
- Python與數據庫的交互問題小結
- Django項目連接MongoDB的三種方法
- Java MongoDB數據庫連接方法梳理