基于RavenDB Python客户端按DateTime范围筛选聚合结果

看起来你已经搞定了基础的Map/Reduce逻辑来聚合唯一Type类型和对应的计数，现在要加上DateTime范围的筛选，对吧？我来给你具体的实现方案，分两种场景来处理：

1. 静态时间过滤（在索引中预先设置条件） #

如果你的时间范围是固定的，可以直接在Map函数里添加过滤条件，这样索引只会处理符合时间要求的文档，能提升查询效率：

from ravendb import AbstractIndexCreationTask
from datetime import datetime

class TypeAggregationByDateIndex(AbstractIndexCreationTask):
    def __init__(self):
        super().__init__()
        # Map阶段：只处理Dates大于指定时间的文档
        self.map = """
            from doc in docs
            where doc.Dates > new DateTime(2020, 2, 9)
            select new {
                Type = doc.Type,
                Dates = new List<DateTime> { doc.Dates },
                TypeCount = 1
            }
        """
        # Reduce阶段：按Type分组，聚合Dates列表和计数
        self.reduce = """
            from result in results
            group result by result.Type into grouped
            select new {
                Type = grouped.Key,
                Dates = grouped.SelectMany(x => x.Dates).Distinct().ToList(),
                TypeCount = grouped.Sum(x => x.TypeCount)
            }
        """

创建这个索引后，查询指定Type（比如BTE）就能直接得到符合时间条件的结果。

2. 动态时间过滤（查询时传入时间参数） #

如果需要根据不同场景动态调整时间范围，更推荐在查询阶段添加过滤条件，这样同一个索引可以复用：

第一步：定义通用的Map/Reduce索引（不带时间过滤） #

class TypeAggregationIndex(AbstractIndexCreationTask):
    def __init__(self):
        super().__init__()
        self.map = """
            from doc in docs
            select new {
                Type = doc.Type,
                Dates = new List<DateTime> { doc.Dates },
                TypeCount = 1
            }
        """
        self.reduce = """
            from result in results
            group result by result.Type into grouped
            select new {
                Type = grouped.Key,
                Dates = grouped.SelectMany(x => x.Dates).Distinct().ToList(),
                TypeCount = grouped.Sum(x => x.TypeCount)
            }
        """

第二步：查询时添加时间范围条件 #

from ravendb import DocumentStore
from datetime import datetime

# 初始化文档存储
with DocumentStore(urls=["http://localhost:8080"], database="YourTargetDB") as store:
    store.initialize()
    
    with store.open_session() as session:
        # 设置目标时间（这里可以根据需求动态传入）
        target_date = datetime(2020, 2, 9)
        
        # 查询BTE类型，同时过滤Dates大于目标时间的结果
        aggregation_result = session.query_index("TypeAggregationIndex", dict) \
            .where_equals("Type", "BTE") \
            .where_greater_than("Dates", target_date) \
            .first_or_default()
        
        # 输出你想要的格式
        if aggregation_result:
            print(f"Type = {aggregation_result['Type']}, TypeCount = {aggregation_result['TypeCount']}, Dates = {aggregation_result['Dates']}")

小提示 #

• 确保你的文档中Dates字段是DateTime类型，而不是字符串，这样RavenDB才能正确进行时间比较；如果是字符串，需要先在Map阶段转换为DateTime对象。
• 如果涉及跨时区场景，建议使用DateTimeOffset类型来避免时间歧义。

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