Elasticsearch 与 GraphQL 整合:构建实时搜索 API

from flask import Flask
from flask_graphql import GraphQLView
from graphene_elasticsearch import ElasticsearchDocumentType
from elasticsearch import Elasticsearch
from graphene import ObjectType, String, Field, Schema
 
# 初始化 Flask 应用
app = Flask(__name__)
 
# 连接到 Elasticsearch 实例
es = Elasticsearch("http://localhost:9200")
 
# 定义 Elasticsearch 文档类型
class ArticleDocumentType(ElasticsearchDocumentType):
    class Meta:
        index = 'article'
        document_type = 'article'
 
# 定义 GraphQL 查询类型
class Query(ObjectType):
    search = Field(ArticleDocumentType, title=String(required=True))
 
    # 定义根据文章标题搜索文章的解析器
    def resolve_search(self, info, title):
        query = {
            "query": {
                "match": {
                    "title": title
                }
            }
        }
        results = es.search(index='article', body=query)
        if results['hits']['total']['value'] > 0:
            return ArticleDocumentType(**results['hits']['hits'][0]['_source'])
 
# 创建 Schema
schema = Schema(query=Query)
 
# 添加 GraphQL 视图到 Flask 应用
app.add_url_rule('/graphql', view_func=GraphQLView.as_view('graphql', schema=schema, graphiql=True))
 
if __name__ == '__main__':
    app.run(debug=True)

这段代码首先导入了必要的模块，并初始化了一个 Flask 应用。然后，它创建了一个 Elasticsearch 实例并定义了一个与 Elasticsearch 文档类型相对应的 GraphQL 类型。接着，它定义了一个 GraphQL Query 类，其中包含一个根据文章标题搜索文章的字段。最后，它创建了一个 Schema 并将其添加到 Flask 应用中，使得我们可以通过 GraphQL 接口进行搜索。