初步探索 MongoDB
使用 mongo image 以及使用 mongo-express GUI tool.
services: db: image: mongo:8.0.12 restart: always container_name: mongodb ports: - "27017:27017" # command: [--auth] environment: MONGO_INITDB_ROOT_USERNAME: root MONGO_INITDB_ROOT_PASSWORD: password volumes: - mongo_data:/data/db mongo-express: image: mongo-express restart: always ports: - 8081:8081 environment: ME_CONFIG_MONGODB_ADMINUSERNAME: root ME_CONFIG_MONGODB_ADMINPASSWORD: password ME_CONFIG_MONGODB_URL: mongodb://root:password@db:27017/ ME_CONFIG_BASICAUTH: false volumes: mongo_data:進入容器
docker exec -it mongodb bash連接
mongosh admin -u root -p password顯示 databases
show dbs顯示 collections (類似 RMDBS 中的 table)
show collections顯示 databases version
db.version()建立新的不存在 database
use test官方文件可參考 mongodb-shell
- Insert a Single Document
use test db.movies.insertOne( { title: "The Favourite", genres: [ "Drama", "History" ], runtime: 121, rated: "R", year: 2018, directors: [ "Yorgos Lanthimos" ], cast: [ "Olivia Colman", "Emma Stone", "Rachel Weisz" ], type: "movie" } ) insertOne()
returns a document that includes the newly inserted document's _id field value. 取回資料
db.movies.find( { title: "The Favourite" } )- Insert Multiple Documents
use test db.movies.insertMany([ { title: "Jurassic World: Fallen Kingdom", genres: [ "Action", "Sci-Fi" ], runtime: 130, rated: "PG-13", year: 2018, directors: [ "J. A. Bayona" ], cast: [ "Chris Pratt", "Bryce Dallas Howard", "Rafe Spall" ], type: "movie" }, { title: "Tag", genres: [ "Comedy", "Action" ], runtime: 105, rated: "R", year: 2018, directors: [ "Jeff Tomsic" ], cast: [ "Annabelle Wallis", "Jeremy Renner", "Jon Hamm" ], type: "movie" } ])insertMany()
returns a document that includes the newly inserted documents' _id field values. 取回全部的資料
db.movies.find( {} )說明語法
use test db.movies.find()類似 RMDBS 中的
SELECT * FROM moviesSpecify Equality Condition
use test db.movies.find( { title: "The Favourite" } )類似 RMDBS 中的
SELECT * FROM movies WHERE title = "The Favourite"Specify Conditions Using Query Operators
use test db.movies.find( { genres: { $in: [ "Drama", "PG-13" ] } } )類似 RMDBS 中的
SELECT * FROM movies WHERE genres in ("Drama", "PG-13")Specify Logical Operators(AND)
use test db.movies.find( { rated: "R", runtime: { $gte: 120 } } )rated 必須是 R, 然後 runtime 必須大於等於 120
Specify Logical Operators(OR)
use test db.movies.find( { year: 2018, $or: [ { runtime: { $gte: 100 } }, { title: "Tag" } ] } )year 必須是 2018, 然後 runtime 必須大於等於 100 或 title 等於 Tag.
- Update a Single Document
use test db.movies.updateOne( { title: "Jurassic World: Fallen Kingdom" }, { $set: {runtime: 77} }) db.movies.find( { title: "Jurassic World: Fallen Kingdom" } ) db.movies.updateOne( { title: "Jurassic World: Fallen Kingdom" },{ $set: {runtime: "77"}, $currentDate: { lastUpdated: true } })如果沒有的欄位, 會自動建立.
有關 $currentDate 的說明可參考 currentDate
- Update Multiple Documents
use test db.movies.updateMany( { runtime: { $lt: 100 } }, { $set: { runtime: 5, nights: 1 } } ) db.movies.find( { nights: 1 } )- Replace a Document
db.movies.replaceOne( { _id: ObjectId("64741819dcd1cd7e37d54731") }, { test123: 893421, limit: 5000, products: [ "Investment", "Brokerage" ] } ) db.movies.findOne( { _id: ObjectId("64741819dcd1cd7e37d54731") } )- Delete All Documents
use test db.movies.deleteMany({})- Delete All Documents that Match a Condition
use test db.movies.deleteMany( { title: "Tag" } )- Delete Only One Document that Matches a Condition
use test db.movies.deleteOne( { _id: ObjectId("64741819dcd1cd7e37d54731") } ) db.movies.findOne( { _id: ObjectId("64741819dcd1cd7e37d54731") } )請直接到這邊下載 MongoDB Compass (GUI)
設定 Authentication Method
如果順利登入的話, 會看到類似的介面
另一種選擇 mongo-express GUI tool.
直接進入 http://0.0.0.0:8081/
可以自己選擇喜歡那款 GUI TOOL
透過 PyMongo documentation 把玩 MongoDB
安裝 pymongo
pip3 install pymongo==4.14.0同步範例 code demo.py
from pymongo import MongoClient from pymongo.errors import ConnectionFailure, ServerSelectionTimeoutError # 建立連接並處理錯誤 try: client = MongoClient('mongodb://root:password@localhost:27017/', serverSelectionTimeoutMS=5000) # 測試連接 client.admin.command('ping') print("成功連接到 MongoDB") except (ConnectionFailure, ServerSelectionTimeoutError) as e: print(f"無法連接到 MongoDB: {e}") exit(1) db = client['test'] movies = db['movies'] def insert_demo(): # insert new_profile = {'user_id': 213, 'name': 'Alan'} result = movies.insert_one(new_profile) print(f"插入文檔 ID: {result.inserted_id}") # read print("\n所有文檔:") for movie in movies.find(): print(movie) def update_demo(): result = movies.update_one({'user_id': 213}, {'$set': {'user_id': 30}}) print(f"更新了 {result.modified_count} 個文檔") # read print("\n更新後的文檔:") for movie in movies.find(): print(movie) def cleanup(): # 清理測試資料 result = movies.delete_many({'user_id': {'$in': [213, 30]}}) print(f"\n清理了 {result.deleted_count} 個文檔") if __name__ == "__main__": try: insert_demo() # update_demo() # cleanup() # 取消註解以清理測試資料 finally: client.close() print("\n已關閉連接")非同步範例 code demo_async.py
import asyncio from pymongo import AsyncMongoClient from pymongo.errors import ConnectionFailure, ServerSelectionTimeoutError # 全域變數 client = None db = None movies = None async def connect(): """建立非同步連接並處理錯誤""" global client, db, movies try: client = AsyncMongoClient('mongodb://root:password@localhost:27017/', serverSelectionTimeoutMS=5000) # 測試連接 await client.admin.command('ping') print("成功連接到 MongoDB") db = client['test'] movies = db['movies'] return True except (ConnectionFailure, ServerSelectionTimeoutError) as e: print(f"無法連接到 MongoDB: {e}") return False async def insert_demo(): """非同步插入示範""" # insert new_profile = {'user_id': 213, 'name': 'Alan'} result = await movies.insert_one(new_profile) print(f"插入文檔 ID: {result.inserted_id}") # read print("\n所有文檔:") async for movie in movies.find(): print(movie) async def update_demo(): """非同步更新示範""" result = await movies.update_one({'user_id': 213}, {'$set': {'user_id': 30}}) print(f"更新了 {result.modified_count} 個文檔") # read print("\n更新後的文檔:") async for movie in movies.find(): print(movie) async def cleanup(): """清理測試資料""" result = await movies.delete_many({'user_id': {'$in': [213, 30]}}) print(f"\n清理了 {result.deleted_count} 個文檔") async def main(): """主程式""" # 建立連接 if not await connect(): return try: await insert_demo() # await update_demo() # await cleanup() # 取消註解以清理測試資料 finally: if client: await client.close() print("\n已關閉連接") if __name__ == "__main__": # 執行非同步主程式 asyncio.run(main())