from elasticsearch7 import Elasticsearch


class ElasticsearchTool():
    def __init__(self,index):
    #     初始化es对象
     self.es = Elasticsearch(
        # ['http://101.132.32.61:9200/'],
         ['172.17.0.7:9200/'],
         sniff_on_connection_fail=True,  # 节点没有响应时，进行刷新，重新连接
        sniffer_timeout=60,  # 每 60 秒刷新一次
         )
     self.index = index  #索引实例化传递的索引名

     self.create_index()

    #创建索引
    def create_index(self):
        #判断索引是否已经存在
        if not self.es.indices.exists(index=self.index):
         self.es.indices.create(index=self.index)
    #向索引中插入数据
    def add_data(self,id,data):
        return self.es.index(index=self.index,id=id,document=data)

    #搜索数据
    def search(self,body):
        return self.es.search(index=self.index,body=body)



我们来测试一下封装是否有问题：



if __name__ == '__main__':

    # 插入数据（进行测试）
    # es_tool =ElasticsearchTool('my_book3')
    # re1 = es_tool.add_data('1',{
    #     'title':'三国演义',
    #     'author':'罗贯中'
    # })
    # print(re1)



# 查询数据（进行测试）
    es_tool = ElasticsearchTool('mybook')



    res = es_tool.search(

        {
            "query": {
                "bool": {
                    "should": [
                        {
                            "match": {
                                "title": '恐龙',
                            }
                        },
                        {
                            "match": {
                                "price": '66'
                            }
                        }
                    ]
                }
            }
        }
    )




    print(res)




//进行监听数据；

 # 监听模型类

导入我们封装的类

from utils.elasticsear_settig import ElasticsearchTool

# # 在插入数据后进行监听
@event.listens_for(ArticleModel,'after_insert',raw=True)
@event.listens_for(ArticleModel,'after_update',raw=True)
def listen_book_model(mapper,connection,target):



    data = {
        'aid':target.object.aid,
        'title':target.object.title,
        'audio_file':target.object.audio_file,
        'is_vip':target.object.is_vip,
        'count':target.object.count,
    }
    es = ElasticsearchTool('myarticle')
    es.add_data(target.object.aid,data)





// 在视图中实现搜索功能：

class SearchView(BaseView):
    def get(self,word):

        from utils.elasticsear_settig import ElasticsearchTool
        es =  ElasticsearchTool('mybook')
        rs = es.search(

        #     '''
        #     'bid':target.object.bid,
        # 'title':target.object.title,
        # 'img':target.object.img,
        # 'author':target.object.author,
        # 'read_num':target.object.read_num,
        # 'intro':target.object.intro,
        # 'is_hot':target.object.is_hot,
        #
        #
        #     '''


            {
                "query": {
                    "bool": {
                        "should": [
                            {
                                "match": {
                                    "title": word,
                                }
                            },

                            {
                                "match": {
                                    "intro": word,
                                }
                            },

                            {
                                "match": {
                                    "price": word
                                }
                            },

                            {
                                "match": {
                                    "img": word,
                                }
                            },
                            {
                                "match": {
                                    "author": word,
                                }
                            },
                        ]
                    }
                }
            }
        )
        book_list = rs['hits']['hits']
        book_all = []
        for book in book_list:
            book_all.append(book['_source'])
            return self.success(book_all)