Python如何操作CouchDB？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}' 已删除。")

# 假设我们有一个名为 '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()

group

reduce

# 沿用之前的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，即使中间有冲突，通过重试也解决了。

ResourceConflict

update_document_with_retry