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/_pipeline_stages.py

@@ -0,0 +1,81 @@
+# 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 Optional
+from abc import ABC
+from abc import abstractmethod
+
+from google.cloud.firestore_v1.types.document import Pipeline as Pipeline_pb
+from google.cloud.firestore_v1.types.document import Value
+from google.cloud.firestore_v1.pipeline_expressions import Expr
+
+
+class Stage(ABC):
+    """Base class for all pipeline stages.
+
+    Each stage represents a specific operation (e.g., filtering, sorting,
+    transforming) within a Firestore pipeline. Subclasses define the specific
+    arguments and behavior for each operation.
+    """
+
+    def __init__(self, custom_name: Optional[str] = None):
+        self.name = custom_name or type(self).__name__.lower()
+
+    def _to_pb(self) -> Pipeline_pb.Stage:
+        return Pipeline_pb.Stage(
+            name=self.name, args=self._pb_args(), options=self._pb_options()
+        )
+
+    @abstractmethod
+    def _pb_args(self) -> list[Value]:
+        """Return Ordered list of arguments the given stage expects"""
+        raise NotImplementedError
+
+    def _pb_options(self) -> dict[str, Value]:
+        """Return optional named arguments that certain functions may support."""
+        return {}
+
+    def __repr__(self):
+        items = ("%s=%r" % (k, v) for k, v in self.__dict__.items() if k != "name")
+        return f"{self.__class__.__name__}({', '.join(items)})"
+
+
+class Collection(Stage):
+    """Specifies a collection as the initial data source."""
+
+    def __init__(self, path: str):
+        super().__init__()
+        if not path.startswith("/"):
+            path = f"/{path}"
+        self.path = path
+
+    def _pb_args(self):
+        return [Value(reference_value=self.path)]
+
+
+class GenericStage(Stage):
+    """Represents a generic, named stage with parameters."""
+
+    def __init__(self, name: str, *params: Expr | Value):
+        super().__init__(name)
+        self.params: list[Value] = [
+            p._to_pb() if isinstance(p, Expr) else p for p in params
+        ]
+
+    def _pb_args(self):
+        return self.params
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(name='{self.name}')"

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,96 @@
+# 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():
+        ...         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,
+    ) -> list[PipelineResult]:
+        """
+        Executes this pipeline and returns results as a list
+
+        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).
+        """
+        return [result async for result in self.stream(transaction=transaction)]
+
+    async def stream(
+        self,
+        transaction: "AsyncTransaction" | None = None,
+    ) -> AsyncIterable[PipelineResult]:
+        """
+        Process this pipeline as a stream, 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,151 @@
+# 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, Sequence, 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.pipeline_expressions import Expr
+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):
+        """
+        Initializes a new pipeline.
+
+        Pipelines should not be instantiated directly. Instead,
+        call client.pipeline() to create an instance
+
+        Args:
+            client: The client associated with the pipeline
+        """
+        self._client = client
+        self.stages: Sequence[stages.Stage] = tuple()
+
+    @classmethod
+    def _create_with_stages(
+        cls, client: Client | AsyncClient, *stages
+    ) -> _BasePipeline:
+        """
+        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.
+        """
+        new_instance = cls(client)
+        new_instance.stages = tuple(stages)
+        return new_instance
+
+    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__._create_with_stages(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,
+            )
+
+    def generic_stage(self, name: str, *params: Expr) -> "_BasePipeline":
+        """
+        Adds a generic, named stage to the pipeline with specified parameters.
+
+        This method provides a flexible way to extend the pipeline's functionality
+        by adding custom stages. Each generic stage is defined by a unique `name`
+        and a set of `params` that control its behavior.
+
+        Example:
+            >>> # Assume we don't have a built-in "where" stage
+            >>> pipeline = client.pipeline().collection("books")
+            >>> pipeline = pipeline.generic_stage("where", [Field.of("published").lt(900)])
+            >>> pipeline = pipeline.select("title", "author")
+
+        Args:
+            name: The name of the generic stage.
+            *params: A sequence of `Expr` objects representing the parameters for the stage.
+
+        Returns:
+            A new Pipeline object with this stage appended to the stage list
+        """
+        return self._append(stages.GenericStage(name, *params))