pip install elasticsearch

from elasticsearch import Elasticsearch

# 連接到本地Elasticsearch集群
es = Elasticsearch(['localhost:9200'])

# 創(chuàng)建名為"my_index"的索引
index_name = "my_index"
es.indices.create(index=index_name, ignore=400)

# 添加文檔到索引中
doc = {
    "title": "Python實(shí)戰(zhàn)Elasticsearch",
    "author": "John Doe",
    "content": "Elasticsearch是一個(gè)功能強(qiáng)大的搜索引擎。",
}
es.index(index=index_name, body=doc)

# 執(zhí)行搜索查詢
query = {
    "query": {
        "match": {
            "content": "搜索引擎"
        }
    }
}
result = es.search(index=index_name, body=query)
for hit in result['hits']['hits']:
    print(hit['_source'])

# 更新文檔
update_doc = {
    "doc": {
        "content": "Elasticsearch是一個(gè)功能強(qiáng)大的搜索和分析引擎。"
    }
}
es.update(index=index_name, id='1', body=update_doc)

# 刪除文檔
es.delete(index=index_name, id='1')

# 刪除索引
es.indices.delete(index=index_name)

# 使用批量操作
actions = [
    { "index": { "_index": index_name }},
    { "title": "文檔1", "author": "Alice" },
    { "index": { "_index": index_name }},
    { "title": "文檔2", "author": "Bob" },
]
es.bulk(body=actions)

try:
    # 嘗試執(zhí)行Elasticsearch操作
    # ...
except Exception as e:
    # 處理異常情況
    print("發(fā)生異常：", e)

# 復(fù)雜查詢示例：布爾查詢
query = {
    "query": {
        "bool": {
            "must": [
                { "match": { "title": "Python" }},
                { "match": { "content": "Elasticsearch" }}
            ],
            "filter": [
                { "range": { "views": { "gte": 100 }}}
            ]
        }
    }
}
result = es.search(index=index_name, body=query)

# 聚合查詢示例：按作者統(tǒng)計(jì)文檔數(shù)量
aggs_query = {
    "aggs": {
        "authors": {
            "terms": { "field": "author.keyword" }
        }
    }
}
result = es.search(index=index_name, body=aggs_query)

from elasticsearch_dsl import Search, Q

# 使用Elasticsearch DSL構(gòu)建查詢
s = Search(using=es, index=index_name)
s = s.query(Q("match", title="Python") & Q("match", content="Elasticsearch"))
response = s.execute()

import logging

from elasticsearch import logger as es_logger# 設(shè)置日志級別為DEBUG
logging.basicConfig(level=logging.DEBUG)

# 打印Elasticsearch客戶端的日志
es_logger.setLevel(logging.DEBUG)

# 使用安全連接協(xié)議HTTPS
es = Elasticsearch(['https://localhost:9200'])