企业官网建设流程全解析

在“100天精通Python”系列的学习路线里，第 40 天进入数据库操作阶段，主角是 pymongo 和 MongoDB。MongoDB 是当前使用非常广泛的文档型 NoSQL 数据库，它以 BSON 格式保存文档，天然适合存储 JSON 风格的数据；pymongo 则是 MongoDB 官方提供的 Python 驱动，负责让 Python 程序与 MongoDB 服务端完成连接、读写和底层通信。对刚学完 Python 基础、第一次接触数据库联动开发的读者来说，这一天的内容会把练习难度从“单机脚本”提升到“客户端与数据库配合”，所以连接、写入、查询、更新、删除每一条链路都值得完整过一遍。

这一天的内容会按一个可复现的顺序展开：先说明 MongoDB 与关系型数据库的核心差异，接着安装并启动 MongoDB，再安装 pymongo，然后建立连接，逐步写完增删改查、排序分页、计数和简单聚合，最后给出常见报错的排查路径和生产环境注意事项。学完这一篇，你不仅能写出可运行的 pymongo 脚本，还能在真正部署项目时知道哪些配置必须改、哪些写法不能照搬到生产环境。

1. 先把 MongoDB 和 pymongo 的技术定位讲清楚

1.1 MongoDB 是文档型数据库，和 MySQL 的核心差异

用一句话理解 MongoDB：它是一种不强制表结构、以文档为单位存储数据的数据库。这里的“文档”不是 Word 文档，而是一条 JSON 风格的记录，MongoDB 内部会把这条记录序列化成 BSON 格式存储。

和 MySQL 这类关系型数据库相比，MongoDB 的核心差异体现在数据模型上。MySQL 要求先建库、建表、定义字段类型，再插入数据；MongoDB 则不需要提前定义集合结构，插入第一条文档时才自动创建数据库和集合。MySQL 通过外键和 JOIN 把数据拆分到多张表；MongoDB 更倾向于把一组相关数据放在同一个文档里，减少跨表关联。

并不是说 MongoDB 可以完全替代 MySQL，而是它们适用的场景不同。日志、用户画像、爬虫结果、配置类数据、频繁变动的 JSON 结构，用 MongoDB 写起来很顺手；强事务、强一致性、复杂报表关联，关系型数据库仍然更稳。学习第 40 天时，先接受它是一个“文档数据库”即可，不需要急着下结论说谁更好。

1.2 pymongo 是官方驱动，负责 Python 和 MongoDB 之间的通信

pymongo 不只是“操作 MongoDB 的 Python 工具”，它本质上是 MongoDB 官方维护的 Python 客户端驱动。驱动要处理的底层工作比你想象得多：

• 建立和维护 TCP 连接，管理连接池。
• 把 Python 的 dict 转换为 BSON 格式发送给服务端。
• 把服务端返回的 BSON 结果转换回 Python 的 dict。
• 处理游标、心跳检测、副本集节点发现等底层机制。

所以你在 pymongo 里写的大部分业务代码，都是操作 dict 和查询条件字典，并不需要手动拼接 JSON 字符串。这一点刚上手时可能感觉不到，等后面接触非官方客户端或者自己实现协议时，才能体会到驱动到底替你省了多少事。

1.3 这一天的学习目标和前置要求

学习这一天之前，需要掌握 Python 基础语法，包括 dict、list、循环、函数和异常处理。还需要能在命令行执行python和pip命令。数据库方面，建议已经把 MongoDB 安装好，并且会用mongosh查看数据，再来练习 pymongo，这样验证结果会容易很多。

这一天的目标不是把 MongoDB 的所有特性讲完，而是把程序里最常用的链路全部跑通：连接、插入、查询、更新、删除、排序分页、计数和简单聚合。后面的章节全部围绕这几条链路展开。

2. 环境准备：MongoDB 服务端和 pymongo 都要装好

2.1 MongoDB 服务端的安装、启动与状态检查

