ZincSearch API接口调用

发布时间: 2025-04-08 10:47

↑

# ZincSearch API接口调用

本文将详细介绍ZincSearch的API接口调用方法，帮助您更好地集成和使用ZincSearch的功能。

## API概述

### 1. 基础信息

- 基础URL：`http://localhost:4080`
- 认证方式：Basic Auth
- 响应格式：JSON
- 请求方法：GET, POST, PUT, DELETE

### 2. 认证设置

```python
import requests
from requests.auth import HTTPBasicAuth

class ZincSearchClient:
    def __init__(self, host, username, password):
        self.host = host
        self.auth = HTTPBasicAuth(username, password)
        self.headers = {'Content-Type': 'application/json'}
    
    def request(self, method, endpoint, data=None):
        url = f"{self.host}{endpoint}"
        response = requests.request(
            method,
            url,
            json=data,
            auth=self.auth,
            headers=self.headers
        )
        return response.json()
```

## 索引管理API

### 1. 创建索引

```python
def create_index(self, index_name, mappings=None):
    endpoint = f"/api/index"
    data = {
        "name": index_name,
        "mappings": mappings or {
            "properties": {
                "text": {"type": "text"},
                "keyword": {"type": "keyword"}
            }
        }
    }
    return self.request('PUT', endpoint, data)
```

### 2. 删除索引

```python
def delete_index(self, index_name):
    endpoint = f"/api/index/{index_name}"
    return self.request('DELETE', endpoint)
```

### 3. 获取索引信息

```python
def get_index(self, index_name):
    endpoint = f"/api/index/{index_name}"
    return self.request('GET', endpoint)
```

## 文档操作API

### 1. 添加文档

```python
def add_document(self, index_name, document, doc_id=None):
    if doc_id:
        endpoint = f"/api/{index_name}/_doc/{doc_id}"
    else:
        endpoint = f"/api/{index_name}/_doc"
    return self.request('POST', endpoint, document)
```

### 2. 批量操作

```python
def bulk_operation(self, operations):
    endpoint = "/api/_bulk"
    return self.request('POST', endpoint, operations)
```

### 3. 更新文档

```python
def update_document(self, index_name, doc_id, document):
    endpoint = f"/api/{index_name}/_update/{doc_id}"
    return self.request('POST', endpoint, document)
```

## 搜索API

### 1. 基础搜索

```python
def search(self, index_name, query):
    endpoint = f"/api/{index_name}/_search"
    return self.request('POST', endpoint, query)
```

### 2. 高级搜索

```python
def advanced_search(self, index_name, query_params):
    search_body = {
        "search_type": "bool",
        "query": {
            "must": query_params.get('must', []),
            "should": query_params.get('should', []),
            "must_not": query_params.get('must_not', [])
        },
        "from": query_params.get('from', 0),
        "size": query_params.get('size', 10),
        "sort": query_params.get('sort', [])
    }
    
    if query_params.get('aggs'):
        search_body['aggs'] = query_params['aggs']
    
    return self.search(index_name, search_body)
```

## 实用工具API

### 1. 健康检查

```python
def health_check(self):
    endpoint = "/healthz"
    return self.request('GET', endpoint)
```

### 2. 版本信息

```python
def version_info(self):
    endpoint = "/version"
    return self.request('GET', endpoint)
```

## 完整客户端示例

