数据库mongodb

检测mongodb是否启动

from pymongo import MongoClient
try:
    with MongoClient("mongodb://127.0.0.1:27017", serverSelectionTimeoutMS=1000) as c:
        c.admin.command("ping")   # 真正发一次握手
    print("MongoDB 已启动")
except Exception:
    print("MongoDB 未启动")
    
######导入

from dotenv import load_dotenv
from urllib.parse import quote_plus
from pymongo import MongoClient
load_dotenv()
mongourl = "mongodb://{}:{}@{}:{}".format(
	quote_plus(os.getenv("mongodbuser")),
	quote_plus(os.getenv("mongodbpassword")),
	os.getenv("mongodbip"),
	os.getenv("mongodbport"),
)
mongodb_client = MongoClient(mongourl)

添加数据

from pymongo import MongoClient
URI = "mongodb://localhost:27017"   # 唯一需要的外部变量
# 1. 新增一条文档
with MongoClient(URI) as client:
    client["test"]["demo"].insert_one({"name": "张三", "age": 25})

# 2. 批量新增多条文档
with MongoClient(URI) as client:
    client["test"]["demo"].insert_many([
        {"name": "李四", "age": 30},
        {"name": "王五", "age": 28}
    ])

删除数据

from pymongo import MongoClient
URI = "mongodb://localhost:27017"   # 唯一需要的外部变量
# 3. 删除第一条 age=30 的文档
with MongoClient(URI) as client:
    client["test"]["demo"].delete_one({"age": 30})

# 4. 删除所有 age>=28 的文档
with MongoClient(URI) as client:
    client["test"]["demo"].delete_many({"age": {"$gte": 28}})

更新数据

from pymongo import MongoClient
URI = "mongodb://localhost:27017"   # 唯一需要的外部变量
# 5. 只改第一条 name=张三 的文档，把 age 改成 26
with MongoClient(URI) as client:
    client["test"]["demo"].update_one(
        {"name": "张三"},          # 匹配条件
        {"$set": {"age": 26}}      # 只改字段
    )

# 6. 批量改所有 age>=28 的文档，统一加 1 岁
with MongoClient(URI) as client:
    client["test"]["demo"].update_many(
        {"age": {"$gte": 28}},
        {"$inc": {"age": 1}}
    )
# 6. 没有就插入,有就更新
with MongoClient(URI) as client:
    client["test"]["demo"].update_one(
    {"name": "张三"},           # 查询条件
    {"$set": {"age": 28, "city": "北京"}},  # 要写入/更新的字段
    upsert=True                 # 关键：没有就插入
)

# 6. 没有就插入,有就跳过 limit可选，只数到指定条数就停（性能优化）
doc = {"name": "张三", "age": 28, "city": "北京"}
with MongoClient(URI) as client:
    if client["test"]["demo"].count_documents({"name": doc["name"]}, limit=1) == 0:
    col.insert_one(doc)   # 不存在才插入

查询数据

from pymongo import MongoClient
URI = "mongodb://localhost:27017"   # 唯一需要的外部变量

# 获取文档中status为失败的个数
with MongoClient(URI) as client:
    doc = client["test"]["demo"].count_documents({"status": "失败"})
    
with MongoClient(URI) as client:
    doc = client["test"]["demo"].find_one()
    print("任意文档：", doc)

# 8. 按条件查一条 name=张三 的文档
with MongoClient(URI) as client:
    doc = client["test"]["demo"].find_one({"name": "张三"})
    print("张三：", doc)

# 9. 按条件查多条 age>=26 的文档，并限制 10 条
with MongoClient(URI) as client:
    cursor = client["test"]["demo"].find({"age": {"$gte": 26}}).limit(10)
    for d in cursor:
        print("age>=26：", d)
        
# 10. 正则查 name 以“张”开头的文档
with MongoClient(URI) as client:
    cursor = client["test"]["demo"].find({"name": {"$regex": "^张"}})
    for d in cursor:
        print("姓张：", d)

# 11. 忽略大小写查 email 包含 "GMAIL" 的文档
with MongoClient(URI) as client:
    cursor = client["test"]["demo"].find(
        {"email": {"$regex": "gmail", "$options": "i"}}
    )
    for d in cursor:
        print("gmail 邮箱：", d)

限制查询方法

限制返回数据:5条

from pymongo import MongoClient

with MongoClient("mongodb://127.0.0.1:27017") as client:
    cursor = client["test"]["demo"].find({"age": {"$gte": 18}}).limit(5)
    for doc in cursor:
        print(doc)

排序（sort）+ 限制: 返回前三

from pymongo import ASCENDING, DESCENDING

with MongoClient("mongodb://127.0.0.1:27017") as client:
    cursor = (
        client["test"]["demo"]
        .find()
        .sort("age", DESCENDING)  # 按 age 降序
        .limit(3)
    )
    for doc in cursor:
        print(doc)

操作符