Elasticsearch-dsl，带有嵌套过滤器和完全匹配的AND和OR条件

Question

三个参数来自前端：

1. State- 字符串
2. Categories- 字符串数组。字符串可以由几个单词组成。
3. Tags- 类似于类别。

所有参数都是可选的。

如果传输了几个，则需要通过AND来实现它们的捆绑包(以及state的巧合，category和tag(。如果提交了多个categories或tags，则至少会匹配其中一个。

也就是说，如果请求到达时带有参数

{"state": "Alaska", "categories": ["category 1", "category 2"]}

答案将是

• state = Alaska，categories = category 1;
• state = Alaska, categories = category 2;
• state = Alaska, categories = [category 1, category 2];
• state = Alaska, categories = [category 1, category 3](至少具有请求的类别之一(。

不适合

• state = Alabama, categories = category 1
• state = Alaska, categories = 3
• state = Alaska, categories = 1 category(类别名称应为 1 合 1"category 1" != "1 category"(

为了elastikserch，我从python(3.7(发送请求。拿了一个图书馆 Elasticsearch-DSL

通过Q对象(在其中match使用(收集了三个过滤器。

combined_filter = state_filter & categories_filter & tags_filter

categories和tags列表分为subfilters through OR。

query = queries.pop()
for item in queries:
query |= item

这样的请求是为elasticsearch创建的。

Bool(minimum_should_match=1, 
must=[Match(state='Alaska'), MatchAll()], 
should=[Match(categories='category 1'), Match(categories='category 2')]
)

为什么此逻辑通过不精确的category/tag名称查找条目？

from typing import List
from elasticsearch import Elasticsearch
from elasticsearch_dsl import Q, Search
from flask import request
from flask.views import MethodView

es = Elasticsearch()

class ArticleSearchAPIView(MethodView):
"""
Search articles using ElasticSearch
"""
@staticmethod
def filter_create(queries: List[Q]) -> Q:
"""
Creates Q.OR filter
"""
query = queries.pop()
for item in queries:
query |= item
return query
def get(self) -> dict:
"""
Search article
First request - with empty params
"""
search = Search(using=es, index=ArticleModel.__tablename__)
state_filter = categories_filter = tags_filter = Q()
result = "Articles not found."
data = request.get_json()
categories = data.get("categories")
tags = data.get("tags")
state = data.get("state")
if state:
state_filter = Q("match", state=state)
if categories:
queries = [Q("match", categories=value) for value in categories]
categories_filter = self.filter_create(queries)
if tags:
queries = [Q("match", tags=value) for value in tags]
tags_filter = self.filter_create(queries)
combined_filter = state_filter & categories_filter & tags_filter
found = (
search.filter(combined_filter)
.execute()
.to_dict()["hits"]
.get("hits")
)
if found:
result = [article["_source"] for article in found]
return {"response": result}

更新

---

Article and Category与Article and Tag的关系 -MTM

映射

{
"articles": {
"mappings": {
"properties": {
...
"categories": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"state": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"tags": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
}
...
}
}
}
}

Answer 1

您可以使用布尔查询。

在布尔查询中 弹性搜索布尔查询

你有"必须"，它等同于"AND"运算符。和"应该"，充当"OR"运算符。

{
"query": {
"bool" : {
"must" : {
"term" : { "user" : "kimchy" }
},
"should" : [
{ "term" : { "tag" : "wow" } },
{ "term" : { "tag" : "elasticsearch" } }
],
}
}
}

Answer 2

我决定在这里使用elasticsearch-dsl是不必要的。

这是我做出的决定。

from typing import Dict, List, Tuple, Union
from elasticsearch import Elasticsearch
from flask import request
from flask.views import MethodView
from .models import AticleModel  # ArticleModel.__tablename__ == "articles"

es = Elasticsearch()

class ArticleSearchAPIView(MethodView):
"""
Search articles using ElasticSearch
"""
def get(
self,
) -> Union[
Dict[str, Union[list, List[str]]],
Tuple[Dict[str, str], int],
Dict[str, Union[list, str]],
]:
"""
Search articles
"""
data = request.get_json()
categories = data.get("categories")
tags = data.get("tags")
state = data.get("state")
result = "Articles not found."
query = {"bool": {"must": []}}
if state:
query["bool"]["must"].append({"term": {"state.keyword": state}})
if categories:
query["bool"]["must"].append(
{"terms": {"categories.keyword": categories}}
)
if tags:
query["bool"]["must"].append({"terms": {"tags.keyword": tags}})
found = es.search(
index=ArticleModel.__tablename__, body={"query": query},
)["hits"].get("hits")
if found:
result = [article["_source"] for article in found]
return {"response": result}