```python
class ZincSearchClient:
    def __init__(self, host, username, password):
        self.host = host
        self.auth = HTTPBasicAuth(username, password)
        self.headers = {'Content-Type': 'application/json'}
    
    # 索引管理
    def create_index(self, index_name, mappings=None):
        endpoint = f"/api/index"
        data = {
            "name": index_name,
            "mappings": mappings
        }
        return self.request('PUT', endpoint, data)
    
    def delete_index(self, index_name):
        endpoint = f"/api/index/{index_name}"
        return self.request('DELETE', endpoint)
    
    def get_index(self, index_name):
        endpoint = f"/api/index/{index_name}"
        return self.request('GET', endpoint)
    
    # 文档操作
    def add_document(self, index_name, document, doc_id=None):
        if doc_id:
            endpoint = f"/api/{index_name}/_doc/{doc_id}"
        else:
            endpoint = f"/api/{index_name}/_doc"
        return self.request('POST', endpoint, document)
    
    def bulk_operation(self, operations):
        endpoint = "/api/_bulk"
        return self.request('POST', endpoint, operations)
    
    def update_document(self, index_name, doc_id, document):
        endpoint = f"/api/{index_name}/_update/{doc_id}"
        return self.request('POST', endpoint, document)
    
    def delete_document(self, index_name, doc_id):
        endpoint = f"/api/{index_name}/_doc/{doc_id}"
        return self.request('DELETE', endpoint)
    
    # 搜索操作
    def search(self, index_name, query):
        endpoint = f"/api/{index_name}/_search"
        return self.request('POST', endpoint, query)
    
    def advanced_search(self, index_name, query_params):
        search_body = {
            "search_type": "bool",
            "query": {
                "must": query_params.get('must', []),
                "should": query_params.get('should', []),
                "must_not": query_params.get('must_not', [])
            },
            "from": query_params.get('from', 0),
            "size": query_params.get('size', 10),
            "sort": query_params.get('sort', [])
        }
        
        if query_params.get('aggs'):
            search_body['aggs'] = query_params['aggs']
        
        return self.search(index_name, search_body)
    
    # 工具方法
    def health_check(self):
        endpoint = "/healthz"
        return self.request('GET', endpoint)
    
    def version_info(self):
        endpoint = "/version"
        return self.request('GET', endpoint)
    
    def request(self, method, endpoint, data=None):
        url = f"{self.host}{endpoint}"
        response = requests.request(
            method,
            url,
            json=data,
            auth=self.auth,
            headers=self.headers
        )
        return response.json()
```

## 使用示例

### 1. 初始化客户端

```python
# 创建客户端实例
client = ZincSearchClient(
    host="http://localhost:4080",
    username="admin",
    password="Complexpass#123"
)

# 检查服务健康状态
health = client.health_check()
print("Health status:", health)
```

### 2. 索引操作

```python
# 创建索引
mappings = {
    "properties": {
        "title": {"type": "text"},
        "content": {"type": "text"},
        "tags": {"type": "keyword"}
    }
}
client.create_index("articles", mappings)

# 添加文档
document = {
    "title": "示例文章",
    "content": "这是一篇示例文章的内容",
    "tags": ["示例", "测试"]
}
client.add_document("articles", document)
```

### 3. 搜索操作

```python
# 基础搜索
search_query = {
    "search_type": "match",
    "query": {
        "term": "示例",
        "field": "title"
    }
}
results = client.search("articles", search_query)

# 高级搜索
query_params = {
    "must": [
        {"term": "示例", "field": "title"}
    ],
    "should": [
        {"term": "测试", "field": "tags"}
    ],
    "size": 20,
    "sort": [
        {"field": "_score", "order": "desc"}
    ]
}
results = client.advanced_search("articles", query_params)
```

## 错误处理

```python
class ZincSearchError(Exception):
    pass

class ZincSearchClient:
    def request(self, method, endpoint, data=None):
        try:
            response = requests.request(
                method,
                f"{self.host}{endpoint}",
                json=data,
                auth=self.auth,
                headers=self.headers
            )
            
            if response.status_code >= 400:
                raise ZincSearchError(f"API Error: {response.text}")
            
            return response.json()
        except requests.exceptions.RequestException as e:
            raise ZincSearchError(f"Request Error: {str(e)}")
```

## 最佳实践

1. API使用
   - 使用连接池
   - 实现重试机制
   - 处理异常情况

2. 性能优化
   - 批量操作
   - 控制请求大小
   - 使用异步调用

3. 安全建议
   - 使用HTTPS
   - 实施访问控制
   - 保护敏感信息

## 常见问题

1. 连接问题
   - 检查网络连接
   - 验证认证信息
   - 确认服务状态

2. 性能问题
   - 优化查询结构
   - 使用批量操作
   - 控制响应大小

元素码农