基于Python 3.x Elasticsearch嵌套字段模型，如何编写DSL查询语句？

针对你在Python 3.x环境下，使用elasticsearch==6.1.1和elasticsearch-dsl==6.1.0处理嵌套字段的需求，结合你给出的VendCompanyData模型结构，我整理了几种常用场景的查询实现，直接就能复用：

前置环境与模型回顾 #

先确认你的模型定义（方便对照查询逻辑）：

from elasticsearch_dsl import DocType, Nested, Integer, Date
from datetime import datetime

class VendCompanyData(DocType):
    data = Nested(
        properties={
            "machine_id": Integer(required=True),
            "product_id": Integer(required=True),
            "quantity": Integer(),
            "insert_time": Date()
        }
    )
    last_export = Date()
    created_at = Date()

    def save(self, **kwargs):
        self.created_at = datetime.now().strftime('%Y-%m-%dT%H:%M:%S.%fZ')
        return super().save(**kwargs)

1. 精确匹配单个嵌套字段属性 #

比如查询所有包含machine_id=1229的文档，必须使用Nested查询（否则会出现跨嵌套对象的错误匹配）：

from elasticsearch_dsl import Search, Q

# 初始化Search对象，替换为你的实际索引名
s = Search(index="vend_company_data")

# 构建嵌套查询：限定在data字段下匹配machine_id=1229
nested_query = Q(
    "nested",
    path="data",
    query=Q("match", data__machine_id=1229)
)

s = s.query(nested_query)

# 执行查询并遍历结果
response = s.execute()
for hit in response:
    print(f"文档ID: {hit.meta.id}, 嵌套数据: {hit.data}")

2. 多条件组合的嵌套字段查询 #

如果要匹配同一个嵌套对象同时满足machine_id=1229和product_id=567的文档，用布尔查询组合条件：

from elasticsearch_dsl import Search, Q

s = Search(index="vend_company_data")

# 嵌套内部的多条件组合：必须同时满足两个条件
inner_query = Q("bool", must=[
    Q("match", data__machine_id=1229),
    Q("match", data__product_id=567)
])

nested_query = Q("nested", path="data", query=inner_query)
s = s.query(nested_query)

response = s.execute()

3. 嵌套字段的范围查询 #

针对数值型（如quantity）或日期型（如insert_time）字段做范围筛选：

from elasticsearch_dsl import Search, Q
from datetime import datetime

s = Search(index="vend_company_data")

inner_query = Q("bool", must=[
    # 匹配quantity大于10的嵌套对象
    Q("range", data__quantity={"gt": 10}),
    # 匹配insert_time在2024年至今的嵌套对象
    Q("range", data__insert_time={
        "gte": "2024-01-01T00:00:00Z",
        "lte": datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ')
    })
])

nested_query = Q("nested", path="data", query=inner_query)
s = s.query(nested_query)
response = s.execute()

4. 仅返回匹配的嵌套对象（过滤冗余数据） #

默认查询会返回文档的所有嵌套对象，若只想获取符合条件的嵌套项，启用inner_hits参数：

from elasticsearch_dsl import Search, Q

s = Search(index="vend_company_data")

nested_query = Q(
    "nested",
    path="data",
    query=Q("match", data__machine_id=1229),
    inner_hits={}  # 开启后仅返回匹配的嵌套子对象
)

s = s.query(nested_query)
response = s.execute()

for hit in response:
    print(f"文档ID: {hit.meta.id}")
    # 提取匹配的嵌套对象
    matched_items = hit.meta.inner_hits.data.hits.hits
    for item in matched_items:
        print(f"匹配的嵌套项: {item['_source']}")

5. 结合根字段与嵌套字段的混合查询 #

比如同时筛选根字段last_export（非嵌套）和嵌套字段product_id的条件：

from elasticsearch_dsl import Search, Q
from datetime import datetime, timedelta

s = Search(index="vend_company_data")

# 根字段条件：last_export早于30天前
root_condition = Q("range", last_export={"lte": (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%dT%H:%M:%SZ')})

# 嵌套字段条件：product_id=567
nested_condition = Q("nested", path="data", query=Q("match", data__product_id=567))

# 组合两个条件：必须同时满足
s = s.query(Q("bool", must=[root_condition, nested_condition]))
response = s.execute()

内容的提问来源于stack exchange，提问作者Marin