Python客户端 #

安装 #

bash

pip install algoliasearch

初始化 #

基本初始化 #

python

from algoliasearch.search_client import SearchClient

client = SearchClient.create('APP_ID', 'API_KEY')
index = client.init_index('products')

环境变量配置 #

python

import os
from algoliasearch.search_client import SearchClient

client = SearchClient.create(
    os.environ.get('ALGOLIA_APP_ID'),
    os.environ.get('ALGOLIA_API_KEY')
)

搜索操作 #

基本搜索 #

python

results = index.search('iphone')

print(f"找到 {results['nbHits']} 条结果")
for hit in results['hits']:
    print(hit['name'])

带参数搜索 #

python

results = index.search('iphone', {
    'hitsPerPage': 20,
    'page': 0,
    'filters': 'brand:Apple',
    'facets': ['brand', 'category'],
    'attributesToHighlight': ['name', 'description']
})

搜索结果处理 #

python

def process_results(results):
    return {
        'items': [
            {
                'id': hit['objectID'],
                'name': hit.get('_highlightResult', {}).get('name', {}).get('value', hit['name']),
                'price': hit['price']
            }
            for hit in results['hits']
        ],
        'total': results['nbHits'],
        'page': results['page'],
        'total_pages': results['nbPages']
    }

索引操作 #

添加记录 #

python

# 添加单条记录
index.save_object({
    'objectID': 'prod-001',
    'name': 'iPhone 15 Pro',
    'price': 999
})

# 批量添加
index.save_objects([
    {'objectID': 'prod-001', 'name': 'Product 1'},
    {'objectID': 'prod-002', 'name': 'Product 2'}
])

更新记录 #

python

# 完全替换
index.save_object({
    'objectID': 'prod-001',
    'name': 'Updated Name',
    'price': 899
})

# 部分更新
index.partial_update_object({
    'objectID': 'prod-001',
    'price': 899
})

删除记录 #

python

# 删除单条
index.delete_object('prod-001')

# 批量删除
index.delete_objects(['prod-001', 'prod-002'])

# 按条件删除
index.delete_by({'filters': 'stock:0'})

# 清空索引
index.clear_objects()

批量操作 #

python

index.batch({
    'requests': [
        {'action': 'addObject', 'body': {'name': 'Product 1'}},
        {'action': 'updateObject', 'body': {'objectID': '1', 'name': 'Updated'}},
        {'action': 'deleteObject', 'objectID': '2'}
    ]
})

索引设置 #

获取设置 #

python

settings = index.get_settings()
print(settings)

更新设置 #

python

index.set_settings({
    'searchableAttributes': ['name', 'description', 'brand'],
    'attributesForFaceting': ['brand', 'category'],
    'customRanking': ['desc(rating)'],
    'hitsPerPage': 20
})

分面搜索 #

获取分面值 #

python

results = index.search('', {
    'facets': ['brand', 'category']
})

print(results['facets'])

搜索分面值 #

python

result = index.search_for_facet_values('brand', 'Ap')
print(result['facetHits'])

同义词操作 #

python

# 添加同义词
index.save_synonym({
    'objectID': 'syn-1',
    'type': 'synonym',
    'synonyms': ['手机', '电话', '移动电话']
})

# 获取同义词
synonym = index.get_synonym('syn-1')

# 删除同义词
index.delete_synonym('syn-1')

规则操作 #

python

# 添加规则
index.save_rule({
    'objectID': 'rule-1',
    'condition': {'pattern': 'sale', 'anchoring': 'contains'},
    'consequence': {
        'params': {'filters': 'onSale:true'}
    }
})

# 获取规则
rule = index.get_rule('rule-1')

# 删除规则
index.delete_rule('rule-1')

Django集成 #

视图示例 #

python

from django.http import JsonResponse
from django.views import View
from algoliasearch.search_client import SearchClient

client = SearchClient.create('APP_ID', 'SEARCH_KEY')
index = client.init_index('products')

