feat: add pipelines structure

google/cloud/firestore_v1/_helpers.py

@@ -120,6 +120,9 @@ def __ne__(self, other):
         else:
             return not equality_val
 
+    def __repr__(self):
+        return f"{type(self).__name__}(latitude={self.latitude}, longitude={self.longitude})"
+
 
 def verify_path(path, is_collection) -> None:
     """Verifies that a ``path`` has the correct form.

google/cloud/firestore_v1/async_client.py

@@ -47,6 +47,8 @@
 from google.cloud.firestore_v1.services.firestore.transports import (
     grpc_asyncio as firestore_grpc_transport,
 )
+from google.cloud.firestore_v1.async_pipeline import AsyncPipeline
+from google.cloud.firestore_v1.pipeline_source import PipelineSource
 
 if TYPE_CHECKING:  # pragma: NO COVER
     import datetime
@@ -427,3 +429,10 @@ def transaction(self, **kwargs) -> AsyncTransaction:
             A transaction attached to this client.
         """
         return AsyncTransaction(self, **kwargs)
+
+    @property
+    def _pipeline_cls(self):
+        return AsyncPipeline
+
+    def pipeline(self) -> PipelineSource:
+        return PipelineSource(self)

google/cloud/firestore_v1/async_pipeline.py

@@ -0,0 +1,79 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+from typing import AsyncIterable, TYPE_CHECKING
+from google.cloud.firestore_v1 import pipeline_stages as stages
+from google.cloud.firestore_v1.base_pipeline import _BasePipeline
+
+if TYPE_CHECKING:  # pragma: NO COVER
+    from google.cloud.firestore_v1.async_client import AsyncClient
+    from google.cloud.firestore_v1.pipeline_result import PipelineResult
+    from google.cloud.firestore_v1.async_transaction import AsyncTransaction
+
+
+class AsyncPipeline(_BasePipeline):
+    """
+    Pipelines allow for complex data transformations and queries involving
+    multiple stages like filtering, projection, aggregation, and vector search.
+
+    This class extends `_BasePipeline` and provides methods to execute the
+    defined pipeline stages using an asynchronous `AsyncClient`.
+
+    Usage Example:
+        >>> from google.cloud.firestore_v1.pipeline_expressions import Field
+        >>>
+        >>> async def run_pipeline():
+        ...     client = AsyncClient(...)
+        ...     pipeline = client.pipeline()
+        ...                      .collection("books")
+        ...                      .where(Field.of("published").gt(1980))
+        ...                      .select("title", "author")
+        ...     async for result in pipeline.execute_async():
+        ...         print(result)
+
+    Use `client.pipeline()` to create instances of this class.
+    """
+
+    def __init__(self, client: AsyncClient, *stages: stages.Stage):
+        """
+        Initializes an asynchronous Pipeline.
+
+        Args:
+            client: The asynchronous `AsyncClient` instance to use for execution.
+            *stages: Initial stages for the pipeline.
+        """
+        super().__init__(client, *stages)
+
+    async def execute(
+        self,
+        transaction: "AsyncTransaction" | None = None,
+    ) -> AsyncIterable[PipelineResult]:
+        """
+        Executes this pipeline, providing results through an Iterable
+
+        Args:
+            transaction
+                (Optional[:class:`~google.cloud.firestore_v1.transaction.Transaction`]):
+                An existing transaction that this query will run in.
+                If a ``transaction`` is used and it already has write operations
+                added, this method cannot be used (i.e. read-after-write is not
+                allowed).
+        """
+        request = self._prep_execute_request(transaction)
+        async for response in await self._client._firestore_api.execute_pipeline(
+            request
+        ):
+            for result in self._execute_response_helper(response):
+                yield result

google/cloud/firestore_v1/base_client.py

@@ -37,6 +37,7 @@
     Optional,
     Tuple,
     Union,
+    Type,
 )
 
 import google.api_core.client_options
@@ -61,6 +62,8 @@
 from google.cloud.firestore_v1.bulk_writer import BulkWriter, BulkWriterOptions
 from google.cloud.firestore_v1.field_path import render_field_path
 from google.cloud.firestore_v1.services.firestore import client as firestore_client