pymongo 是客户端，必须连接到一个正在运行的 MongoDB 服务端。不同操作系统安装 MongoDB 的方式差别很大，官方安装文档会更准确，这里重点说明安装完成之后如何启动和检查。

在 Linux systemd 环境下，安装完成后的典型操作是：

sudo systemctl start mongod sudo systemctl enable mongod sudo systemctl status mongod

执行systemctl status mongod时，预期能看到active (running)状态。Windows 下安装后，MongoDB 通常注册为系统服务，可以直接在服务管理器中确认运行状态；macOS 使用 Homebrew 安装时，可以通过brew services list查看。

服务启动后，使用mongosh验证服务端是否可连接：

mongosh --eval "db.version()"

如果返回一个版本号字符串，说明服务端已经正常监听默认端口 27017。

注意：不同操作系统、不同 MongoDB 版本的安装命令并不完全一样。上面命令用于说明安装完成后的启动与检查思路，实际安装前要先确认自己的系统和目标版本。

2.2 确认 Python 版本并安装 pymongo

pymongo 4.x 要求 Python 3.7 以上。先确认当前环境：

python --version pip --version

安装 pymongo：

pip install pymongo

建议在虚拟环境中安装，避免污染全局 Python 环境。如果已经进入虚拟环境，直接用上面的命令即可。

验证安装结果：

python -c "import pymongo; print(pymongo.__version__)"

能输出版本号，说明 pymongo 已经可以被 Python 正常导入。

2.3 环境检查清单

启动和安装完成后，建议按下面的表格逐项检查一次，避免后面写代码时把“环境问题”误判成“代码问题”。

环境这一关过了，后面才能把注意力放在 pymongo 的 API 和查询写法上。

3. 建立连接并理解 Database、Collection 和文档

3.1 MongoClient 连接串的构成

pymongo 的连接入口是MongoClient：

from pymongo import MongoClient client = MongoClient("mongodb://localhost:27017/")

这个连接串由几部分组成：

一个带认证的典型连接串是：

client = MongoClient("mongodb://admin:123456@localhost:27017/admin?authSource=admin")

这里authSource=admin表示认证数据库是 admin，而不是业务数据库。

有一个排查时很重要的特性：MongoClient创建时不会立刻发起显式的业务请求，真正执行数据库操作时才建立连接。所以连接串写错了，往往不是创建MongoClient时报错，而是第一次执行insert_one、find等操作时才抛出异常。

3.2 获取数据库和集合的方式

连接成功后，可以通过字典方式获取数据库和集合：

db = client["school"] students = db["students"]

也可以使用属性方式：

db = client.school students = db.students

两种方式等价。属性方式写起来快，但当集合名与 pymongo 的方法名或属性名冲突时会有问题。比如某个集合恰好叫insert_one，db.insert_one会被解析成方法而不是集合。因此推荐使用client["school"]、db["students"]这种括号写法，尤其当集合名包含特殊字符或与关键字冲突时。

关键认知：在 MongoDB 中，数据库和集合都可以隐式创建。第一次向students集合写入文档时，school数据库和students集合会自动出现，不需要提前执行建库建表语句。

3.3 插入第一条数据验证连接

先跑一个最小插入脚本，完整验证环境：

from pymongo import MongoClient client = MongoClient("mongodb://localhost:27017/") db = client["school"] students = db["students"] doc = { "name": "张三", "age": 20, "class_name": "Python 提高班", } result = students.insert_one(doc) print(result.inserted_id) client.close()

正常输出是一个ObjectId('...')，例如：

ObjectId('6779f1a2b4c3d4e5f6a7b8c9')

这个ObjectId是 MongoDB 自动生成的_id字段值。insert_one返回的是InsertOneResult对象，inserted_id就是新文档的主键。

此时切换到mongosh里可以看到同样的数据：

use school db.students.find().pretty()

输出中会显示_id、name、age、class_name四个字段。到这一步，说明 pymongo 与 MongoDB 的整个通信链路已经打通。

4. 增删改查完整实战

