Python如何操作CouchDB？couchdb-python

couchdb-python

couchdb-python

pip install couchdb

import couchdb
import json

# 1. 连接到CouchDB服务器
# 假设CouchDB运行在本地默认端口5984，没有认证或使用默认admin/password
# 如果有认证，可以这样：server = couchdb.Server('http://admin:password@localhost:5984/')
try:
    server = couchdb.Server('http://localhost:5984/')
    print("成功连接到CouchDB服务器。")
except Exception as e:
    print(f"连接CouchDB失败: {e}")
    # 实际应用中，这里应该有更健壮的错误处理

# 2. 选择或创建数据库
db_name = 'my_test_database'
if db_name in server:
    db = server[db_name]
    print(f"数据库 '{db_name}' 已存在，已连接。")
else:
    db = server.create(db_name)
    print(f"数据库 '{db_name}' 不存在，已创建。")

# 3. 创建（保存）文档
doc_id, doc_rev = db.save({'type': 'book', 'title': 'Python & CouchDB Guide', 'author': 'Alice', 'year': 2023})
print(f"文档 '{doc_id}' (rev: {doc_rev}) 已创建。")

# 4. 读取文档
# 方式一：通过ID直接获取
fetched_doc = db[doc_id]
print(f"通过ID读取文档: {json.dumps(fetched_doc, indent=2, ensure_ascii=False)}")

# 方式二：如果不知道ID，但知道rev，也可以
# fetched_doc_with_rev = db.get(doc_id, rev=doc_rev)

# 5. 更新文档
# CouchDB是基于MVCC（多版本并发控制）的，更新文档需要提供当前文档的_rev
fetched_doc['status'] = 'published'
doc_id, doc_rev = db.save(fetched_doc) # 注意：save会返回新的_id和_rev
print(f"文档 '{doc_id}' (rev: {doc_rev}) 已更新。")

# 再次读取，确认更新
updated_doc = db[doc_id]
print(f"更新后文档: {json.dumps(updated_doc, indent=2, ensure_ascii=False)}")

# 6. 删除文档
# 删除也需要提供_rev
db.delete(updated_doc) # 或者 db.delete(doc_id) 也可以，但内部会先获取最新rev
print(f"文档 '{doc_id}' 已删除。")

# 尝试再次读取，会抛出ResourceNotFound异常
try:
    db[doc_id]
except couchdb.http.ResourceNotFound:
    print(f"文档 '{doc_id}' 已确认删除，无法找到。")

# 7. 批量操作（效率更高）
docs_to_save = [
    {'type': 'article', 'title': 'The Power of NoSQL', 'tags': ['database', 'nosql']},
    {'type': 'article', 'title': 'Distributed Systems Basics', 'tags': ['architecture', 'cloud']}
]
# save_documents 会返回一个包含 (id, rev) 或 (id, error_message) 的列表
results = db.update(docs_to_save) # update方法可以用于批量保存或更新
print("\n批量保存结果:")
for success, doc_id, rev_or_error in results:
    if success:
        print(f"  成功保存文档 '{doc_id}' (rev: {rev_or_error})")
    else:
        print(f"  保存文档 '{doc_id}' 失败: {rev_or_error}")

# 8. 删除数据库（谨慎操作）
# server.delete(db_name)
# print(f"数据库 '{db_name}' 已删除。")

couchdb-python

_rev

couchdb-python

# 假设我们有一个名为 'my_test_database' 的数据库
# db = server['my_test_database'] # 沿用上面的db对象

# 1. 创建一个设计文档和视图
# 视图的Map函数：根据文档类型和年份发出键值对
# 视图的Reduce函数：统计每个年份的文档数量
design_doc_id = '_design/my_app'
try:
    design_doc = db[design_doc_id]
except couchdb.http.ResourceNotFound:
    design_doc = {
        '_id': design_doc_id,
        'views': {
            'docs_by_year': {
                'map': """
function(doc) {
    if (doc.type && doc.year) {
        emit(doc.year, 1);
    }
}
                """,
                'reduce': """
_sum
                """
            },
            'all_books': {
                'map': """
function(doc) {
    if (doc.type === 'book') {
        emit(doc._id, doc.title);
    }
}
                """
            }
        }
    }
    db.save(design_doc)
    print(f"设计文档 '{design_doc_id}' 已创建或更新。")

# 2. 插入一些测试数据
test_docs = [
    {'type': 'book', 'title': 'Data Science Handbook', 'author': 'Charlie', 'year': 2022},
    {'type': 'article', 'title': 'AI in Healthcare', 'author': 'David', 'year': 2023},
    {'type': 'book', 'title': 'Cloud Native Patterns', 'author': 'Eve', 'year': 2022},
    {'type': 'article', 'title': 'Quantum Computing Intro', 'author': 'Frank', 'year': 2024},
]
db.update(test_docs)
print("测试文档已插入。")

# 3. 查询视图
# 查询 'docs_by_year' 视图，使用 reduce 聚合
print("\n查询 'docs_by_year' 视图 (按年份统计):")
for row in db.view('my_app/docs_by_year', group=True):
    print(f"  年份: {row.key}, 数量: {row.value}")