+from google.cloud.firestore_v1.pipeline_source import PipelineSource
+from google.cloud.firestore_v1.base_pipeline import _BasePipeline
 
 DEFAULT_DATABASE = "(default)"
 """str: The default database used in a :class:`~google.cloud.firestore_v1.client.Client`."""
@@ -500,6 +503,20 @@ def batch(self) -> BaseWriteBatch:
     def transaction(self, **kwargs) -> BaseTransaction:
         raise NotImplementedError
 
+    def pipeline(self) -> PipelineSource:
+        """
+        Start a pipeline with this client.
+
+        Returns:
+            :class:`~google.cloud.firestore_v1.pipeline_source.PipelineSource`:
+            A pipeline that uses this client`
+        """
+        raise NotImplementedError
+
+    @property
+    def _pipeline_cls(self) -> Type["_BasePipeline"]:
+        raise NotImplementedError
+
 
 def _reference_info(references: list) -> Tuple[list, dict]:
     """Get information about document references.

google/cloud/firestore_v1/base_pipeline.py

@@ -0,0 +1,110 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+from typing import Iterable, TYPE_CHECKING
+from google.cloud.firestore_v1 import pipeline_stages as stages
+from google.cloud.firestore_v1.types.pipeline import (
+    StructuredPipeline as StructuredPipeline_pb,
+)
+from google.cloud.firestore_v1.types.firestore import ExecutePipelineRequest
+from google.cloud.firestore_v1.pipeline_result import PipelineResult
+from google.cloud.firestore_v1 import _helpers
+
+if TYPE_CHECKING:  # pragma: NO COVER
+    from google.cloud.firestore_v1.client import Client
+    from google.cloud.firestore_v1.async_client import AsyncClient
+    from google.cloud.firestore_v1.types.firestore import ExecutePipelineResponse
+    from google.cloud.firestore_v1.transaction import BaseTransaction
+
+
+class _BasePipeline:
+    """
+    Base class for building Firestore data transformation and query pipelines.
+
+    This class is not intended to be instantiated directly.
+    Use `client.collection.("...").pipeline()` to create pipeline instances.
+    """
+
+    def __init__(self, client: Client | AsyncClient, *stages: stages.Stage):
+        """
+        Initializes a new pipeline with the given stages.
+
+        Pipeline classes should not be instantiated directly.
+
+        Args:
+            client: The client associated with the pipeline
+            *stages: Initial stages for the pipeline.
+        """
+        self._client = client
+        self.stages = tuple(stages)
+
+    def __repr__(self):
+        cls_str = type(self).__name__
+        if not self.stages:
+            return f"{cls_str}()"
+        elif len(self.stages) == 1:
+            return f"{cls_str}({self.stages[0]!r})"
+        else:
+            stages_str = ",\n  ".join([repr(s) for s in self.stages])
+            return f"{cls_str}(\n  {stages_str}\n)"
+
+    def _to_pb(self) -> StructuredPipeline_pb:
+        return StructuredPipeline_pb(
+            pipeline={"stages": [s._to_pb() for s in self.stages]}
+        )
+
+    def _append(self, new_stage):
+        """
+        Create a new Pipeline object with a new stage appended
+        """
+        return self.__class__(self._client, *self.stages, new_stage)
+
+    def _prep_execute_request(
+        self, transaction: BaseTransaction | None
+    ) -> ExecutePipelineRequest:
+        """
+        shared logic for creating an ExecutePipelineRequest
+        """
+        database_name = (
+            f"projects/{self._client.project}/databases/{self._client._database}"
+        )
+        transaction_id = (
+            _helpers.get_transaction_id(transaction)
+            if transaction is not None
+            else None
+        )
+        request = ExecutePipelineRequest(
+            database=database_name,
+            transaction=transaction_id,
+            structured_pipeline=self._to_pb(),
+        )
+        return request
+
+    def _execute_response_helper(
+        self, response: ExecutePipelineResponse
+    ) -> Iterable[PipelineResult]:
+        """
+        shared logic for unpacking an ExecutePipelineReponse into PipelineResults
+        """
+        for doc in response.results:
+            ref = self._client.document(doc.name) if doc.name else None
+            yield PipelineResult(
+                self._client,
+                doc.fields,
+                ref,
+                response._pb.execution_time,
+                doc._pb.create_time if doc.create_time else None,
+                doc._pb.update_time if doc.update_time else None,
+            )

google/cloud/firestore_v1/client.py

@@ -49,6 +49,8 @@
     grpc as firestore_grpc_transport,
 )
 from google.cloud.firestore_v1.transaction import Transaction
+from google.cloud.firestore_v1.pipeline import Pipeline
+from google.cloud.firestore_v1.pipeline_source import PipelineSource
 
 if TYPE_CHECKING:  # pragma: NO COVER
     from google.cloud.firestore_v1.bulk_writer import BulkWriter
@@ -408,3 +410,10 @@ def transaction(self, **kwargs) -> Transaction:
             A transaction attached to this client.
         """
         return Transaction(self, **kwargs)
