Python脚本如何操作文档数据库

wen 实用脚本 2026-07-10 1

本文目录导读：

Python脚本如何操作文档数据库

安装驱动
PyMongo基础操作
MongoEngine ODM操作
实际应用示例
性能优化建议
错误处理

我来详细介绍Python操作文档数据库（主要是MongoDB）的方法。

安装驱动

MongoDB驱动

pip install pymongo
# 或
pip install mongoengine  # ODM方式

其他文档数据库

pip install couchdb   # CouchDB
pip install rethinkdb # RethinkDB

PyMongo基础操作

连接数据库

from pymongo import MongoClient
# 本地连接
client = MongoClient('mongodb://localhost:27017/')
# 带认证的连接
client = MongoClient('mongodb://username:password@localhost:27017/')
# 连接数据库
db = client['mydatabase']
collection = db['users']

CRUD操作

# 插入文档
def insert_document():
    user = {
        'name': '张三',
        'age': 25,
        'email': 'zhangsan@example.com',
        'skills': ['Python', 'JavaScript'],
        'address': {
            'city': '北京',
            'district': '朝阳区'
        }
    }
    # 插入单个文档
    result = collection.insert_one(user)
    print(f'插入ID: {result.inserted_id}')
    # 插入多个文档
    users = [
        {'name': '李四', 'age': 30},
        {'name': '王五', 'age': 28}
    ]
    result = collection.insert_many(users)
    print(f'插入IDs: {result.inserted_ids}')
# 查询文档
def query_documents():
    # 查询所有
    for doc in collection.find():
        print(doc)
    # 条件查询
    result = collection.find({'age': {'$gt': 25}})
    # 查询单个文档
    user = collection.find_one({'name': '张三'})
    # 复杂查询
    result = collection.find({
        '$or': [
            {'age': {'$lt': 20}},
            {'age': {'$gt': 30}}
        ],
        'skills': 'Python'
    }).sort('age', -1).limit(10)
# 更新文档
def update_documents():
    # 更新单个
    collection.update_one(
        {'name': '张三'},
        {'$set': {'age': 26, 'updated': True}}
    )
    # 更新多个
    collection.update_many(
        {'age': {'$lt': 20}},
        {'$inc': {'age': 1}}  # 年龄加1
    )
    # 替换文档
    collection.replace_one(
        {'name': '张三'},
        {'name': '张三', 'age': 27, 'role': 'admin'}
    )
# 删除文档
def delete_documents():
    # 删除单个
    collection.delete_one({'name': '张三'})
    # 删除多个
    collection.delete_many({'age': {'$lt': 18}})
    # 删除集合
    collection.drop()

高级查询操作

# 聚合管道
def aggregation_example():
    pipeline = [
        {'$match': {'status': 'active'}},
        {'$group': {
            '_id': '$city',
            'count': {'$sum': 1},
            'avg_age': {'$avg': '$age'}
        }},
        {'$sort': {'count': -1}},
        {'$limit': 5}
    ]
    result = collection.aggregate(pipeline)
    for doc in result:
        print(doc)
# 索引操作
def index_operations():
    # 创建索引
    collection.create_index('email', unique=True)
    collection.create_index([('name', 1), ('age', -1)])
    # 创建文本索引
    collection.create_index([('description', 'text')])
    # 文本搜索
    result = collection.find(
        {'$text': {'$search': 'python mongodb'}},
        {'score': {'$meta': 'textScore'}}
    ).sort([('score', {'$meta': 'textScore'})])
# 地理位置查询
def geo_query():
    # 创建2dsphere索引
    collection.create_index([('location', '2dsphere')])
    # 查询附近地点
    result = collection.find({
        'location': {
            '$near': {
                '$geometry': {
                    'type': 'Point',
                    'coordinates': [116.4, 39.9]  # 北京坐标
                },
                '$maxDistance': 1000,  # 米
                '$minDistance': 0
            }
        }
    })

MongoEngine ODM操作

定义模型

from mongoengine import Document, StringField, IntField, ListField, DictField, connect
# 连接数据库
connect('mydatabase', host='localhost', port=27017)
class User(Document):
    name = StringField(required=True, max_length=50)
    age = IntField(min_value=0, max_value=150)
    email = StringField(required=True, unique=True)
    skills = ListField(StringField())
    address = DictField()
    meta = {
        'indexes': [
            'email',
            ('name', 'age')
        ]
    }

