Python操作MongoDB快速入门教程

Docker安装MongoDB

拉取镜像：

docker pull mongo:6.0.2

创建容器：

docker run --name mongo -d -p 27017:27017 mongo:6.0.2

设置用户名和密码：

# 创建mongo容器后，进入容器
docker exec -it mongo bash

# 进入mongo shell
mongosh

# 进入admin数据库
use admin

# 创建用户名和密码
db.createUser({user:'zhangdapeng',pwd:'zhangdapeng520',roles:[{role:'root',db:'admin'}]})

校验用户名和密码：

# 认证登录db.auth('用户名','密码')，打印1则代表认证通过
db.auth('zhangdapeng','zhangdapeng520')

安装依赖

pip install ./dist/zdppy_mongo-0.1.0.tar.gz

连接MongoDB

import time

from mongo.pymongo import MongoClient

# 会开进程，主进程需要等待
client = MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
print(client)

# 等待1秒钟
time.sleep(1)
print("建立连接成功：", client)

创建和删除数据库

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)


# 删除数据库
client.drop_database("user_manager")

创建集合

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 删除数据库
client.drop_database("user_manager")

添加数据

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
r = db_user.insert_one({"name": "张三", "age": 23})
print(r)

# 删除数据库
client.drop_database("user_manager")

查询数据

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
r = db_user.insert_one({"name": "张三", "age": 23})
print(r)

# 查询数据
r = db_user.find_one({"name": "张三"})
print(r)

# 删除数据库
client.drop_database("user_manager")

批量添加数据

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
r = db_user.insert_many([{"name": f"张三{i}", "age": i % 20} for i in range(100)])
print(r)

# 删除数据库
client.drop_database("user_manager")

查询所有数据

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
r = db_user.insert_many([{"name": f"张三{i}", "age": i % 20} for i in range(100)])
print(r)

# 查询数据
r = db_user.find()
print(list(r))

# 删除数据库
client.drop_database("user_manager")

查询数据总数

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据


# 添加数据
r = db_user.insert_many([{"name": f"张三{i}", "age": i % 20} for i in range(100)])
print(r)

# 查询数据
r = db_user.find()
print(list(r))

# 查询数据
print(db_user.count_documents({}))

# 删除数据库
client.drop_database("user_manager")

批量执行操作

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
r = db_user.bulk_write([
    mongo.DeleteMany({}),  # 删除所有数据
    mongo.InsertOne({"name": "张三", "age": 23}),  # 新增一条数据
    mongo.InsertOne({"name": "李四", "age": 23}),  # 新增一条数据
    mongo.UpdateOne({"name": "李四"}, {"$set": {"age": 24}}),  # 更新数据
    mongo.UpdateOne({"name": "李四"}, {"$inc": {"age": 1}}),  # 更新数据
    mongo.ReplaceOne({"name": "李四"}, {"name": "李四", "age": 34}),  # 替换数据
])
print(r)

# 查询数据
r = db_user.find()
print(list(r))

# 删除数据库
client.drop_database("user_manager")

获取批量执行错误

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
data = [
    mongo.DeleteMany({}),  # 删除所有数据
    mongo.InsertOne({"name": "张三", "age": 23}),  # 新增一条数据
    mongo.InsertOne({"name": "李四", "age": 23}),  # 新增一条数据
    mongo.UpdateOne({"name": "李四"}, {"$set": {"age": 24}}),  # 更新数据
    mongo.UpdateOne({"name": "李四"}, {"$inc": {"age": 1}}),  # 更新数据
    mongo.ReplaceOne({"name": "李四"}, {"name": "李四", "age": 34}),  # 替换数据
]
try:
    r = db_user.bulk_write(data)
    print(r)
except mongo.BulkWriteError as e:
    print("批量执行操作失败：", e)

# 查询数据
r = db_user.find()
print(list(r))

# 删除数据库
client.drop_database("user_manager")

批量更新

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
data = [
    {"name": "张三", "age": 23},
    {"name": "李四", "age": 33},
    {"name": "王五", "age": 25},
]
r = db_user.insert_many(data)
print(r)

# 更新
# 年龄小于30岁的自增1
db_user.update_many(
    {"age": {"$lt": 30}},
    {"$set": {"age": 33}}
)

# 查询数据
r = db_user.find()
print(list(r))

# 删除数据库
client.drop_database("user_manager")

日期类型

import time
import datetime
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
r = db_user.insert_one({"last_modified": datetime.datetime.now()})
print(r)

# 查询数据
r = db_user.find()
print(list(r))

# 删除数据库
client.drop_database("user_manager")

分页查询

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
data = [
    {"name": "张三", "age": 23},
    {"name": "李四", "age": 33},
    {"name": "王五", "age": 25},
]
r = db_user.insert_many(data)
print(r)


# 查询数据
r = db_user.find().limit(2).skip(2)
print(list(r))

# 删除数据库
client.drop_database("user_manager")

聚合查询

import time
import mongo

# 建立链接，会开进程，主进程要等待一会儿
client = mongo.MongoClient('mongodb://zhangdapeng:zhangdapeng520@localhost:27017/')
time.sleep(1)
print("建立连接成功：", client)

# 创建数据库
db = client["user_manager"]
print(db)

# 创建集合
db_user = db["user"]
print(db_user)

# 添加数据
data = [
    {"name": "张三", "age": 23},
    {"name": "李四", "age": 33},
    {"name": "王五", "age": 25},
]
r = db_user.insert_many(data)
print(r)

# 查询数据
r = db_user.aggregate([{"$group": {"_id": None, "count": {"$sum": 1}, "avg_age": {"$avg": "$age"}}}])
print(list(r))

# 删除数据库
client.drop_database("user_manager")