+
+    @property
+    def _pipeline_cls(self):
+        return Pipeline
+
+    def pipeline(self) -> PipelineSource:
+        return PipelineSource(self)

google/cloud/firestore_v1/field_path.py

@@ -16,7 +16,7 @@
 from __future__ import annotations
 import re
 from collections import abc
-from typing import Iterable, cast
+from typing import Any, Iterable, cast, MutableMapping
 
 _FIELD_PATH_MISSING_TOP = "{!r} is not contained in the data"
 _FIELD_PATH_MISSING_KEY = "{!r} is not contained in the data for the key {!r}"
@@ -170,7 +170,7 @@ def render_field_path(field_names: Iterable[str]):
 get_field_path = render_field_path  # backward-compatibility
 
 
-def get_nested_value(field_path: str, data: dict):
+def get_nested_value(field_path: str, data: MutableMapping[str, Any]):
     """Get a (potentially nested) value from a dictionary.
 
     If the data is nested, for example:

google/cloud/firestore_v1/pipeline.py

@@ -0,0 +1,73 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+from typing import Iterable, TYPE_CHECKING
+from google.cloud.firestore_v1 import pipeline_stages as stages
+from google.cloud.firestore_v1.base_pipeline import _BasePipeline
+
+if TYPE_CHECKING:  # pragma: NO COVER
+    from google.cloud.firestore_v1.client import Client
+    from google.cloud.firestore_v1.pipeline_result import PipelineResult
+    from google.cloud.firestore_v1.transaction import Transaction
+
+
+class Pipeline(_BasePipeline):
+    """
+    Pipelines allow for complex data transformations and queries involving
+    multiple stages like filtering, projection, aggregation, and vector search.
+
+    Usage Example:
+        >>> from google.cloud.firestore_v1.pipeline_expressions import Field
+        >>>
+        >>> def run_pipeline():
+        ...     client = Client(...)
+        ...     pipeline = client.pipeline()
+        ...                      .collection("books")
+        ...                      .where(Field.of("published").gt(1980))
+        ...                      .select("title", "author")
+        ...     for result in pipeline.execute():
+        ...         print(result)
+
+    Use `client.pipeline()` to create instances of this class.
+    """
+
+    def __init__(self, client: Client, *stages: stages.Stage):
+        """
+        Initializes a Pipeline.
+
+        Args:
+            client: The `Client` instance to use for execution.
+            *stages: Initial stages for the pipeline.
+        """
+        super().__init__(client, *stages)
+
+    def execute(
+        self,
+        transaction: "Transaction" | None = None,
+    ) -> Iterable[PipelineResult]:
+        """
+        Executes this pipeline, providing results through an Iterable
+
+        Args:
+            transaction
+                (Optional[:class:`~google.cloud.firestore_v1.transaction.Transaction`]):
+                An existing transaction that this query will run in.
+                If a ``transaction`` is used and it already has write operations
+                added, this method cannot be used (i.e. read-after-write is not
+                allowed).
+        """
+        request = self._prep_execute_request(transaction)
+        for response in self._client._firestore_api.execute_pipeline(request):
+            yield from self._execute_response_helper(response)