使用ODM操作

# 创建文档
def create_user():
    user = User(
        name='张三',
        age=25,
        email='zhangsan@example.com',
        skills=['Python', 'MongoDB'],
        address={'city': '北京', 'district': '海淀'}
    )
    user.save()
# 查询文档
def query_users():
    # 查询所有
    users = User.objects.all()
    # 条件查询
    users = User.objects(age__gte=18, skills__in=['Python'])
    # 排序和限制
    users = User.objects.order_by('-age').limit(10)
    # 聚合查询
    result = User.objects.aggregate(
        {'$group': {
            '_id': '$address.city',
            'count': {'$sum': 1}
        }}
    )
# 更新文档
def update_user():
    # 更新单个
    User.objects(name='张三').update_one(set__age=26)
    # 批量更新
    User.objects(age__lt=18).update(inc__age=1)
    # 原子操作
    User.objects(id=user_id).update_one(
        push__skills='JavaScript'
    )
# 删除操作
def delete_users():
    User.objects(age__lt=18).delete()
    User.objects(name='张三').first().delete()

实际应用示例

用户管理系统

class UserManager:
    def __init__(self, collection):
        self.collection = collection
    def register_user(self, username, password, email):
        """注册用户"""
        user = {
            'username': username,
            'password': self._hash_password(password),
            'email': email,
            'created_at': datetime.now(),
            'last_login': None,
            'status': 'active',
            'roles': ['user']
        }
        try:
            result = self.collection.insert_one(user)
            return str(result.inserted_id)
        except pymongo.errors.DuplicateKeyError:
            return None
    def find_users_by_criteria(self, criteria):
        """按条件查询用户"""
        query = {}
        if 'age_range' in criteria:
            query['age'] = {
                '$gte': criteria['age_range'][0],
                '$lte': criteria['age_range'][1]
            }
        if 'skills' in criteria:
            query['skills'] = {'$all': criteria['skills']}
        if 'city' in criteria:
            query['address.city'] = criteria['city']
        # 使用聚合框架
        pipeline = [
            {'$match': query},
            {'$lookup': {
                'from': 'orders',
                'localField': '_id',
                'foreignField': 'user_id',
                'as': 'orders'
            }},
            {'$project': {
                'username': 1,
                'email': 1,
                'order_count': {'$size': '$orders'}
            }}
        ]
        return list(self.collection.aggregate(pipeline))
    def _hash_password(self, password):
        """密码哈希处理"""
        import hashlib
        return hashlib.sha256(password.encode()).hexdigest()

性能优化建议

# 批量操作
def bulk_operations():
    operations = [
        pymongo.UpdateOne(
            {'_id': ObjectId(id)},
            {'$set': {'status': 'processed'}}
        ) for id in ids
    ]
    collection.bulk_write(operations)
# 使用游标
def use_cursor():
    cursor = collection.find(
        {'status': 'pending'},
        no_cursor_timeout=True  # 防止长时间操作超时
    ).batch_size(100)  # 设置批处理大小
    for doc in cursor:
        process_document(doc)
    cursor.close()  # 记得关闭游标
# 连接池配置
client = MongoClient(
    'mongodb://localhost:27017/',
    maxPoolSize=50,  # 最大连接数
    minPoolSize=10,   # 最小连接数
    maxIdleTimeMS=10000  # 空闲连接超时
)

错误处理

from pymongo.errors import (
    ConnectionFailure,
    OperationFailure,
    DuplicateKeyError,
    BulkWriteError
)
def safe_operation():
    try:
        result = collection.insert_one(document)
    except ConnectionFailure:
        print("数据库连接失败")
        # 重试逻辑
        retry_connection()
    except DuplicateKeyError:
        print("唯一键冲突")
        # 更新现有文档
        collection.update_one(filter, {'$set': document}, upsert=True)
    except OperationFailure as e:
        print(f"操作失败: {e.details}")
    except BulkWriteError as e:
        print(f"批量写入错误: {e.details}")

这些是Python操作文档数据库的主要方法，MongoDB是最常用的文档数据库，PyMongo提供了完整的API支持，根据项目需求,可以选择原生驱动或ODM框架。