CouchDB HTTP API操作的自动化脚本实现
2024-08-173.5k 阅读
CouchDB HTTP API 基础
CouchDB 简介
CouchDB 是一个面向文档的数据库，它以 JSON 格式存储数据，通过 HTTP 协议进行交互。这种设计使得 CouchDB 非常灵活，易于与各种编程语言和平台集成。它支持数据的分布式存储和复制，具备高可用性和容错能力，适用于构建 Web 应用、移动应用等多种场景。
HTTP API 核心操作
- 数据库操作
- 创建数据库：向 CouchDB 服务器发送
PUT请求到/db_name路径，其中db_name是你想要创建的数据库名称。例如，使用curl命令：
- 创建数据库：向 CouchDB 服务器发送
curl -X PUT http://localhost:5984/mydb -H 'Content-Type: application/json'
- **删除数据库**：发送 `DELETE` 请求到 `/db_name` 路径。如：
curl -X DELETE http://localhost:5984/mydb
- **列出所有数据库**：发送 `GET` 请求到 `/_all_dbs` 路径，响应会返回一个包含所有数据库名称的 JSON 数组。
curl http://localhost:5984/_all_dbs
- 文档操作
- 创建文档：向数据库路径
/db_name发送POST请求，并在请求体中包含要创建的文档内容（JSON 格式）。CouchDB 会自动为文档生成一个唯一的_id。
- 创建文档：向数据库路径
curl -X POST http://localhost:5984/mydb -H 'Content-Type: application/json' -d '{"name": "John Doe", "age": 30}'
- **获取文档**：发送 `GET` 请求到 `/db_name/doc_id` 路径，其中 `doc_id` 是文档的唯一标识符。
curl http://localhost:5984/mydb/1234567890abcdef
- **更新文档**：获取文档时，响应中会包含一个 `_rev` 字段，它表示文档的版本。更新文档时，需要在请求体中包含 `_rev` 字段以及更新后的内容，然后发送 `PUT` 请求到 `/db_name/doc_id` 路径。
# 首先获取文档
curl -X GET http://localhost:5984/mydb/1234567890abcdef
# 假设返回的 _rev 是 1-abcdef
curl -X PUT http://localhost:5984/mydb/1234567890abcdef -H 'Content-Type: application/json' -d '{"name": "Jane Doe", "age": 31, "_rev": "1-abcdef"}'
- **删除文档**：发送 `DELETE` 请求到 `/db_name/doc_id?rev=doc_rev` 路径，其中 `doc_rev` 是文档的当前 `_rev` 值。
curl -X DELETE http://localhost:5984/mydb/1234567890abcdef?rev=1-abcdef
使用 Python 实现自动化脚本
安装必要的库
在 Python 中，我们可以使用 requests 库来与 CouchDB 的 HTTP API 进行交互。通过 pip 安装：
pip install requests
数据库操作脚本
- 创建数据库
import requests
def create_database(db_name, url='http://localhost:5984'):
db_url = f'{url}/{db_name}'
response = requests.put(db_url)
if response.status_code == 201:
print(f"数据库 {db_name} 创建成功")
elif response.status_code == 412:
print(f"数据库 {db_name} 已存在")
else:
print(f"创建数据库失败，状态码: {response.status_code}")
- 删除数据库
def delete_database(db_name, url='http://localhost:5984'):
db_url = f'{url}/{db_name}'
response = requests.delete(db_url)
if response.status_code == 200:
print(f"数据库 {db_name} 删除成功")
elif response.status_code == 404:
print(f"数据库 {db_name} 不存在")
else:
print(f"删除数据库失败，状态码: {response.status_code}")
- 列出所有数据库
def list_databases(url='http://localhost:5984'):
all_dbs_url = f'{url}/_all_dbs'
response = requests.get(all_dbs_url)
if response.status_code == 200:
dbs = response.json()
print("所有数据库:")
for db in dbs:
print(db)
else:
print(f"获取数据库列表失败，状态码: {response.status_code}")
文档操作脚本
- 创建文档
def create_document(db_name, doc, url='http://localhost:5984'):
db_url = f'{url}/{db_name}'
response = requests.post(db_url, json=doc)
if response.status_code == 201:
print(f"文档创建成功，ID: {response.json()['id']}")
else:
print(f"创建文档失败，状态码: {response.status_code}")
- 获取文档
def get_document(db_name, doc_id, url='http://localhost:5984'):
doc_url = f'{url}/{db_name}/{doc_id}'
response = requests.get(doc_url)
if response.status_code == 200:
print(f"文档内容: {response.json()}")
elif response.status_code == 404:
print(f"文档 {doc_id} 不存在")
else:
print(f"获取文档失败，状态码: {response.status_code}")
- 更新文档
def update_document(db_name, doc_id, updated_doc, url='http://localhost:5984'):
doc_url = f'{url}/{db_name}/{doc_id}'
# 首先获取当前文档的 _rev
current_doc_response = requests.get(doc_url)
if current_doc_response.status_code == 200:
current_rev = current_doc_response.json()['_rev']
updated_doc['_rev'] = current_rev
response = requests.put(doc_url, json=updated_doc)
if response.status_code == 201:
print(f"文档 {doc_id} 更新成功")
else:
print(f"更新文档失败，状态码: {response.status_code}")
elif current_doc_response.status_code == 404:
print(f"文档 {doc_id} 不存在")
else:
print(f"获取当前文档失败，状态码: {current_doc_response.status_code}")
- 删除文档
def delete_document(db_name, doc_id, url='http://localhost:5984'):
doc_url = f'{url}/{db_name}/{doc_id}'
# 首先获取当前文档的 _rev
current_doc_response = requests.get(doc_url)
if current_doc_response.status_code == 200:
current_rev = current_doc_response.json()['_rev']
delete_url = f'{doc_url}?rev={current_rev}'
response = requests.delete(delete_url)
if response.status_code == 200:
print(f"文档 {doc_id} 删除成功")
else:
print(f"删除文档失败，状态码: {response.status_code}")
elif current_doc_response.status_code == 404:
print(f"文档 {doc_id} 不存在")
else:
print(f"获取当前文档失败，状态码: {current_doc_response.status_code}")
综合示例脚本
if __name__ == '__main__':
# 创建数据库
create_database('test_db')
# 列出所有数据库
list_databases()
# 创建文档
doc = {"name": "示例文档", "description": "这是一个测试文档"}
create_document('test_db', doc)
# 获取文档
get_document('test_db', '刚刚创建文档的ID')
# 更新文档
updated_doc = {"name": "更新后的示例文档", "description": "这是更新后的测试文档"}
update_document('test_db', '刚刚创建文档的ID', updated_doc)
# 删除文档
delete_document('test_db', '刚刚创建文档的ID')
# 删除数据库
delete_database('test_db')
使用 JavaScript（Node.js）实现自动化脚本
安装必要的库
在 Node.js 项目中，我们可以使用 request 库（虽然现在有更现代的 node - fetch 等库，但 request 库在处理 HTTP 请求方面仍然广泛使用且功能强大）来与 CouchDB 的 HTTP API 交互。首先初始化一个 package.json 文件：
npm init -y
然后安装 request 库：
npm install request
数据库操作脚本
- 创建数据库
const request = require('request');
function createDatabase(dbName, url = 'http://localhost:5984') {
const dbUrl = `${url}/${dbName}`;
request.put(dbUrl, function (error, response, body) {
if (!error && response.statusCode === 201) {
console.log(`数据库 ${dbName} 创建成功`);
} else if (response.statusCode === 412) {
console.log(`数据库 ${dbName} 已存在`);
} else {
console.log(`创建数据库失败，状态码: ${response.statusCode}`);
}
});
}
- 删除数据库
function deleteDatabase(dbName, url = 'http://localhost:5984') {
const dbUrl = `${url}/${dbName}`;
request.delete(dbUrl, function (error, response, body) {
if (!error && response.statusCode === 200) {
console.log(`数据库 ${dbName} 删除成功`);
} else if (response.statusCode === 404) {
console.log(`数据库 ${dbName} 不存在`);
} else {
console.log(`删除数据库失败，状态码: ${response.statusCode}`);
}
});
}
- 列出所有数据库
function listDatabases(url = 'http://localhost:5984') {
const allDbsUrl = `${url}/_all_dbs`;
request.get(allDbsUrl, function (error, response, body) {
if (!error && response.statusCode === 200) {
const dbs = JSON.parse(body);
console.log("所有数据库:");
dbs.forEach(db => {
console.log(db);
});
} else {
console.log(`获取数据库列表失败，状态码: ${response.statusCode}`);
}
});
}
文档操作脚本
- 创建文档
function createDocument(dbName, doc, url = 'http://localhost:5984') {
const dbUrl = `${url}/${dbName}`;
request.post({
url: dbUrl,
json: doc
}, function (error, response, body) {
if (!error && response.statusCode === 201) {
console.log(`文档创建成功，ID: ${body.id}`);
} else {
console.log(`创建文档失败，状态码: ${response.statusCode}`);
}
});
}
- 获取文档
function getDocument(dbName, docId, url = 'http://localhost:5984') {
const docUrl = `${url}/${dbName}/${docId}`;
request.get(docUrl, function (error, response, body) {
if (!error && response.statusCode === 200) {
console.log(`文档内容: ${JSON.parse(body)}`);
} else if (response.statusCode === 404) {
console.log(`文档 ${docId} 不存在`);
} else {
console.log(`获取文档失败，状态码: ${response.statusCode}`);
}
});
}
- 更新文档
function updateDocument(dbName, docId, updatedDoc, url = 'http://localhost:5984') {
const docUrl = `${url}/${dbName}/${docId}`;
// 首先获取当前文档的 _rev
request.get(docUrl, function (error, response, body) {
if (!error && response.statusCode === 200) {
const currentRev = JSON.parse(body)._rev;
updatedDoc._rev = currentRev;
request.put({
url: docUrl,
json: updatedDoc
}, function (error, response, body) {
if (!error && response.statusCode === 201) {
console.log(`文档 ${docId} 更新成功`);
} else {
console.log(`更新文档失败，状态码: ${response.statusCode}`);
}
});
} else if (response.statusCode === 404) {
console.log(`文档 ${docId} 不存在`);
} else {
console.log(`获取当前文档失败，状态码: ${response.statusCode}`);
}
});
}
- 删除文档
function deleteDocument(dbName, docId, url = 'http://localhost:5984') {
const docUrl = `${url}/${dbName}/${docId}`;
// 首先获取当前文档的 _rev
request.get(docUrl, function (error, response, body) {
if (!error && response.statusCode === 200) {
const currentRev = JSON.parse(body)._rev;
const deleteUrl = `${docUrl}?rev=${currentRev}`;
request.delete(deleteUrl, function (error, response, body) {
if (!error && response.statusCode === 200) {
console.log(`文档 ${docId} 删除成功`);
} else {
console.log(`删除文档失败，状态码: ${response.statusCode}`);
}
});
} else if (response.statusCode === 404) {
console.log(`文档 ${docId} 不存在`);
} else {
console.log(`获取当前文档失败，状态码: ${response.statusCode}`);
}
});
}
综合示例脚本
if (require.main === module) {
// 创建数据库
createDatabase('test_db');
// 列出所有数据库
listDatabases();
// 创建文档
const doc = {name: "示例文档", description: "这是一个测试文档"};
createDocument('test_db', doc);
// 获取文档
getDocument('test_db', '刚刚创建文档的ID');
// 更新文档
const updatedDoc = {name: "更新后的示例文档", description: "这是更新后的测试文档"};
updateDocument('test_db', '刚刚创建文档的ID', updatedDoc);
// 删除文档
deleteDocument('test_db', '刚刚创建文档的ID');
// 删除数据库
deleteDatabase('test_db');
}
自动化脚本的优化与扩展
错误处理优化
在上述脚本中，虽然已经对常见的 HTTP 状态码进行了处理，但可以进一步完善错误处理。例如，在 Python 脚本中，可以捕获 requests 库可能抛出的异常，如网络连接错误等。
import requests
def create_database(db_name, url='http://localhost:5984'):
try:
db_url = f'{url}/{db_name}'
response = requests.put(db_url)
if response.status_code == 201:
print(f"数据库 {db_name} 创建成功")
elif response.status_code == 412:
print(f"数据库 {db_name} 已存在")
else:
print(f"创建数据库失败，状态码: {response.status_code}")
except requests.RequestException as e:
print(f"请求发生错误: {e}")
在 JavaScript 脚本中，同样可以在 request 回调函数中更好地处理错误，比如区分不同类型的网络错误等。
配置管理
为了使脚本更具可扩展性和灵活性，可以将 CouchDB 服务器的 URL、认证信息（如果需要）等配置参数提取到一个配置文件中。在 Python 中，可以使用 configparser 模块来读取配置文件。
- 创建配置文件（config.ini）
[CouchDB]
url = http://localhost:5984
username = admin
password = password
- 读取配置文件并应用到脚本中
import requests
from configparser import ConfigParser
def create_database(db_name):
config = ConfigParser()
config.read('config.ini')
url = config.get('CouchDB', 'url')
auth = (config.get('CouchDB', 'username'), config.get('CouchDB', 'password'))
try:
db_url = f'{url}/{db_name}'
response = requests.put(db_url, auth=auth)
if response.status_code == 201:
print(f"数据库 {db_name} 创建成功")
elif response.status_code == 412:
print(f"数据库 {db_name} 已存在")
else:
print(f"创建数据库失败，状态码: {response.status_code}")
except requests.RequestException as e:
print(f"请求发生错误: {e}")
在 JavaScript 中，可以使用 ini - parser 等库来读取类似的配置文件，并将配置应用到请求中。
批量操作
在实际应用中，可能需要对多个文档进行批量创建、更新或删除操作。以批量创建文档为例，在 Python 中可以这样实现：
def batch_create_documents(db_name, docs):
config = ConfigParser()
config.read('config.ini')
url = config.get('CouchDB', 'url')
auth = (config.get('CouchDB', 'username'), config.get('CouchDB', 'password'))
db_url = f'{url}/{db_name}/_bulk_docs'
data = {'docs': docs}
try:
response = requests.post(db_url, json=data, auth=auth)
if response.status_code == 201:
print("批量创建文档成功")
else:
print(f"批量创建文档失败，状态码: {response.status_code}")
except requests.RequestException as e:
print(f"请求发生错误: {e}")
在 JavaScript 中，可以利用 async - await 结合循环来实现类似的批量操作，通过控制并发数量来避免对服务器造成过大压力。
与其他系统集成
与 Web 应用集成
CouchDB 的自动化脚本可以与 Web 应用后端集成。例如，在 Python 的 Flask 应用中，可以调用上述 CouchDB 操作脚本。
from flask import Flask, jsonify, request
import requests
from configparser import ConfigParser
app = Flask(__name__)
def create_database(db_name):
config = ConfigParser()
config.read('config.ini')
url = config.get('CouchDB', 'url')
auth = (config.get('CouchDB', 'username'), config.get('CouchDB', 'password'))
try:
db_url = f'{url}/{db_name}'
response = requests.put(db_url, auth=auth)
if response.status_code == 201:
return True
elif response.status_code == 412:
return True
else:
return False
except requests.RequestException as e:
return False
@app.route('/create - database', methods=['POST'])
def create_db_route():
data = request.get_json()
db_name = data.get('db_name')
if create_database(db_name):
return jsonify({'message': '数据库创建成功或已存在'}), 200
else:
return jsonify({'message': '数据库创建失败'}), 500
if __name__ == '__main__':
app.run(debug=True)
在 JavaScript 的 Express 应用中，也可以类似地集成 CouchDB 操作，通过路由来暴露 CouchDB 的 API 功能给前端应用。
与数据处理流程集成
CouchDB 的自动化脚本还可以与数据处理流程集成。比如，在一个数据采集项目中，采集到的数据可以通过自动化脚本直接存储到 CouchDB 中。假设使用 Python 进行数据采集，采集到的数据存储在一个列表中：
data_list = [{"name": "data1", "value": 10}, {"name": "data2", "value": 20}]
for data in data_list:
create_document('data_db', data)
这样，数据采集和存储过程就实现了自动化集成，并且可以通过进一步扩展脚本来实现数据的定期更新、删除等操作，以适应不同的数据处理需求。
通过以上对 CouchDB HTTP API 操作自动化脚本的详细介绍，包括基础操作、使用不同编程语言实现脚本、脚本的优化扩展以及与其他系统的集成，希望读者能够掌握如何根据实际需求编写高效、灵活的 CouchDB 自动化脚本，充分发挥 CouchDB 在各种项目中的优势。