4.1 插入单个文档和批量插入

单个文档插入使用insert_one：

doc = {"name": "李四", "age": 22, "class_name": "Python 基础班"} result = students.insert_one(doc) print(result.inserted_id)

批量插入使用insert_many，参数是一个文档列表：

docs = [ {"name": "王五", "age": 21, "class_name": "Python 基础班"}, {"name": "赵六", "age": 19, "class_name": "Python 提高班"}, {"name": "孙七", "age": 23, "class_name": "数据分析班"}, ] result = students.insert_many(docs) print(result.inserted_ids)

insert_many返回InsertManyResult，inserted_ids是一个ObjectId列表，顺序与传入的文档顺序一致。

插入时要注意：如果没有显式提供_id字段，MongoDB 会自动生成ObjectId。如果显式提供了_id，那么同一个集合内不能重复，否则会抛出DuplicateKeyError。

4.2 查询单个文档和遍历查询结果

查询单个文档使用find_one，返回一个 dict 或None：

one = students.find_one({"name": "张三"}) print(one)

输出示例：

{'_id': ObjectId('6779...'), 'name': '张三', 'age': 20, 'class_name': 'Python 提高班'}

注意_id是ObjectId类型，不是字符串。直接打印时看到ObjectId('...')是正常的。

查询多个文档使用find，返回的是一个 Cursor 对象：

cursor = students.find() for s in cursor: print(s["name"], s["age"])

Cursor 有两个容易踩坑的特点。第一，它是惰性的，遍历时才真正向服务端拉取数据；第二，它只能从头到尾迭代一次，如果需要多次使用，要先转换成列表：

all_students = list(students.find()) print(len(all_students))

4.3 条件查询：比较运算符、逻辑运算符和字段判断

pymongo 的查询条件就是 Python 字典，操作符以$开头。最常用的条件写法如下：

结合集合里的测试数据，实际写法如下：

# 年龄大于等于 20 且小于 30，可以直接把范围写在一个字段条件里 cursor = students.find({"age": {"$gte": 20, "$lt": 30}}) for s in cursor: print(s["name"], s["age"]) # 查询基础班或提高班的学生 cursor = students.find({"class_name": {"$in": ["Python 基础班", "Python 提高班"]}}) for s in cursor: print(s["name"], s["class_name"]) # 查询姓名以“张”开头的人 cursor = students.find({"name": {"$regex": "^张"}}) for s in cursor: print(s["name"])

多条件默认就是 AND 语义，直接写多个字段即可：

cursor = students.find({"class_name": "Python 提高班", "age": {"$gte": 20}})

4.4 更新文档：$set、$inc、$push 与 update_many

更新使用update_one和update_many，它们都接收两个字典：第一个是过滤条件，第二个是更新操作。

result = students.update_one( {"name": "张三"}, {"$set": {"age": 26}} ) print(result.matched_count) # 匹配到的文档数 print(result.modified_count) # 实际修改的文档数

更新操作符最常用的是下面几个：

批量更新示例：

# 给基础班所有学生加 10 分 result = students.update_many( {"class_name": "Python 基础班"}, {"$inc": {"score": 10}} ) print(result.modified_count)

一个必须区分清楚的点：matched_count表示有多少条文档匹配了过滤条件，modified_count表示实际发生了修改的条数。如果把某人的 age 从 26 改成 26，matched_count是 1，modified_count却是 0。排查更新不生效时，这两个数字要分开看。

4.5 删除文档：按条件删除与清空集合

删除使用delete_one和delete_many：

result = students.delete_one({"name": "李四"}) print(result.deleted_count) result = students.delete_many({"class_name": "Python 提高班"}) print(result.deleted_count)

清空整个集合可以使用空过滤条件：

result = students.delete_many({}) print(result.deleted_count)

delete_many({})会删除集合内所有文档，但集合本身还存在。如果想把集合也一起删掉，可以执行：

students.drop()

