Moving to aggregates

flowmachine/flowmachine/core/server/query_schemas/per_subscriber_aggregate.py

@@ -1,6 +1,7 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
+from functools import reduce
 
 from marshmallow import fields
 from marshmallow.validate import OneOf
@@ -15,8 +16,8 @@
 
 
 class PerSubscriberAggregateExposed(BaseExposedQuery):
-    def __init__(self, subscriber_query, agg_method):
-        self.subscriber_query = subscriber_query
+    def __init__(self, subscriber_queries, agg_method):
+        self.subscriber_query = reduce(lambda x, y: x.join(y), subscriber_queries)
         self.agg_method = agg_method
 
     def _flomachine_query_obj(self):
@@ -29,7 +30,8 @@ def _flomachine_query_obj(self):
 
 class PerSubscriberAggregateSchema(BaseSchema):
     query_kind = fields.String(validate=OneOf(["per_subscriber_aggregate"]))
-    subscriber_query = fields.Nested(SubscriberSubset)
+    # TODO: Can we make this a list or a single query?
+    subscriber_query = fields.List(SubscriberSubset)
     agg_method = fields.String(validate=OneOf(agg_methods))
 
     __model__ = PerSubscriberAggregateExposed