class SearchView(View):
    def get(self, request):
        query = request.GET.get('q', '')
        page = int(request.GET.get('page', 0))
        
        results = index.search(query, {
            'page': page,
            'hitsPerPage': 20,
            'attributesToHighlight': ['name', 'description']
        })
        
        return JsonResponse({
            'success': True,
            'data': results
        })

模型同步 #

python

from django.db import models
from algoliasearch.search_client import SearchClient

client = SearchClient.create('APP_ID', 'ADMIN_KEY')
index = client.init_index('products')

class Product(models.Model):
    name = models.CharField(max_length=200)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    brand = models.CharField(max_length=100)
    category = models.CharField(max_length=100)
    
    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)
        self.index_to_algolia()
    
    def delete(self, *args, **kwargs):
        index.delete_object(str(self.id))
        super().delete(*args, **kwargs)
    
    def index_to_algolia(self):
        index.save_object({
            'objectID': str(self.id),
            'name': self.name,
            'price': float(self.price),
            'brand': self.brand,
            'category': self.category
        })

Flask集成 #

python

from flask import Flask, request, jsonify
from algoliasearch.search_client import SearchClient

app = Flask(__name__)
client = SearchClient.create('APP_ID', 'API_KEY')
index = client.init_index('products')

@app.route('/search')
def search():
    query = request.args.get('q', '')
    page = int(request.args.get('page', 0))
    
    results = index.search(query, {
        'page': page,
        'hitsPerPage': 20
    })
    
    return jsonify({
        'success': True,
        'data': results
    })

@app.route('/index', methods=['POST'])
def index_products():
    products = request.json.get('products', [])
    
    index.save_objects(products)
    
    return jsonify({'success': True})

if __name__ == '__main__':
    app.run(debug=True)

数据同步 #

数据库同步 #

python

import psycopg2
from algoliasearch.search_client import SearchClient

client = SearchClient.create('APP_ID', 'ADMIN_KEY')
index = client.init_index('products')

def sync_products():
    conn = psycopg2.connect(os.environ.get('DATABASE_URL'))
    cursor = conn.cursor()
    
    cursor.execute('SELECT id, name, price, brand, category FROM products WHERE active = true')
    
    products = [
        {
            'objectID': str(row[0]),
            'name': row[1],
            'price': float(row[2]),
            'brand': row[3],
            'category': row[4]
        }
        for row in cursor.fetchall()
    ]
    
    index.save_objects(products)
    
    cursor.close()
    conn.close()
    
    print(f'Synced {len(products)} products')

错误处理 #

python

from algoliasearch.exceptions import AlgoliaException

try:
    results = index.search('iphone')
except AlgoliaException as e:
    print(f'Algolia error: {e}')

完整示例 #

python

import os
from flask import Flask, request, jsonify
from algoliasearch.search_client import SearchClient
from algoliasearch.exceptions import AlgoliaException

app = Flask(__name__)

client = SearchClient.create(
    os.environ.get('ALGOLIA_APP_ID'),
    os.environ.get('ALGOLIA_API_KEY')
)
index = client.init_index('products')

@app.route('/api/search', methods=['GET'])
def search():
    try:
        query = request.args.get('q', '')
        page = int(request.args.get('page', 0))
        brand = request.args.get('brand')
        category = request.args.get('category')
        
        filters = []
        if brand:
            filters.append(f'brand:{brand}')
        if category:
            filters.append(f'category:{category}')
        
        results = index.search(query, {
            'page': page,
            'hitsPerPage': 20,
            'filters': ' AND '.join(filters) if filters else None,
            'facets': ['brand', 'category'],
            'attributesToHighlight': ['name', 'description']
        })
        
        return jsonify({
            'success': True,
            'data': results
        })
    
    except AlgoliaException as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/index', methods=['POST'])
def index_products():
    try:
        products = request.json.get('products', [])
        index.save_objects(products)
        return jsonify({'success': True})
    except AlgoliaException as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

if __name__ == '__main__':
    app.run(debug=True)

总结 #

Python客户端要点：

要点	说明
初始化	SearchClient.create(APP_ID, API_KEY)
搜索	index.search(query, params)
索引	save_objects, delete_objects
设置	set_settings, get_settings
框架	Django, Flask集成

接下来，让我们学习索引设计原则。