学习环境随意删没问题，生产环境一定要谨慎。delete_many的过滤条件写空字典时，一次就会清空数据，所以执行前必须确认条件是否正确。

5. 排序、分页、计数与聚合

5.1 sort、skip、limit 组合实现分页

排序使用sort方法。按年龄升序：

cursor = students.find().sort("age", 1) for s in cursor: print(s["name"], s["age"])

sort的第二个参数，1表示升序，-1表示降序，等价写法是：

from pymongo import ASCENDING, DESCENDING cursor = students.find().sort("age", ASCENDING)

多字段排序时，传入一个由元组组成的列表：

cursor = students.find().sort([("age", -1), ("name", 1)])

先按年龄降序，年龄相同的人再按姓名升序。

分页查询使用skip和limit组合：

page = 2 page_size = 3 cursor = students.find().sort("age", -1).skip((page - 1) * page_size).limit(page_size) for s in cursor: print(s["name"], s["age"])

这里skip跳过前面的数据，limit限制返回数量。第 2 页每页 3 条，就是跳过 3 条再取 3 条。

当数据量很大时，skip跳过的行数越多，性能越差。生产环境做深分页时，更推荐基于_id或排序字段的范围查询来取下一页，而不是无限增大skip。

5.2 count_documents 统计文档数量

统计数量使用count_documents：

total = students.count_documents({}) basic_count = students.count_documents({"class_name": "Python 基础班"}) print(total, basic_count)

一个常见的兼容性坑：旧版本 pymongo 里的count()方法从 4.0 开始已经移除，继续调用会报AttributeError。遇到这个问题，统一改成count_documents即可。

5.3 aggregate 实现简单的分组统计

聚合是 MongoDB 里功能最强的一部分，第 40 天先掌握最简单的分组统计即可。比如按班级分组，统计每个班级的平均分和人数：

pipeline = [ { "$group": { "_id": "$class_name", "avg_score": {"$avg": "$score"}, "count": {"$sum": 1} } }, {"$sort": {"avg_score": -1}} ] for item in students.aggregate(pipeline): print(item)

输出示例：

{'_id': 'Python 提高班', 'avg_score': 95.0, 'count': 1} {'_id': 'Python 基础班', 'avg_score': 82.0, 'count': 2}

aggregate接收的是一个列表，列表中的每个字典代表一个聚合阶段，数据会按顺序流经这些阶段。$group的_id指定分组字段，$avg计算平均值，$sum累加计数，最后一个$sort对分组结果排序。

聚合管道的学习曲线比普通查询陡一些，但理解了这个“管道逐级处理数据”的思路，后面看$lookup、$unwind这些操作符就会顺很多。

6. 通过一个完整脚本走通全流程并验证

6.1 完整示例脚本

把前面的内容整理成一个完整脚本，保存为demo_pymongo.py：

from pymongo import MongoClient def main(): client = MongoClient("mongodb://localhost:27017/") db = client["school"] students = db["students"] # 清空集合，保证每次运行结果一致 students.delete_many({}) # 批量插入测试数据 students.insert_many([ {"name": "张三", "age": 20, "class_name": "Python 基础班", "score": 88}, {"name": "李四", "age": 22, "class_name": "Python 提高班", "score": 95}, {"name": "王五", "age": 21, "class_name": "Python 基础班", "score": 76}, {"name": "赵六", "age": 19, "class_name": "数据分析班", "score": 69}, ]) print("总人数:", students.count_documents({})) print("按班级统计平均分:") pipeline = [ {"$group": {"_id": "$class_name", "avg_score": {"$avg": "$score"}}} ] for item in students.aggregate(pipeline): print(item) # 王五加 5 分 result = students.update_one( {"name": "王五"}, {"$inc": {"score": 5}} ) print("更新匹配数:", result.matched_count, "实际修改数:", result.modified_count) print("按年龄升序展示:") for s in students.find().sort("age", 1): print(s["name"], s["age"], s["score"]) client.close() if __name__ == "__main__": main()

运行脚本：

python demo_pymongo.py

