made query.pipeline private

Author: daniel-sanche

google/cloud/firestore_v1/base_aggregation.py

@@ -48,6 +48,7 @@
     from google.cloud.firestore_v1.stream_generator import (
         StreamGenerator,
     )
+    from google.cloud.firestore_v1.pipeline_source import PipelineSource
 
     import datetime
 
@@ -356,19 +357,20 @@ def stream(
             A generator of the query results.
         """
 
-    def pipeline(self):
+    def _build_pipeline(self, source: "PipelineSource"):
         """
         Convert this query into a Pipeline
 
         Queries containing a `cursor` or `limit_to_last` are not currently supported
 
+        Args:
+            source: the PipelineSource to build the pipeline off of
         Raises:
-            - ValueError: raised if Query wasn't created with an associated client
             - NotImplementedError: raised if the query contains a `cursor` or `limit_to_last`
         Returns:
             a Pipeline representing the query
         """
         # use autoindexer to keep track of which field number to use for un-aliased fields
         autoindexer = itertools.count(start=1)
         exprs = [a._to_pipeline_expr(autoindexer) for a in self._aggregations]
-        return self._nested_query.pipeline().aggregate(*exprs)
+        return self._nested_query._build_pipeline(source).aggregate(*exprs)

google/cloud/firestore_v1/base_query.py

@@ -67,6 +67,7 @@
     from google.cloud.firestore_v1.query_profile import ExplainOptions
     from google.cloud.firestore_v1.query_results import QueryResultsList
     from google.cloud.firestore_v1.stream_generator import StreamGenerator
+    from google.cloud.firestore_v1.pipeline_source import PipelineSource
 
     import datetime
 
@@ -1129,24 +1130,23 @@ def recursive(self: QueryType) -> QueryType:
 
         return copied
 
-    def pipeline(self):
+    def _build_pipeline(self, source: "PipelineSource"):
         """
         Convert this query into a Pipeline
 
         Queries containing a `cursor` or `limit_to_last` are not currently supported
 
+        Args:
+            source: the PipelineSource to build the pipeline off of
         Raises:
-            - ValueError: raised if Query wasn't created with an associated client
             - NotImplementedError: raised if the query contains a `cursor` or `limit_to_last`
         Returns:
             a Pipeline representing the query
         """
-        if not self._client:
-            raise ValueError("Query does not have an associated client")
         if self._all_descendants:
-            ppl = self._client.pipeline().collection_group(self._parent.id)
+            ppl = source.collection_group(self._parent.id)
         else:
-            ppl = self._client.pipeline().collection(self._parent._path)
+            ppl = source.collection(self._parent._path)
 
         # Filters
         for filter_ in self._field_filters:

tests/unit/v1/test_aggregation.py

@@ -1040,7 +1040,7 @@ def test_aggreation_to_pipeline_sum(field, in_alias, out_alias):
     query = make_query(parent)
     aggregation_query = make_aggregation_query(query)
     aggregation_query.sum(field, alias=in_alias)
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     assert isinstance(pipeline, Pipeline)
     assert len(pipeline.stages) == 2
     assert isinstance(pipeline.stages[0], Collection)
@@ -1071,7 +1071,7 @@ def test_aggreation_to_pipeline_avg(field, in_alias, out_alias):
     query = make_query(parent)
     aggregation_query = make_aggregation_query(query)
     aggregation_query.avg(field, alias=in_alias)
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     assert isinstance(pipeline, Pipeline)
     assert len(pipeline.stages) == 2
     assert isinstance(pipeline.stages[0], Collection)
@@ -1102,7 +1102,7 @@ def test_aggreation_to_pipeline_count(in_alias, out_alias):
     query = make_query(parent)
     aggregation_query = make_aggregation_query(query)
     aggregation_query.count(alias=in_alias)
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     assert isinstance(pipeline, Pipeline)
     assert len(pipeline.stages) == 2
     assert isinstance(pipeline.stages[0], Collection)
@@ -1127,7 +1127,7 @@ def test_aggreation_to_pipeline_count_increment():
     aggregation_query = make_aggregation_query(query)
     for _ in range(n):
         aggregation_query.count()
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     aggregate_stage = pipeline.stages[1]
     assert len(aggregate_stage.accumulators) == n
     for i in range(n):
@@ -1146,7 +1146,7 @@ def test_aggreation_to_pipeline_complex():
     aggregation_query.count()
     aggregation_query.avg("other")
     aggregation_query.sum("final")
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     assert isinstance(pipeline, Pipeline)
     assert len(pipeline.stages) == 3
     assert isinstance(pipeline.stages[0], Collection)

tests/unit/v1/test_async_aggregation.py

@@ -716,7 +716,7 @@ def test_async_aggreation_to_pipeline_sum(field, in_alias, out_alias):
     query = make_async_query(parent)
     aggregation_query = make_async_aggregation_query(query)
     aggregation_query.sum(field, alias=in_alias)
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     assert isinstance(pipeline, AsyncPipeline)
     assert len(pipeline.stages) == 2
     assert isinstance(pipeline.stages[0], Collection)
@@ -747,7 +747,7 @@ def test_async_aggreation_to_pipeline_avg(field, in_alias, out_alias):
     query = make_async_query(parent)
     aggregation_query = make_async_aggregation_query(query)
     aggregation_query.avg(field, alias=in_alias)
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     assert isinstance(pipeline, AsyncPipeline)
     assert len(pipeline.stages) == 2
     assert isinstance(pipeline.stages[0], Collection)
@@ -778,7 +778,7 @@ def test_async_aggreation_to_pipeline_count(in_alias, out_alias):
     query = make_async_query(parent)
     aggregation_query = make_async_aggregation_query(query)
     aggregation_query.count(alias=in_alias)
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     assert isinstance(pipeline, AsyncPipeline)
     assert len(pipeline.stages) == 2
     assert isinstance(pipeline.stages[0], Collection)
@@ -803,7 +803,7 @@ def test_aggreation_to_pipeline_count_increment():
     aggregation_query = make_async_aggregation_query(query)
     for _ in range(n):
         aggregation_query.count()
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     aggregate_stage = pipeline.stages[1]
     assert len(aggregate_stage.accumulators) == n
     for i in range(n):
@@ -822,7 +822,7 @@ def test_async_aggreation_to_pipeline_complex():
     aggregation_query.count()
     aggregation_query.avg("other")
     aggregation_query.sum("final")
-    pipeline = aggregation_query.pipeline()
+    pipeline = aggregation_query._build_pipeline(client.pipeline())
     assert isinstance(pipeline, AsyncPipeline)
     assert len(pipeline.stages) == 3
     assert isinstance(pipeline.stages[0], Collection)

tests/unit/v1/test_async_query.py

@@ -917,7 +917,7 @@ def test_asyncquery_collection_pipeline_type():
     client = make_async_client()
     parent = client.collection("test")
     query = parent._query()
-    ppl = query.pipeline()
+    ppl = query._build_pipeline(client.pipeline())
     assert isinstance(ppl, AsyncPipeline)
 
 
@@ -926,5 +926,5 @@ def test_asyncquery_collectiongroup_pipeline_type():
 
     client = make_async_client()
     query = client.collection_group("test")
-    ppl = query.pipeline()
+    ppl = query._build_pipeline(client.pipeline())
     assert isinstance(ppl, AsyncPipeline)

tests/unit/v1/test_base_query.py

@@ -1994,18 +1994,10 @@ def test__collection_group_query_response_to_snapshot_response():
     assert snapshot.update_time == response_pb._pb.document.update_time
 
 
-def test__query_pipeline_no_client():
-    mock_parent = mock.Mock()
-    mock_parent._client = None
-    query = _make_base_query(mock_parent)
-    with pytest.raises(ValueError, match="client"):
-        query.pipeline()
-
-
 def test__query_pipeline_decendants():
     client = make_client()
     query = client.collection_group("my_col")
-    pipeline = query.pipeline()
+    pipeline = query._build_pipeline(client.pipeline())
 
     assert len(pipeline.stages) == 1
     stage = pipeline.stages[0]
@@ -2025,7 +2017,7 @@ def test__query_pipeline_no_decendants(in_path, out_path):
     client = make_client()
     collection = client.collection(in_path)
     query = collection._query()
-    pipeline = query.pipeline()
+    pipeline = query._build_pipeline(client.pipeline())
 
     assert len(pipeline.stages) == 1
     stage = pipeline.stages[0]
@@ -2043,7 +2035,7 @@ def test__query_pipeline_composite_filter():
     with mock.patch.object(
         expr.BooleanExpression, "_from_query_filter_pb"
     ) as convert_mock:
-        pipeline = query.pipeline()
+        pipeline = query._build_pipeline(client.pipeline())
         convert_mock.assert_called_once_with(in_filter._to_pb(), client)
         assert len(pipeline.stages) == 2
         stage = pipeline.stages[1]
@@ -2054,7 +2046,7 @@ def test__query_pipeline_composite_filter():
 def test__query_pipeline_projections():
     client = make_client()
     query = client.collection("my_col").select(["field_a", "field_b.c"])
-    pipeline = query.pipeline()
+    pipeline = query._build_pipeline(client.pipeline())
 
     assert len(pipeline.stages) == 2
     stage = pipeline.stages[1]
@@ -2069,7 +2061,7 @@ def test__query_pipeline_order_exists_multiple():
 
     client = make_client()
     query = client.collection("my_col").order_by("field_a").order_by("field_b")
-    pipeline = query.pipeline()
+    pipeline = query._build_pipeline(client.pipeline())
 
     # should have collection, where, and sort
     # we're interested in where
@@ -2089,7 +2081,7 @@ def test__query_pipeline_order_exists_multiple():
 def test__query_pipeline_order_exists_single():
     client = make_client()
     query_single = client.collection("my_col").order_by("field_c")
-    pipeline_single = query_single.pipeline()
+    pipeline_single = query_single._build_pipeline(client.pipeline())
 
     # should have collection, where, and sort
     # we're interested in where
@@ -2110,7 +2102,7 @@ def test__query_pipeline_order_sorts():
         .order_by("field_a", direction=BaseQuery.ASCENDING)
         .order_by("field_b", direction=BaseQuery.DESCENDING)
     )
-    pipeline = query.pipeline()
+    pipeline = query._build_pipeline(client.pipeline())
 
     assert len(pipeline.stages) == 3
     sort_stage = pipeline.stages[2]
@@ -2128,21 +2120,21 @@ def test__query_pipeline_unsupported():
     client = make_client()
     query_start = client.collection("my_col").start_at({"field_a": "value"})
     with pytest.raises(NotImplementedError, match="cursors"):
-        query_start.pipeline()
+        query_start._build_pipeline(client.pipeline())
 
     query_end = client.collection("my_col").end_at({"field_a": "value"})
     with pytest.raises(NotImplementedError, match="cursors"):
-        query_end.pipeline()
+        query_end._build_pipeline(client.pipeline())
 
     query_limit_last = client.collection("my_col").limit_to_last(10)
     with pytest.raises(NotImplementedError, match="limit_to_last"):
-        query_limit_last.pipeline()
+        query_limit_last._build_pipeline(client.pipeline())
 
 
 def test__query_pipeline_limit():
     client = make_client()
     query = client.collection("my_col").limit(15)
-    pipeline = query.pipeline()
+    pipeline = query._build_pipeline(client.pipeline())
 
     assert len(pipeline.stages) == 2
     stage = pipeline.stages[1]
@@ -2153,7 +2145,7 @@ def test__query_pipeline_limit():
 def test__query_pipeline_offset():
     client = make_client()
     query = client.collection("my_col").offset(5)
-    pipeline = query.pipeline()
+    pipeline = query._build_pipeline(client.pipeline())
 
     assert len(pipeline.stages) == 2
     stage = pipeline.stages[1]

tests/unit/v1/test_query.py

@@ -1054,7 +1054,7 @@ def test_asyncquery_collection_pipeline_type():
     client = make_client()
     parent = client.collection("test")
     query = parent._query()
-    ppl = query.pipeline()
+    ppl = query._build_pipeline(client.pipeline())
     assert isinstance(ppl, Pipeline)
 
 
@@ -1063,5 +1063,5 @@ def test_asyncquery_collectiongroup_pipeline_type():
 
     client = make_client()
     query = client.collection_group("test")
-    ppl = query.pipeline()
+    ppl = query._build_pipeline(client.pipeline())
     assert isinstance(ppl, Pipeline)