# 查询 'all_books' 视图，获取所有书籍
print("\n查询 'all_books' 视图 (所有书籍):")
for row in db.view('my_app/all_books'):
    print(f"  ID: {row.id}, 标题: {row.value}")

# 4. 带参数查询视图
# 查询2022年的所有文档
print("\n查询 'docs_by_year' 视图 (仅2022年，不聚合):")
for row in db.view('my_app/docs_by_year', key=2022):
    print(f"  文档ID: {row.id}, 键: {row.key}, 值: {row.value}")

# 查询2023年及之后的文档
print("\n查询 'docs_by_year' 视图 (2023年及之后):")
for row in db.view('my_app/docs_by_year', startkey=2023):
    print(f"  文档ID: {row.id}, 键: {row.key}, 值: {row.value}")

# 分页查询 (跳过前1个，取2个)
print("\n分页查询 'all_books' 视图 (跳过1个，取2个):")
for row in db.view('my_app/all_books', skip=1, limit=2):
    print(f"  ID: {row.id}, 标题: {row.value}")

# 注意：当视图被首次查询或数据有变化时，CouchDB会异步地构建或更新索引。
# 这意味着第一次查询可能会慢一点，但后续查询会非常快。

db.view()

key

startkey

endkey

limit

skip

group

reduce

_rev

couchdb-python

# 沿用之前的db对象
# db = server['my_test_database']

# 1. 模拟并发冲突
# 假设我们有一个文档
doc_id_conflict, _ = db.save({'type': 'product', 'name': 'Laptop', 'price': 1200})
print(f"\n创建冲突测试文档: {doc_id_conflict}")

# 客户端A获取文档
doc_a = db[doc_id_conflict]
print(f"客户端A获取文档: {doc_a['_rev']}")

# 客户端B也获取文档（此时和A获取的是同一个版本）
doc_b = db[doc_id_conflict]
print(f"客户端B获取文档: {doc_b['_rev']}")

# 客户端A修改并保存
doc_a['price'] = 1150
try:
    db.save(doc_a)
    print("客户端A成功更新文档。")
except couchdb.http.ResourceConflict as e:
    print(f"客户端A更新失败（不应该发生）: {e}")

# 客户端B修改并尝试保存（此时会发生冲突，因为_rev已过期）
doc_b['quantity'] = 10
try:
    db.save(doc_b)
    print("客户端B成功更新文档。") # 这行不会被执行
except couchdb.http.ResourceConflict as e:
    print(f"客户端B更新失败，发生冲突: {e}")
    # 冲突发生时，CouchDB会在文档中创建一个新的“冲突版本”
    # 你可以通过获取文档的_conflicts字段来查看这些冲突版本

# 2. 冲突解决策略
# 常见的策略是：
#   a. 重试：重新获取最新版本，合并修改，然后再次尝试保存。
#   b. 业务逻辑决策：根据业务规则，决定保留哪个版本或如何合并。
# couchdb-python的update()方法在处理单个文档时，可以帮你自动重试几次。
# 但对于复杂的合并逻辑，你需要自己实现。

def update_document_with_retry(doc_id, update_func, max_retries=5):
    """
    一个简单的冲突解决函数：重试更新，直到成功或达到最大重试次数。
    update_func 是一个接受文档并返回修改后文档的函数。
    """
    for i in range(max_retries):
        try:
            doc = db[doc_id]
            updated_doc = update_func(doc)
            db.save(updated_doc)
            print(f"第 {i+1} 次尝试：文档 '{doc_id}' 更新成功。")
            return True
        except couchdb.http.ResourceConflict:
            print(f"第 {i+1} 次尝试：文档 '{doc_id}' 发生冲突，重试...")
            # 冲突时，不需要特别做什么，下次循环会重新获取最新版本
        except couchdb.http.ResourceNotFound:
            print(f"文档 '{doc_id}' 不存在。")
            return False
        except Exception as e:
            print(f"更新文档 '{doc_id}' 时发生未知错误: {e}")
            return False
    print(f"文档 '{doc_id}' 达到最大重试次数，更新失败。")
    return False

# 模拟一个需要更新的场景
doc_id_retry, _ = db.save({'type': 'counter', 'value': 0})

def increment_counter(doc):
    doc['value'] += 1
    return doc

# 多个“客户端”同时尝试更新
import threading
def client_task(doc_id, task_name):
    print(f"{task_name} 启动，尝试更新...")
    if update_document_with_retry(doc_id, increment_counter):
        print(f"{task_name} 完成更新。")
    else:
        print(f"{task_name} 更新失败。")

threads = []
for i in range(3):
    thread = threading.Thread(target=client_task, args=(doc_id_retry, f"客户端{i+1}"))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

final_doc = db[doc_id_retry]
print(f"\n最终文档 '{doc_id_retry}' 的值: {final_doc['value']}")
# 理论上，最终值应该是3，即使中间有冲突，通过重试也解决了。

_rev

_rev

_rev

ResourceConflict

update_document_with_retry