正常输出大致如下：

总人数: 4 按班级统计平均分: {'_id': 'Python 提高班', 'avg_score': 95.0} {'_id': '数据分析班', 'avg_score': 69.0} {'_id': 'Python 基础班', 'avg_score': 81.5} 更新匹配数: 1 实际修改数: 1 按年龄升序展示: 赵六 19 69 张三 20 88 王五 21 81 李四 22 95

脚本里先执行delete_many({})清空集合，是为了保证脚本可以重复运行，且每次结果一致。这种写法只适合学习环境，生产环境不要轻易这样清空数据。

6.2 与 mongosh 交叉验证结果

脚本运行后，可以在mongosh中交叉验证数据：

use school db.students.find().sort({age: 1}).pretty() db.students.countDocuments()

输出中的文档内容和脚本打印的结果应该一致。pymongo 与 mongosh 只是两个不同的客户端，访问的是同一个服务端和同一份数据，结果不可能互相矛盾。如果两边对不上，优先怀疑是否连接了不同的数据库或不同的 MongoDB 实例。

6.3 验证的关键点

不要只验证脚本不报错，还要核对数据本身的正确性：

• inserted_id是否生成了ObjectId。
• find_one返回值是不是 dict，找不到时是不是None。
• 更新后matched_count与modified_count是否符合预期。
• 删除后deleted_count是否正确。
• 聚合结果的分组和平均值是否正确。
• 脚本结束时client.close()是否执行。

注意：程序能启动、没有异常，只代表语法和执行路径没问题，不代表业务逻辑正确。数据库程序必须把写入、更新、删除后的实际数据再查出来核对一遍。

7. 常见报错与排查链路

7.1 ServerSelectionTimeoutError：连接不上

现象：

pymongo.errors.ServerSelectionTimeoutError: localhost:27017: [Errno 111] Connection refused

可能原因：

• MongoDB 服务没有启动。
• 服务端口被占用或改成了其他端口。
• 防火墙拦截了 27017 端口。

检查顺序：

sudo systemctl status mongod sudo ss -tlnp | grep 27017

如果服务未运行，先启动服务；如果端口不是 27017，需要把连接串改成对应端口；如果启用了防火墙，确认是否放行了 MongoDB 端口。

7.2 OperationFailure：认证失败

现象：

pymongo.errors.OperationFailure: Authentication failed.

可能原因：

• 连接串里没有带用户名密码。
• 用户名或密码错误。
• 用户创建在 admin 库，但连接时没有指定authSource=admin。

处理方式：在连接串中显式指定认证库和账号。

client = MongoClient("mongodb://myUser:myPass@localhost:27017/admin?authSource=admin")

开发环境学习时可以先不开认证，但生产环境必须开启认证，并且按最小权限给应用创建专用账号，不要直接使用 root。

7.3 用字符串查询 _id 导致查不到结果

现象：日期字符串能打印出来，但find_one({"_id": "6779f1a2b4c3d4e5f6a7b8c9"})返回None。

原因：_id是ObjectId类型，不是字符串。直接用字符串查询，等于拿字符串去匹配ObjectId，必然查不到。

处理方式：查询前先转换类型。

from bson.objectid import ObjectId student = students.find_one({"_id": ObjectId("6779f1a2b4c3d4e5f6a7b8c9")}) print(student)

7.4 更新后 modified_count 为 0

现象：update_one返回的matched_count是 1，但modified_count是 0。

原因：$set设置的字段值和当前值相同，MongoDB 认为没有实际变化，所以不算修改。这不是 bug，是 MongoDB 的更新语义。

处理方式：如果业务关心“匹配到但未变更”的情况，用matched_count判断；如果只关心是否改动了数据，用modified_count。

7.5 调用 count() 方法报 AttributeError

现象：

AttributeError: 'Collection' object has no attribute 'count'

原因：pymongo 4.0 已经移除了count()方法。

处理方式：统一使用count_documents({})。

常见报错可以整理成一张速查表：