def convert_query(query):
"""
Convert Elasticsearch query to use keyword and text fields appropriately
"""
if isinstance(query, dict):
for key, value in query.items():
if key in ["term", "terms", 'range']:
new_value = dict()
for k, v in value.items():
if not k.endswith(".keyword"):
k = f"{k}.keyword"
new_value.update({k: v})
query[key] = new_value
elif key in ["match", "match_phrase", "match_phrase_prefix"]:
new_value = dict()
for k, v in value.items():
if k.endswith(".keyword"):
k = k[:-len(".keyword")]
new_value.update({k: v})
query[key] = new_value
elif key == "multi_match":
new_fields = list()
fields = query[key].get("fields", [])
for field in fields:
if field.endswith(".keyword"):
field = field[:-len(".keyword")]
new_fields.append(field)
query[key] = {
"query": query[key].get("query"),
"fields": new_fields
}
elif key == "exists":
field = query[key].get("field")
query[key] = {"field": field if field.endswith(".keyword") else f"{field}.keyword"}
else:
query[key] = convert_query(value)
elif isinstance(query, list):
query = [convert_query(q) for q in query]
return query
query = {
"query": {
"bool": {
"must": [
{
"term": {
"my_field": "some_value"
}
},
{
"terms": {
"my_field1": ["some_value"]
}
},
{
"match": {
"my_text_field.keyword": "some text to match"
}
},
{"match_phrase": {"my_text_field.keyword": "some text to match"}},
{"match_phrase_prefix": {"my_text_field.keyword": "some text to"}},
{
"multi_match": {
"query": "full text search",
"fields": ["title.keyword", "body.keyword"]
}
},
{
"range": {
"age": {
"gte": 20,
"lt": 30
}
}
},
{"exists": {"field": "title"}}
],
"filter": {
"term": {
"my_other_field": "other_value"
}
}
}
}
}
converted_query = convert_query(query)
print(converted_query)
