class BundledDocumentMetadata(proto.Message):
    r"""Metadata describing a Firestore document saved in the bundle.

    Attributes:
        name (str):
            The document key of a bundled document.
        read_time (google.protobuf.timestamp_pb2.Timestamp):
            The snapshot version of the document data
            bundled.
        exists (bool):
            Whether the document exists.
        queries (Sequence[str]):
            The names of the queries in this bundle that
            this document matches to.
    """

    name = proto.Field(proto.STRING, number=1,)
    read_time = proto.Field(proto.MESSAGE, number=2, message=timestamp_pb2.Timestamp,)
    exists = proto.Field(proto.BOOL, number=3,)
    queries = proto.RepeatedField(proto.STRING, number=4,) class BundledQuery(proto.Message):
    r"""Encodes a query saved in the bundle.

    .. _oneof: https://proto-plus-python.readthedocs.io/en/stable/fields.html#oneofs-mutually-exclusive-fields

    Attributes:
        parent (str):
            The parent resource name.
        structured_query (google.firestore.v1.query_pb2.StructuredQuery):
            A structured query.

            This field is a member of `oneof`_ ``query_type``.
        limit_type (google.cloud.bundle.types.BundledQuery.LimitType):

    """

    class LimitType(proto.Enum):
        r"""If the query is a limit query, should the limit be applied to
        the beginning or the end of results.
        """
        FIRST = 0
        LAST = 1

    parent = proto.Field(proto.STRING, number=1,)
    structured_query = proto.Field(
        proto.MESSAGE, number=2, oneof="query_type", message=query_pb2.StructuredQuery,
    )
    limit_type = proto.Field(proto.ENUM, number=3, enum=LimitType,) class BundleElement(proto.Message):
    r"""A Firestore bundle is a length-prefixed stream of JSON
    representations of ``BundleElement``. Only one ``BundleMetadata`` is
    expected, and it should be the first element. The named queries
    follow after ``metadata``. Every ``document_metadata`` is
    immediately followed by a ``document``.

    This message has `oneof`_ fields (mutually exclusive fields).
    For each oneof, at most one member field can be set at the same time.
    Setting any member of the oneof automatically clears all other
    members.

    .. _oneof: https://proto-plus-python.readthedocs.io/en/stable/fields.html#oneofs-mutually-exclusive-fields

    Attributes:
        metadata (google.cloud.bundle.types.BundleMetadata):

            This field is a member of `oneof`_ ``element_type``.
        named_query (google.cloud.bundle.types.NamedQuery):

            This field is a member of `oneof`_ ``element_type``.
        document_metadata (google.cloud.bundle.types.BundledDocumentMetadata):

            This field is a member of `oneof`_ ``element_type``.
        document (google.firestore.v1.document_pb2.Document):

            This field is a member of `oneof`_ ``element_type``.
    """

    metadata = proto.Field(
        proto.MESSAGE, number=1, oneof="element_type", message="BundleMetadata",
    )
    named_query = proto.Field(
        proto.MESSAGE, number=2, oneof="element_type", message="NamedQuery",
    )
    document_metadata = proto.Field(
        proto.MESSAGE,
        number=3,
        oneof="element_type",
        message="BundledDocumentMetadata",
    )
    document = proto.Field(
        proto.MESSAGE, number=4, oneof="element_type", message=document_pb2.Document,
    ) class BundleMetadata(proto.Message):
    r"""Metadata describing the bundle file/stream.

    Attributes:
        id (str):
            The ID of the bundle.
        create_time (google.protobuf.timestamp_pb2.Timestamp):
            Time at which the documents snapshot is taken
            for this bundle.
        version (int):
            The schema version of the bundle.
        total_documents (int):
            The number of documents in the bundle.
        total_bytes (int):
            The size of the bundle in bytes, excluding this
            ``BundleMetadata``.
    """

    id = proto.Field(proto.STRING, number=1,)
    create_time = proto.Field(proto.MESSAGE, number=2, message=timestamp_pb2.Timestamp,)
    version = proto.Field(proto.UINT32, number=3,)
    total_documents = proto.Field(proto.UINT32, number=4,)
    total_bytes = proto.Field(proto.UINT64, number=5,) class NamedQuery(proto.Message):
    r"""A Query associated with a name, created as part of the bundle
    file, and can be read by client SDKs once the bundle containing
    them is loaded.

    Attributes:
        name (str):
            Name of the query, such that client can use
            the name to load this query from bundle, and
            resume from when the query results are
            materialized into this bundle.
        bundled_query (google.cloud.bundle.types.BundledQuery):
            The query saved in the bundle.
        read_time (google.protobuf.timestamp_pb2.Timestamp):
            The read time of the query, when it is used
            to build the bundle. This is useful to resume
            the query from the bundle, once it is loaded by
            client SDKs.
    """

    name = proto.Field(proto.STRING, number=1,)
    bundled_query = proto.Field(proto.MESSAGE, number=2, message="BundledQuery",)
    read_time = proto.Field(proto.MESSAGE, number=3, message=timestamp_pb2.Timestamp,) def limit_type_of_query(query: BaseQuery) -> int:
    """BundledQuery.LimitType equivalent of this query.
    """

    return (
        BundledQuery.LimitType.LAST
        if query._limit_to_last
        else BundledQuery.LimitType.FIRST
    ) class AsyncQuery(BaseQuery):
    """Represents a query to the Firestore API.

    Instances of this class are considered immutable: all methods that
    would modify an instance instead return a new instance.

    Args:
        parent (:class:`~google.cloud.firestore_v1.collection.CollectionReference`):
            The collection that this query applies to.
        projection (Optional[:class:`google.cloud.proto.firestore.v1.\
            query.StructuredQuery.Projection`]):
            A projection of document fields to limit the query results to.
        field_filters (Optional[Tuple[:class:`google.cloud.proto.firestore.v1.\
            query.StructuredQuery.FieldFilter`, ...]]):
            The filters to be applied in the query.
        orders (Optional[Tuple[:class:`google.cloud.proto.firestore.v1.\
            query.StructuredQuery.Order`, ...]]):
            The "order by" entries to use in the query.
        limit (Optional[int]):
            The maximum number of documents the query is allowed to return.
        offset (Optional[int]):
            The number of results to skip.
        start_at (Optional[Tuple[dict, bool]]):
            Two-tuple of :

            * a mapping of fields. Any field that is present in this mapping
              must also be present in ``orders``
            * an ``after`` flag

            The fields and the flag combine to form a cursor used as
            a starting point in a query result set. If the ``after``
            flag is :data:`True`, the results will start just after any
            documents which have fields matching the cursor, otherwise
            any matching documents will be included in the result set.
            When the query is formed, the document values
            will be used in the order given by ``orders``.
        end_at (Optional[Tuple[dict, bool]]):
            Two-tuple of:

            * a mapping of fields. Any field that is present in this mapping
              must also be present in ``orders``
            * a ``before`` flag

            The fields and the flag combine to form a cursor used as
            an ending point in a query result set. If the ``before``
            flag is :data:`True`, the results will end just before any
            documents which have fields matching the cursor, otherwise
            any matching documents will be included in the result set.
            When the query is formed, the document values
            will be used in the order given by ``orders``.
        all_descendants (Optional[bool]):
            When false, selects only collections that are immediate children
            of the `parent` specified in the containing `RunQueryRequest`.
            When true, selects all descendant collections.
        recursive (Optional[bool]):
            When true, returns all documents and all documents in any subcollections
            below them. Defaults to false.
    """

    def __init__(
        self,
        parent,
        projection=None,
        field_filters=(),
        orders=(),
        limit=None,
        limit_to_last=False,
        offset=None,
        start_at=None,
        end_at=None,
        all_descendants=False,
        recursive=False,
    ) -> None:
        super(AsyncQuery, self).__init__(
            parent=parent,
            projection=projection,
            field_filters=field_filters,
            orders=orders,
            limit=limit,
            limit_to_last=limit_to_last,
            offset=offset,
            start_at=start_at,
            end_at=end_at,
            all_descendants=all_descendants,
            recursive=recursive,
        )

    async def _chunkify(
        self, chunk_size: int
    ) -> AsyncGenerator[List[DocumentSnapshot], None]:
        max_to_return: Optional[int] = self._limit
        num_returned: int = 0
        original: AsyncQuery = self._copy()
        last_document: Optional[DocumentSnapshot] = None

        while True:
            # Optionally trim the `chunk_size` down to honor a previously
            # applied limit as set by `self.limit()`
            _chunk_size: int = original._resolve_chunk_size(num_returned, chunk_size)

            # Apply the optionally pruned limit and the cursor, if we are past
            # the first page.
            _q = original.limit(_chunk_size)

            if last_document:
                _q = _q.start_after(last_document)

            snapshots = await _q.get()

            if snapshots:
                last_document = snapshots[-1]

            num_returned += len(snapshots)

            yield snapshots

            # Terminate the iterator if we have reached either of two end
            # conditions:
            #   1. There are no more documents, or
            #   2. We have reached the desired overall limit
            if len(snapshots) < _chunk_size or (
                max_to_return and num_returned >= max_to_return
            ):
                return

    async def get(
        self,
        transaction: Transaction = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> list:
        """Read the documents in the collection that match this query.

        This sends a ``RunQuery`` RPC and returns a list of documents
        returned in the stream of ``RunQueryResponse`` messages.

        Args:
            transaction
                (Optional[:class:`~google.cloud.firestore_v1.transaction.Transaction`]):
                An existing transaction that this query will run in.
            retry (google.api_core.retry.Retry): Designation of what errors, if any,
                should be retried.  Defaults to a system-specified policy.
            timeout (float): The timeout for this request.  Defaults to a
                system-specified value.

        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).

        Returns:
            list: The documents in the collection that match this query.
        """
        is_limited_to_last = self._limit_to_last

        if self._limit_to_last:
            # In order to fetch up to `self._limit` results from the end of the
            # query flip the defined ordering on the query to start from the
            # end, retrieving up to `self._limit` results from the backend.
            for order in self._orders:
                order.direction = _enum_from_direction(
                    self.DESCENDING
                    if order.direction == self.ASCENDING
                    else self.ASCENDING
                )
            self._limit_to_last = False

        result = self.stream(transaction=transaction, retry=retry, timeout=timeout)
        result = [d async for d in result]
        if is_limited_to_last:
            result = list(reversed(result))

        return result

    async def stream(
        self,
        transaction=None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> AsyncGenerator[async_document.DocumentSnapshot, None]:
        """Read the documents in the collection that match this query.

        This sends a ``RunQuery`` RPC and then returns an iterator which
        consumes each document returned in the stream of ``RunQueryResponse``
        messages.

        .. note::

           The underlying stream of responses will time out after
           the ``max_rpc_timeout_millis`` value set in the GAPIC
           client configuration for the ``RunQuery`` API.  Snapshots
           not consumed from the iterator before that point will be lost.

        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).

        Args:
            transaction
                (Optional[:class:`~google.cloud.firestore_v1.transaction.Transaction`]):
                An existing transaction that this query will run in.
            retry (google.api_core.retry.Retry): Designation of what errors, if any,
                should be retried.  Defaults to a system-specified policy.
            timeout (float): The timeout for this request.  Defaults to a
                system-specified value.

        Yields:
            :class:`~google.cloud.firestore_v1.async_document.DocumentSnapshot`:
            The next document that fulfills the query.
        """
        request, expected_prefix, kwargs = self._prep_stream(
            transaction, retry, timeout,
        )

        response_iterator = await self._client._firestore_api.run_query(
            request=request, metadata=self._client._rpc_metadata, **kwargs,
        )

        async for response in response_iterator:
            if self._all_descendants:
                snapshot = _collection_group_query_response_to_snapshot(
                    response, self._parent
                )
            else:
                snapshot = _query_response_to_snapshot(
                    response, self._parent, expected_prefix
                )
            if snapshot is not None:
                yield snapshot

    @staticmethod
    def _get_collection_reference_class() -> Type[
        "firestore_v1.async_collection.AsyncCollectionReference"
    ]:
        from google.cloud.firestore_v1.async_collection import AsyncCollectionReference

        return AsyncCollectionReference class BaseClient(ClientWithProject):
    """Client for interacting with Google Cloud Firestore API.

    .. note::

        Since the Cloud Firestore API requires the gRPC transport, no
        ``_http`` argument is accepted by this class.

    Args:
        project (Optional[str]): The project which the client acts on behalf
            of. If not passed, falls back to the default inferred
            from the environment.
        credentials (Optional[~google.auth.credentials.Credentials]): The
            OAuth2 Credentials to use for this client. If not passed, falls
            back to the default inferred from the environment.
        database (Optional[str]): The database name that the client targets.
            For now, :attr:`DEFAULT_DATABASE` (the default value) is the
            only valid database.
        client_info (Optional[google.api_core.gapic_v1.client_info.ClientInfo]):
            The client info used to send a user-agent string along with API
            requests. If ``None``, then default info will be used. Generally,
            you only need to set this if you're developing your own library
            or partner tool.
        client_options (Union[dict, google.api_core.client_options.ClientOptions]):
            Client options used to set user options on the client. API Endpoint
            should be set through client_options.
    """

    SCOPE = (
        "https://www.googleapis.com/auth/cloud-platform",
        "https://www.googleapis.com/auth/datastore",
    )
    """The scopes required for authenticating with the Firestore service."""

    _firestore_api_internal = None
    _database_string_internal = None
    _rpc_metadata_internal = None

    def __init__(
        self,
        project=None,
        credentials=None,
        database=DEFAULT_DATABASE,
        client_info=_CLIENT_INFO,
        client_options=None,
    ) -> None:
        # NOTE: This API has no use for the _http argument, but sending it
        #       will have no impact since the _http() @property only lazily
        #       creates a working HTTP object.
        self._emulator_host = os.getenv(_FIRESTORE_EMULATOR_HOST)

        if self._emulator_host is not None:
            if credentials is None:
                credentials = AnonymousCredentials()
            if project is None:
                project = _DEFAULT_EMULATOR_PROJECT

        super(BaseClient, self).__init__(
            project=project,
            credentials=credentials,
            client_options=client_options,
            _http=None,
        )
        self._client_info = client_info
        if client_options:
            if type(client_options) == dict:
                client_options = google.api_core.client_options.from_dict(
                    client_options
                )
        self._client_options = client_options

        self._database = database

    def _firestore_api_helper(self, transport, client_class, client_module) -> Any:
        """Lazy-loading getter GAPIC Firestore API.
        Returns:
            The GAPIC client with the credentials of the current client.
        """
        if self._firestore_api_internal is None:
            # Use a custom channel.
            # We need this in order to set appropriate keepalive options.

            if self._emulator_host is not None:
                channel = self._emulator_channel(transport)
            else:
                channel = transport.create_channel(
                    self._target,
                    credentials=self._credentials,
                    options={"grpc.keepalive_time_ms": 30000}.items(),
                )

            self._transport = transport(host=self._target, channel=channel)

            self._firestore_api_internal = client_class(
                transport=self._transport, client_options=self._client_options
            )
            client_module._client_info = self._client_info

        return self._firestore_api_internal

    def _emulator_channel(self, transport):
        """
        Creates an insecure channel to communicate with the local emulator.
        If credentials are provided the token is extracted and added to the
        headers. This supports local testing of firestore rules if the credentials
        have been created from a signed custom token.

        :return: grpc.Channel or grpc.aio.Channel
        """
        # Insecure channels are used for the emulator as secure channels
        # cannot be used to communicate on some environments.
        # https://github.com/googleapis/python-firestore/issues/359
        # Default the token to a non-empty string, in this case "owner".
        token = "owner"
        if (
            self._credentials is not None
            and getattr(self._credentials, "id_token", None) is not None
        ):
            token = self._credentials.id_token
        options = [("Authorization", f"Bearer {token}")]

        if "GrpcAsyncIOTransport" in str(transport.__name__):
            return grpc.aio.insecure_channel(self._emulator_host, options=options)
        else:
            return grpc.insecure_channel(self._emulator_host, options=options)

    def _target_helper(self, client_class) -> str:
        """Return the target (where the API is).
        Eg. "firestore.googleapis.com"

        Returns:
            str: The location of the API.
        """
        if self._emulator_host is not None:
            return self._emulator_host
        elif self._client_options and self._client_options.api_endpoint:
            return self._client_options.api_endpoint
        else:
            return client_class.DEFAULT_ENDPOINT

    @property
    def _database_string(self):
        """The database string corresponding to this client's project.

        This value is lazy-loaded and cached.

        Will be of the form

            ``projects/{project_id}/databases/{database_id}``

        but ``database_id == '(default)'`` for the time being.

        Returns:
            str: The fully-qualified database string for the current
            project. (The default database is also in this string.)
        """
        if self._database_string_internal is None:
            db_str = google.api_core.path_template.expand(
                "projects/{project}/databases/{database}",
                project=self.project,
                database=self._database,
            )

            self._database_string_internal = db_str

        return self._database_string_internal

    @property
    def _rpc_metadata(self):
        """The RPC metadata for this client's associated database.

        Returns:
            Sequence[Tuple(str, str)]: RPC metadata with resource prefix
            for the database associated with this client.
        """
        if self._rpc_metadata_internal is None:
            self._rpc_metadata_internal = _helpers.metadata_with_prefix(
                self._database_string
            )

            if self._emulator_host is not None:
                # The emulator requires additional metadata to be set.
                self._rpc_metadata_internal.append(("authorization", "Bearer owner"))

        return self._rpc_metadata_internal

    def collection(self, *collection_path) -> BaseCollectionReference:
        raise NotImplementedError

    def collection_group(self, collection_id: str) -> BaseQuery:
        raise NotImplementedError

    def _get_collection_reference(self, collection_id: str) -> BaseCollectionReference:
        """Checks validity of collection_id and then uses subclasses collection implementation.

        Args:
            collection_id (str) Identifies the collections to query over.

                Every collection or subcollection with this ID as the last segment of its
                path will be included. Cannot contain a slash.

        Returns:
            The created collection.
        """
        if "/" in collection_id:
            raise ValueError(
                "Invalid collection_id "
                + collection_id
                + ". Collection IDs must not contain '/'."
            )

        return self.collection(collection_id)

    def document(self, *document_path) -> BaseDocumentReference:
        raise NotImplementedError

    def bulk_writer(self, options: Optional[BulkWriterOptions] = None) -> BulkWriter:
        """Get a BulkWriter instance from this client.

        Args:
            :class:`@google.cloud.firestore_v1.bulk_writer.BulkWriterOptions`:
            Optional control parameters for the
            :class:`@google.cloud.firestore_v1.bulk_writer.BulkWriter` returned.

        Returns:
            :class:`@google.cloud.firestore_v1.bulk_writer.BulkWriter`:
            A utility to efficiently create and save many `WriteBatch` instances
            to the server.
        """
        return BulkWriter(client=self, options=options)

    def _document_path_helper(self, *document_path) -> List[str]:
        """Standardize the format of path to tuple of path segments and strip the database string from path if present.

        Args:
            document_path (Tuple[str, ...]): Can either be

                * A single ``/``-delimited path to a document
                * A tuple of document path segments
        """
        path = _path_helper(document_path)
        base_path = self._database_string + "/documents/"
        joined_path = _helpers.DOCUMENT_PATH_DELIMITER.join(path)
        if joined_path.startswith(base_path):
            joined_path = joined_path[len(base_path) :]
        return joined_path.split(_helpers.DOCUMENT_PATH_DELIMITER)

    def recursive_delete(
        self,
        reference: Union[BaseCollectionReference, BaseDocumentReference],
        bulk_writer: Optional["BulkWriter"] = None,  # type: ignore
    ) -> int:
        raise NotImplementedError

    @staticmethod
    def field_path(*field_names: str) -> str:
        """Create a **field path** from a list of nested field names.

        A **field path** is a ``.``-delimited concatenation of the field
        names. It is used to represent a nested field. For example,
        in the data

        .. code-block:: python

           data = {
              'aa': {
                  'bb': {
                      'cc': 10,
                  },
              },
           }

        the field path ``'aa.bb.cc'`` represents the data stored in
        ``data['aa']['bb']['cc']``.

        Args:
            field_names: The list of field names.

        Returns:
            str: The ``.``-delimited field path.
        """
        return render_field_path(field_names)

    @staticmethod
    def write_option(
        **kwargs,
    ) -> Union[
        _helpers.ExistsOption, _helpers.LastUpdateOption,
    ]:
        """Create a write option for write operations.

        Write operations include :meth:`~google.cloud.DocumentReference.set`,
        :meth:`~google.cloud.DocumentReference.update` and
        :meth:`~google.cloud.DocumentReference.delete`.

        One of the following keyword arguments must be provided:

        * ``last_update_time`` (:class:`google.protobuf.timestamp_pb2.\
               Timestamp`): A timestamp. When set, the target document must
               exist and have been last updated at that time. Protobuf
               ``update_time`` timestamps are typically returned from methods
               that perform write operations as part of a "write result"
               protobuf or directly.
        * ``exists`` (:class:`bool`): Indicates if the document being modified
              should already exist.

        Providing no argument would make the option have no effect (so
        it is not allowed). Providing multiple would be an apparent
        contradiction, since ``last_update_time`` assumes that the
        document **was** updated (it can't have been updated if it
        doesn't exist) and ``exists`` indicate that it is unknown if the
        document exists or not.

        Args:
            kwargs (Dict[str, Any]): The keyword arguments described above.

        Raises:
            TypeError: If anything other than exactly one argument is
                provided by the caller.

        Returns:
            :class:`~google.cloud.firestore_v1.client.WriteOption`:
            The option to be used to configure a write message.
        """
        if len(kwargs) != 1:
            raise TypeError(_BAD_OPTION_ERR)

        name, value = kwargs.popitem()
        if name == "last_update_time":
            return _helpers.LastUpdateOption(value)
        elif name == "exists":
            return _helpers.ExistsOption(value)
        else:
            extra = "{!r} was provided".format(name)
            raise TypeError(_BAD_OPTION_ERR, extra)

    def _prep_get_all(
        self,
        references: list,
        field_paths: Iterable[str] = None,
        transaction: BaseTransaction = None,
        retry: retries.Retry = None,
        timeout: float = None,
    ) -> Tuple[dict, dict, dict]:
        """Shared setup for async/sync :meth:`get_all`."""
        document_paths, reference_map = _reference_info(references)
        mask = _get_doc_mask(field_paths)
        request = {
            "database": self._database_string,
            "documents": document_paths,
            "mask": mask,
            "transaction": _helpers.get_transaction_id(transaction),
        }
        kwargs = _helpers.make_retry_timeout_kwargs(retry, timeout)

        return request, reference_map, kwargs

    def get_all(
        self,
        references: list,
        field_paths: Iterable[str] = None,
        transaction: BaseTransaction = None,
        retry: retries.Retry = None,
        timeout: float = None,
    ) -> Union[
        AsyncGenerator[DocumentSnapshot, Any], Generator[DocumentSnapshot, Any, Any]
    ]:
        raise NotImplementedError

    def _prep_collections(
        self, retry: retries.Retry = None, timeout: float = None,
    ) -> Tuple[dict, dict]:
        """Shared setup for async/sync :meth:`collections`."""
        request = {"parent": "{}/documents".format(self._database_string)}
        kwargs = _helpers.make_retry_timeout_kwargs(retry, timeout)

        return request, kwargs

    def collections(
        self, retry: retries.Retry = None, timeout: float = None,
    ) -> Union[
        AsyncGenerator[BaseCollectionReference, Any],
        Generator[BaseCollectionReference, Any, Any],
    ]:
        raise NotImplementedError

    def batch(self) -> BaseWriteBatch:
        raise NotImplementedError

    def transaction(self, **kwargs) -> BaseTransaction:
        raise NotImplementedError class DocumentSnapshot(object):
    """A snapshot of document data in a Firestore database.

    This represents data retrieved at a specific time and may not contain
    all fields stored for the document (i.e. a hand-picked selection of
    fields may have been retrieved).

    Instances of this class are not intended to be constructed by hand,
    rather they'll be returned as responses to various methods, such as
    :meth:`~google.cloud.DocumentReference.get`.

    Args:
        reference (:class:`~google.cloud.firestore_v1.document.DocumentReference`):
            A document reference corresponding to the document that contains
            the data in this snapshot.
        data (Dict[str, Any]):
            The data retrieved in the snapshot.
        exists (bool):
            Indicates if the document existed at the time the snapshot was
            retrieved.
        read_time (:class:`proto.datetime_helpers.DatetimeWithNanoseconds`):
            The time that this snapshot was read from the server.
        create_time (:class:`proto.datetime_helpers.DatetimeWithNanoseconds`):
            The time that this document was created.
        update_time (:class:`proto.datetime_helpers.DatetimeWithNanoseconds`):
            The time that this document was last updated.
    """

    def __init__(
        self, reference, data, exists, read_time, create_time, update_time
    ) -> None:
        self._reference = reference
        # We want immutable data, so callers can't modify this value
        # out from under us.
        self._data = copy.deepcopy(data)
        self._exists = exists
        self.read_time = read_time
        self.create_time = create_time
        self.update_time = update_time

    def __eq__(self, other):
        if not isinstance(other, self.__class__):
            return NotImplemented
        return self._reference == other._reference and self._data == other._data

    def __hash__(self):
        return hash(self._reference) + hash(self.update_time)

    @property
    def _client(self):
        """The client that owns the document reference for this snapshot.

        Returns:
            :class:`~google.cloud.firestore_v1.client.Client`:
            The client that owns this document.
        """
        return self._reference._client

    @property
    def exists(self):
        """Existence flag.

        Indicates if the document existed at the time this snapshot
        was retrieved.

        Returns:
            bool: The existence flag.
        """
        return self._exists

    @property
    def id(self):
        """The document identifier (within its collection).

        Returns:
            str: The last component of the path of the document.
        """
        return self._reference.id

    @property
    def reference(self):
        """Document reference corresponding to document that owns this data.

        Returns:
            :class:`~google.cloud.firestore_v1.document.DocumentReference`:
            A document reference corresponding to this document.
        """
        return self._reference

    def get(self, field_path: str) -> Any:
        """Get a value from the snapshot data.

        If the data is nested, for example:

        .. code-block:: python

           >>> snapshot.to_dict()
           {
               'top1': {
                   'middle2': {
                       'bottom3': 20,
                       'bottom4': 22,
                   },
                   'middle5': True,
               },
               'top6': b'\x00\x01 foo',
           }

        a **field path** can be used to access the nested data. For
        example:

        .. code-block:: python

           >>> snapshot.get('top1')
           {
               'middle2': {
                   'bottom3': 20,
                   'bottom4': 22,
               },
               'middle5': True,
           }
           >>> snapshot.get('top1.middle2')
           {
               'bottom3': 20,
               'bottom4': 22,
           }
           >>> snapshot.get('top1.middle2.bottom3')
           20

        See :meth:`~google.cloud.firestore_v1.client.Client.field_path` for
        more information on **field paths**.

        A copy is returned since the data may contain mutable values,
        but the data stored in the snapshot must remain immutable.

        Args:
            field_path (str): A field path (``.``-delimited list of
                field names).

        Returns:
            Any or None:
                (A copy of) the value stored for the ``field_path`` or
                None if snapshot document does not exist.

        Raises:
            KeyError: If the ``field_path`` does not match nested data
                in the snapshot.
        """
        if not self._exists:
            return None
        nested_data = field_path_module.get_nested_value(field_path, self._data)
        return copy.deepcopy(nested_data)

    def to_dict(self) -> Union[Dict[str, Any], None]:
        """Retrieve the data contained in this snapshot.

        A copy is returned since the data may contain mutable values,
        but the data stored in the snapshot must remain immutable.

        Returns:
            Dict[str, Any] or None:
                The data in the snapshot.  Returns None if reference
                does not exist.
        """
        if not self._exists:
            return None
        return copy.deepcopy(self._data)

    def _to_protobuf(self) -> Optional[Document]:
        return _helpers.document_snapshot_to_protobuf(self) class BaseQuery(object):
    """Represents a query to the Firestore API.

    Instances of this class are considered immutable: all methods that
    would modify an instance instead return a new instance.

    Args:
        parent (:class:`~google.cloud.firestore_v1.collection.CollectionReference`):
            The collection that this query applies to.
        projection (Optional[:class:`google.cloud.proto.firestore.v1.\
            query.StructuredQuery.Projection`]):
            A projection of document fields to limit the query results to.
        field_filters (Optional[Tuple[:class:`google.cloud.proto.firestore.v1.\
            query.StructuredQuery.FieldFilter`, ...]]):
            The filters to be applied in the query.
        orders (Optional[Tuple[:class:`google.cloud.proto.firestore.v1.\
            query.StructuredQuery.Order`, ...]]):
            The "order by" entries to use in the query.
        limit (Optional[int]):
            The maximum number of documents the query is allowed to return.
        limit_to_last (Optional[bool]):
            Denotes whether a provided limit is applied to the end of the result set.
        offset (Optional[int]):
            The number of results to skip.
        start_at (Optional[Tuple[dict, bool]]):
            Two-tuple of :

            * a mapping of fields. Any field that is present in this mapping
              must also be present in ``orders``
            * an ``after`` flag

            The fields and the flag combine to form a cursor used as
            a starting point in a query result set. If the ``after``
            flag is :data:`True`, the results will start just after any
            documents which have fields matching the cursor, otherwise
            any matching documents will be included in the result set.
            When the query is formed, the document values
            will be used in the order given by ``orders``.
        end_at (Optional[Tuple[dict, bool]]):
            Two-tuple of:

            * a mapping of fields. Any field that is present in this mapping
              must also be present in ``orders``
            * a ``before`` flag

            The fields and the flag combine to form a cursor used as
            an ending point in a query result set. If the ``before``
            flag is :data:`True`, the results will end just before any
            documents which have fields matching the cursor, otherwise
            any matching documents will be included in the result set.
            When the query is formed, the document values
            will be used in the order given by ``orders``.
        all_descendants (Optional[bool]):
            When false, selects only collections that are immediate children
            of the `parent` specified in the containing `RunQueryRequest`.
            When true, selects all descendant collections.
        recursive (Optional[bool]):
            When true, returns all documents and all documents in any subcollections
            below them. Defaults to false.
    """

    ASCENDING = "ASCENDING"
    """str: Sort query results in ascending order on a field."""
    DESCENDING = "DESCENDING"
    """str: Sort query results in descending order on a field."""

    def __init__(
        self,
        parent,
        projection=None,
        field_filters=(),
        orders=(),
        limit=None,
        limit_to_last=False,
        offset=None,
        start_at=None,
        end_at=None,
        all_descendants=False,
        recursive=False,
    ) -> None:
        self._parent = parent
        self._projection = projection
        self._field_filters = field_filters
        self._orders = orders
        self._limit = limit
        self._limit_to_last = limit_to_last
        self._offset = offset
        self._start_at = start_at
        self._end_at = end_at
        self._all_descendants = all_descendants
        self._recursive = recursive

    def __eq__(self, other):
        if not isinstance(other, self.__class__):
            return NotImplemented
        return (
            self._parent == other._parent
            and self._projection == other._projection
            and self._field_filters == other._field_filters
            and self._orders == other._orders
            and self._limit == other._limit
            and self._limit_to_last == other._limit_to_last
            and self._offset == other._offset
            and self._start_at == other._start_at
            and self._end_at == other._end_at
            and self._all_descendants == other._all_descendants
        )

    @property
    def _client(self):
        """The client of the parent collection.

        Returns:
            :class:`~google.cloud.firestore_v1.client.Client`:
            The client that owns this query.
        """
        return self._parent._client

    def select(self, field_paths: Iterable[str]) -> "BaseQuery":
        """Project documents matching query to a limited set of fields.

        See :meth:`~google.cloud.firestore_v1.client.Client.field_path` for
        more information on **field paths**.

        If the current query already has a projection set (i.e. has already
        called :meth:`~google.cloud.firestore_v1.query.Query.select`), this
        will overwrite it.

        Args:
            field_paths (Iterable[str, ...]): An iterable of field paths
                (``.``-delimited list of field names) to use as a projection
                of document fields in the query results.

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A "projected" query. Acts as a copy of the current query,
            modified with the newly added projection.
        Raises:
            ValueError: If any ``field_path`` is invalid.
        """
        field_paths = list(field_paths)
        for field_path in field_paths:
            field_path_module.split_field_path(field_path)  # raises

        new_projection = query.StructuredQuery.Projection(
            fields=[
                query.StructuredQuery.FieldReference(field_path=field_path)
                for field_path in field_paths
            ]
        )
        return self._copy(projection=new_projection)

    def _copy(
        self,
        *,
        projection: Optional[query.StructuredQuery.Projection] = _not_passed,
        field_filters: Optional[Tuple[query.StructuredQuery.FieldFilter]] = _not_passed,
        orders: Optional[Tuple[query.StructuredQuery.Order]] = _not_passed,
        limit: Optional[int] = _not_passed,
        limit_to_last: Optional[bool] = _not_passed,
        offset: Optional[int] = _not_passed,
        start_at: Optional[Tuple[dict, bool]] = _not_passed,
        end_at: Optional[Tuple[dict, bool]] = _not_passed,
        all_descendants: Optional[bool] = _not_passed,
        recursive: Optional[bool] = _not_passed,
    ) -> "BaseQuery":
        return self.__class__(
            self._parent,
            projection=self._evaluate_param(projection, self._projection),
            field_filters=self._evaluate_param(field_filters, self._field_filters),
            orders=self._evaluate_param(orders, self._orders),
            limit=self._evaluate_param(limit, self._limit),
            limit_to_last=self._evaluate_param(limit_to_last, self._limit_to_last),
            offset=self._evaluate_param(offset, self._offset),
            start_at=self._evaluate_param(start_at, self._start_at),
            end_at=self._evaluate_param(end_at, self._end_at),
            all_descendants=self._evaluate_param(
                all_descendants, self._all_descendants
            ),
            recursive=self._evaluate_param(recursive, self._recursive),
        )

    def _evaluate_param(self, value, fallback_value):
        """Helper which allows `None` to be passed into `copy` and be set on the
        copy instead of being misinterpreted as an unpassed parameter."""
        return value if value is not _not_passed else fallback_value

    def where(self, field_path: str, op_string: str, value) -> "BaseQuery":
        """Filter the query on a field.

        See :meth:`~google.cloud.firestore_v1.client.Client.field_path` for
        more information on **field paths**.

        Returns a new :class:`~google.cloud.firestore_v1.query.Query` that
        filters on a specific field path, according to an operation (e.g.
        ``==`` or "equals") and a particular value to be paired with that
        operation.

        Args:
            field_path (str): A field path (``.``-delimited list of
                field names) for the field to filter on.
            op_string (str): A comparison operation in the form of a string.
                Acceptable values are ``<``, ``<=``, ``==``, ``!=``, ``>=``, ``>``,
                ``in``, ``not-in``, ``array_contains`` and ``array_contains_any``.
            value (Any): The value to compare the field against in the filter.
                If ``value`` is :data:`None` or a NaN, then ``==`` is the only
                allowed operation.

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A filtered query. Acts as a copy of the current query,
            modified with the newly added filter.

        Raises:
            ValueError: If ``field_path`` is invalid.
            ValueError: If ``value`` is a NaN or :data:`None` and
                ``op_string`` is not ``==``.
        """
        field_path_module.split_field_path(field_path)  # raises

        if value is None:
            if op_string != _EQ_OP:
                raise ValueError(_BAD_OP_NAN_NULL)
            filter_pb = query.StructuredQuery.UnaryFilter(
                field=query.StructuredQuery.FieldReference(field_path=field_path),
                op=StructuredQuery.UnaryFilter.Operator.IS_NULL,
            )
        elif _isnan(value):
            if op_string != _EQ_OP:
                raise ValueError(_BAD_OP_NAN_NULL)
            filter_pb = query.StructuredQuery.UnaryFilter(
                field=query.StructuredQuery.FieldReference(field_path=field_path),
                op=StructuredQuery.UnaryFilter.Operator.IS_NAN,
            )
        elif isinstance(value, (transforms.Sentinel, transforms._ValueList)):
            raise ValueError(_INVALID_WHERE_TRANSFORM)
        else:
            filter_pb = query.StructuredQuery.FieldFilter(
                field=query.StructuredQuery.FieldReference(field_path=field_path),
                op=_enum_from_op_string(op_string),
                value=_helpers.encode_value(value),
            )

        new_filters = self._field_filters + (filter_pb,)
        return self._copy(field_filters=new_filters)

    @staticmethod
    def _make_order(field_path, direction) -> StructuredQuery.Order:
        """Helper for :meth:`order_by`."""
        return query.StructuredQuery.Order(
            field=query.StructuredQuery.FieldReference(field_path=field_path),
            direction=_enum_from_direction(direction),
        )

    def order_by(self, field_path: str, direction: str = ASCENDING) -> "BaseQuery":
        """Modify the query to add an order clause on a specific field.

        See :meth:`~google.cloud.firestore_v1.client.Client.field_path` for
        more information on **field paths**.

        Successive :meth:`~google.cloud.firestore_v1.query.Query.order_by`
        calls will further refine the ordering of results returned by the query
        (i.e. the new "order by" fields will be added to existing ones).

        Args:
            field_path (str): A field path (``.``-delimited list of
                field names) on which to order the query results.
            direction (Optional[str]): The direction to order by. Must be one
                of :attr:`ASCENDING` or :attr:`DESCENDING`, defaults to
                :attr:`ASCENDING`.

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            An ordered query. Acts as a copy of the current query, modified
            with the newly added "order by" constraint.

        Raises:
            ValueError: If ``field_path`` is invalid.
            ValueError: If ``direction`` is not one of :attr:`ASCENDING` or
                :attr:`DESCENDING`.
        """
        field_path_module.split_field_path(field_path)  # raises

        order_pb = self._make_order(field_path, direction)

        new_orders = self._orders + (order_pb,)
        return self._copy(orders=new_orders)

    def limit(self, count: int) -> "BaseQuery":
        """Limit a query to return at most `count` matching results.

        If the current query already has a `limit` set, this will override it.

        .. note::
           `limit` and `limit_to_last` are mutually exclusive.
           Setting `limit` will drop previously set `limit_to_last`.

        Args:
            count (int): Maximum number of documents to return that match
                the query.
        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A limited query. Acts as a copy of the current query, modified
            with the newly added "limit" filter.
        """
        return self._copy(limit=count, limit_to_last=False)

    def limit_to_last(self, count: int) -> "BaseQuery":
        """Limit a query to return the last `count` matching results.
        If the current query already has a `limit_to_last`
        set, this will override it.

        .. note::
           `limit` and `limit_to_last` are mutually exclusive.
           Setting `limit_to_last` will drop previously set `limit`.

        Args:
            count (int): Maximum number of documents to return that match
                the query.
        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A limited query. Acts as a copy of the current query, modified
            with the newly added "limit" filter.
        """
        return self._copy(limit=count, limit_to_last=True)

    def _resolve_chunk_size(self, num_loaded: int, chunk_size: int) -> int:
        """Utility function for chunkify."""
        if self._limit is not None and (num_loaded + chunk_size) > self._limit:
            return max(self._limit - num_loaded, 0)
        return chunk_size

    def offset(self, num_to_skip: int) -> "BaseQuery":
        """Skip to an offset in a query.

        If the current query already has specified an offset, this will
        overwrite it.

        Args:
            num_to_skip (int): The number of results to skip at the beginning
                of query results. (Must be non-negative.)

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            An offset query. Acts as a copy of the current query, modified
            with the newly added "offset" field.
        """
        return self._copy(offset=num_to_skip)

    def _check_snapshot(self, document_snapshot) -> None:
        """Validate local snapshots for non-collection-group queries.

        Raises:
            ValueError: for non-collection-group queries, if the snapshot
                is from a different collection.
        """
        if self._all_descendants:
            return

        if document_snapshot.reference._path[:-1] != self._parent._path:
            raise ValueError("Cannot use snapshot from another collection as a cursor.")

    def _cursor_helper(
        self,
        document_fields_or_snapshot: Union[DocumentSnapshot, dict, list, tuple],
        before: bool,
        start: bool,
    ) -> "BaseQuery":
        """Set values to be used for a ``start_at`` or ``end_at`` cursor.

        The values will later be used in a query protobuf.

        When the query is sent to the server, the ``document_fields_or_snapshot`` will
        be used in the order given by fields set by
        :meth:`~google.cloud.firestore_v1.query.Query.order_by`.

        Args:
            document_fields_or_snapshot
                (Union[:class:`~google.cloud.firestore_v1.document.DocumentSnapshot`, dict, list, tuple]):
                a document snapshot or a dictionary/list/tuple of fields
                representing a query results cursor. A cursor is a collection
                of values that represent a position in a query result set.
            before (bool): Flag indicating if the document in
                ``document_fields_or_snapshot`` should (:data:`False`) or
                shouldn't (:data:`True`) be included in the result set.
            start (Optional[bool]): determines if the cursor is a ``start_at``
                cursor (:data:`True`) or an ``end_at`` cursor (:data:`False`).

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A query with cursor. Acts as a copy of the current query, modified
            with the newly added "start at" cursor.
        """
        if isinstance(document_fields_or_snapshot, tuple):
            document_fields_or_snapshot = list(document_fields_or_snapshot)
        elif isinstance(document_fields_or_snapshot, document.DocumentSnapshot):
            self._check_snapshot(document_fields_or_snapshot)
        else:
            # NOTE: We copy so that the caller can't modify after calling.
            document_fields_or_snapshot = copy.deepcopy(document_fields_or_snapshot)

        cursor_pair = document_fields_or_snapshot, before
        query_kwargs = {
            "projection": self._projection,
            "field_filters": self._field_filters,
            "orders": self._orders,
            "limit": self._limit,
            "offset": self._offset,
            "all_descendants": self._all_descendants,
        }
        if start:
            query_kwargs["start_at"] = cursor_pair
            query_kwargs["end_at"] = self._end_at
        else:
            query_kwargs["start_at"] = self._start_at
            query_kwargs["end_at"] = cursor_pair

        return self._copy(**query_kwargs)

    def start_at(
        self, document_fields_or_snapshot: Union[DocumentSnapshot, dict, list, tuple]
    ) -> "BaseQuery":
        """Start query results at a particular document value.

        The result set will **include** the document specified by
        ``document_fields_or_snapshot``.

        If the current query already has specified a start cursor -- either
        via this method or
        :meth:`~google.cloud.firestore_v1.query.Query.start_after` -- this
        will overwrite it.

        When the query is sent to the server, the ``document_fields`` will
        be used in the order given by fields set by
        :meth:`~google.cloud.firestore_v1.query.Query.order_by`.

        Args:
            document_fields_or_snapshot
                (Union[:class:`~google.cloud.firestore_v1.document.DocumentSnapshot`, dict, list, tuple]):
                a document snapshot or a dictionary/list/tuple of fields
                representing a query results cursor. A cursor is a collection
                of values that represent a position in a query result set.

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A query with cursor. Acts as
            a copy of the current query, modified with the newly added
            "start at" cursor.
        """
        return self._cursor_helper(document_fields_or_snapshot, before=True, start=True)

    def start_after(
        self, document_fields_or_snapshot: Union[DocumentSnapshot, dict, list, tuple]
    ) -> "BaseQuery":
        """Start query results after a particular document value.

        The result set will **exclude** the document specified by
        ``document_fields_or_snapshot``.

        If the current query already has specified a start cursor -- either
        via this method or
        :meth:`~google.cloud.firestore_v1.query.Query.start_at` -- this will
        overwrite it.

        When the query is sent to the server, the ``document_fields_or_snapshot`` will
        be used in the order given by fields set by
        :meth:`~google.cloud.firestore_v1.query.Query.order_by`.

        Args:
            document_fields_or_snapshot
                (Union[:class:`~google.cloud.firestore_v1.document.DocumentSnapshot`, dict, list, tuple]):
                a document snapshot or a dictionary/list/tuple of fields
                representing a query results cursor. A cursor is a collection
                of values that represent a position in a query result set.

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A query with cursor. Acts as a copy of the current query, modified
            with the newly added "start after" cursor.
        """
        return self._cursor_helper(
            document_fields_or_snapshot, before=False, start=True
        )

    def end_before(
        self, document_fields_or_snapshot: Union[DocumentSnapshot, dict, list, tuple]
    ) -> "BaseQuery":
        """End query results before a particular document value.

        The result set will **exclude** the document specified by
        ``document_fields_or_snapshot``.

        If the current query already has specified an end cursor -- either
        via this method or
        :meth:`~google.cloud.firestore_v1.query.Query.end_at` -- this will
        overwrite it.

        When the query is sent to the server, the ``document_fields_or_snapshot`` will
        be used in the order given by fields set by
        :meth:`~google.cloud.firestore_v1.query.Query.order_by`.

        Args:
            document_fields_or_snapshot
                (Union[:class:`~google.cloud.firestore_v1.document.DocumentSnapshot`, dict, list, tuple]):
                a document snapshot or a dictionary/list/tuple of fields
                representing a query results cursor. A cursor is a collection
                of values that represent a position in a query result set.

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A query with cursor. Acts as a copy of the current query, modified
            with the newly added "end before" cursor.
        """
        return self._cursor_helper(
            document_fields_or_snapshot, before=True, start=False
        )

    def end_at(
        self, document_fields_or_snapshot: Union[DocumentSnapshot, dict, list, tuple]
    ) -> "BaseQuery":
        """End query results at a particular document value.

        The result set will **include** the document specified by
        ``document_fields_or_snapshot``.

        If the current query already has specified an end cursor -- either
        via this method or
        :meth:`~google.cloud.firestore_v1.query.Query.end_before` -- this will
        overwrite it.

        When the query is sent to the server, the ``document_fields_or_snapshot`` will
        be used in the order given by fields set by
        :meth:`~google.cloud.firestore_v1.query.Query.order_by`.

        Args:
            document_fields_or_snapshot
                (Union[:class:`~google.cloud.firestore_v1.document.DocumentSnapshot`, dict, list, tuple]):
                a document snapshot or a dictionary/list/tuple of fields
                representing a query results cursor. A cursor is a collection
                of values that represent a position in a query result set.

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`:
            A query with cursor. Acts as a copy of the current query, modified
            with the newly added "end at" cursor.
        """
        return self._cursor_helper(
            document_fields_or_snapshot, before=False, start=False
        )

    def _filters_pb(self) -> StructuredQuery.Filter:
        """Convert all the filters into a single generic Filter protobuf.

        This may be a lone field filter or unary filter, may be a composite
        filter or may be :data:`None`.

        Returns:
            :class:`google.cloud.firestore_v1.types.StructuredQuery.Filter`:
            A "generic" filter representing the current query's filters.
        """
        num_filters = len(self._field_filters)
        if num_filters == 0:
            return None
        elif num_filters == 1:
            return _filter_pb(self._field_filters[0])
        else:
            composite_filter = query.StructuredQuery.CompositeFilter(
                op=StructuredQuery.CompositeFilter.Operator.AND,
                filters=[_filter_pb(filter_) for filter_ in self._field_filters],
            )
            return query.StructuredQuery.Filter(composite_filter=composite_filter)

    @staticmethod
    def _normalize_projection(projection) -> StructuredQuery.Projection:
        """Helper:  convert field paths to message."""
        if projection is not None:

            fields = list(projection.fields)

            if not fields:
                field_ref = query.StructuredQuery.FieldReference(field_path="__name__")
                return query.StructuredQuery.Projection(fields=[field_ref])

        return projection

    def _normalize_orders(self) -> list:
        """Helper:  adjust orders based on cursors, where clauses."""
        orders = list(self._orders)
        _has_snapshot_cursor = False

        if self._start_at:
            if isinstance(self._start_at[0], document.DocumentSnapshot):
                _has_snapshot_cursor = True

        if self._end_at:
            if isinstance(self._end_at[0], document.DocumentSnapshot):
                _has_snapshot_cursor = True

        if _has_snapshot_cursor:
            should_order = [
                _enum_from_op_string(key)
                for key in _COMPARISON_OPERATORS
                if key not in (_EQ_OP, "array_contains")
            ]
            order_keys = [order.field.field_path for order in orders]
            for filter_ in self._field_filters:
                # FieldFilter.Operator should not compare equal to
                # UnaryFilter.Operator, but it does
                if isinstance(filter_.op, StructuredQuery.FieldFilter.Operator):
                    field = filter_.field.field_path
                    if filter_.op in should_order and field not in order_keys:
                        orders.append(self._make_order(field, "ASCENDING"))
            if not orders:
                orders.append(self._make_order("__name__", "ASCENDING"))
            else:
                order_keys = [order.field.field_path for order in orders]
                if "__name__" not in order_keys:
                    direction = orders[-1].direction  # enum?
                    orders.append(self._make_order("__name__", direction))

        return orders

    def _normalize_cursor(self, cursor, orders) -> Optional[Tuple[Any, Any]]:
        """Helper: convert cursor to a list of values based on orders."""
        if cursor is None:
            return

        if not orders:
            raise ValueError(_NO_ORDERS_FOR_CURSOR)

        document_fields, before = cursor

        order_keys = [order.field.field_path for order in orders]

        if isinstance(document_fields, document.DocumentSnapshot):
            snapshot = document_fields
            document_fields = snapshot.to_dict()
            document_fields["__name__"] = snapshot.reference

        if isinstance(document_fields, dict):
            # Transform to list using orders
            values = []
            data = document_fields

            # It isn't required that all order by have a cursor.
            # However, we need to be sure they are specified in order without gaps
            for order_key in order_keys[: len(data)]:
                try:
                    if order_key in data:
                        values.append(data[order_key])
                    else:
                        values.append(
                            field_path_module.get_nested_value(order_key, data)
                        )
                except KeyError:
                    msg = _MISSING_ORDER_BY.format(order_key, data)
                    raise ValueError(msg)

            document_fields = values

        if document_fields and len(document_fields) > len(orders):
            msg = _MISMATCH_CURSOR_W_ORDER_BY.format(document_fields, order_keys)
            raise ValueError(msg)

        _transform_bases = (transforms.Sentinel, transforms._ValueList)

        for index, key_field in enumerate(zip(order_keys, document_fields)):
            key, field = key_field

            if isinstance(field, _transform_bases):
                msg = _INVALID_CURSOR_TRANSFORM
                raise ValueError(msg)

            if key == "__name__" and isinstance(field, str):
                document_fields[index] = self._parent.document(field)

        return document_fields, before

    def _to_protobuf(self) -> StructuredQuery:
        """Convert the current query into the equivalent protobuf.

        Returns:
            :class:`google.cloud.firestore_v1.types.StructuredQuery`:
            The query protobuf.
        """
        projection = self._normalize_projection(self._projection)
        orders = self._normalize_orders()
        start_at = self._normalize_cursor(self._start_at, orders)
        end_at = self._normalize_cursor(self._end_at, orders)

        query_kwargs = {
            "select": projection,
            "from_": [
                query.StructuredQuery.CollectionSelector(
                    collection_id=self._parent.id, all_descendants=self._all_descendants
                )
            ],
            "where": self._filters_pb(),
            "order_by": orders,
            "start_at": _cursor_pb(start_at),
            "end_at": _cursor_pb(end_at),
        }
        if self._offset is not None:
            query_kwargs["offset"] = self._offset
        if self._limit is not None:
            query_kwargs["limit"] = wrappers_pb2.Int32Value(value=self._limit)
        return query.StructuredQuery(**query_kwargs)

    def get(
        self, transaction=None, retry: retries.Retry = None, timeout: float = None,
    ) -> Iterable[DocumentSnapshot]:
        raise NotImplementedError

    def _prep_stream(
        self, transaction=None, retry: retries.Retry = None, timeout: float = None,
    ) -> Tuple[dict, str, dict]:
        """Shared setup for async / sync :meth:`stream`"""
        if self._limit_to_last:
            raise ValueError(
                "Query results for queries that include limit_to_last() "
                "constraints cannot be streamed. Use Query.get() instead."
            )

        parent_path, expected_prefix = self._parent._parent_info()
        request = {
            "parent": parent_path,
            "structured_query": self._to_protobuf(),
            "transaction": _helpers.get_transaction_id(transaction),
        }
        kwargs = _helpers.make_retry_timeout_kwargs(retry, timeout)

        return request, expected_prefix, kwargs

    def stream(
        self, transaction=None, retry: retries.Retry = None, timeout: float = None,
    ) -> Generator[document.DocumentSnapshot, Any, None]:
        raise NotImplementedError

    def on_snapshot(self, callback) -> NoReturn:
        raise NotImplementedError

    def recursive(self) -> "BaseQuery":
        """Returns a copy of this query whose iterator will yield all matching
        documents as well as each of their descendent subcollections and documents.

        This differs from the `all_descendents` flag, which only returns descendents
        whose subcollection names match the parent collection's name. To return
        all descendents, regardless of their subcollection name, use this.
        """
        copied = self._copy(recursive=True, all_descendants=True)
        if copied._parent and copied._parent.id:
            original_collection_id = "/".join(copied._parent._path)

            # Reset the parent to nothing so we can recurse through the entire
            # database. This is required to have
            # `CollectionSelector.collection_id` not override
            # `CollectionSelector.all_descendants`, which happens if both are
            # set.
            copied._parent = copied._get_collection_reference_class()("")
            copied._parent._client = self._parent._client

            # But wait! We don't want to load the entire database; only the
            # collection the user originally specified. To accomplish that, we
            # add the following arcane filters.

            REFERENCE_NAME_MIN_ID = "__id-9223372036854775808__"
            start_at = f"{original_collection_id}/{REFERENCE_NAME_MIN_ID}"

            # The backend interprets this null character is flipping the filter
            # to mean the end of the range instead of the beginning.
            nullChar = "\0"
            end_at = f"{original_collection_id}{nullChar}/{REFERENCE_NAME_MIN_ID}"

            copied = (
                copied.order_by(field_path_module.FieldPath.document_id())
                .start_at({field_path_module.FieldPath.document_id(): start_at})
                .end_at({field_path_module.FieldPath.document_id(): end_at})
            )

        return copied

    def _comparator(self, doc1, doc2) -> int:
        _orders = self._orders

        # Add implicit sorting by name, using the last specified direction.
        if len(_orders) == 0:
            lastDirection = BaseQuery.ASCENDING
        else:
            if _orders[-1].direction == 1:
                lastDirection = BaseQuery.ASCENDING
            else:
                lastDirection = BaseQuery.DESCENDING

        orderBys = list(_orders)

        order_pb = query.StructuredQuery.Order(
            field=query.StructuredQuery.FieldReference(field_path="id"),
            direction=_enum_from_direction(lastDirection),
        )
        orderBys.append(order_pb)

        for orderBy in orderBys:
            if orderBy.field.field_path == "id":
                # If ordering by docuent id, compare resource paths.
                comp = Order()._compare_to(doc1.reference._path, doc2.reference._path)
            else:
                if (
                    orderBy.field.field_path not in doc1._data
                    or orderBy.field.field_path not in doc2._data
                ):
                    raise ValueError(
                        "Can only compare fields that exist in the "
                        "DocumentSnapshot. Please include the fields you are "
                        "ordering on in your select() call."
                    )
                v1 = doc1._data[orderBy.field.field_path]
                v2 = doc2._data[orderBy.field.field_path]
                encoded_v1 = _helpers.encode_value(v1)
                encoded_v2 = _helpers.encode_value(v2)
                comp = Order().compare(encoded_v1, encoded_v2)

            if comp != 0:
                # 1 == Ascending, -1 == Descending
                return orderBy.direction * comp

        return 0 class DocumentReference(BaseDocumentReference):
    """A reference to a document in a Firestore database.

    The document may already exist or can be created by this class.

    Args:
        path (Tuple[str, ...]): The components in the document path.
            This is a series of strings representing each collection and
            sub-collection ID, as well as the document IDs for any documents
            that contain a sub-collection (as well as the base document).
        kwargs (dict): The keyword arguments for the constructor. The only
            supported keyword is ``client`` and it must be a
            :class:`~google.cloud.firestore_v1.client.Client`. It represents
            the client that created this document reference.

    Raises:
        ValueError: if

            * the ``path`` is empty
            * there are an even number of elements
            * a collection ID in ``path`` is not a string
            * a document ID in ``path`` is not a string
        TypeError: If a keyword other than ``client`` is used.
    """

    def __init__(self, *path, **kwargs) -> None:
        super(DocumentReference, self).__init__(*path, **kwargs)

    def create(
        self,
        document_data: dict,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> write.WriteResult:
        """Create a document in the Firestore database.

        >>> document_data = {"a": 1, "b": {"c": "Two"}}
        >>> document.get().to_dict() is None  # does not exist
        True
        >>> document.create(document_data)
        >>> document.get().to_dict() == document_data  # exists
        True

        Args:
            document_data (dict): Property names and values to use for
                creating a document.
            retry (google.api_core.retry.Retry): Designation of what errors, if any,
                should be retried.  Defaults to a system-specified policy.
            timeout (float): The timeout for this request.  Defaults to a
                system-specified value.

        Returns:
            :class:`~google.cloud.firestore_v1.types.WriteResult`:
                The write result corresponding to the committed document.
                A write result contains an ``update_time`` field.

        Raises:
            :class:`google.cloud.exceptions.Conflict`:
                If the document already exists.
        """
        batch, kwargs = self._prep_create(document_data, retry, timeout)
        write_results = batch.commit(**kwargs)
        return _first_write_result(write_results)

    def set(
        self,
        document_data: dict,
        merge: bool = False,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> write.WriteResult:
        """Create / replace / merge a document in the Firestore database.

        - To "upsert" a document (create if it doesn't exist, replace completely
          if it does), leave the ``merge`` argument at its default:

          >>> document_data = {"a": 1, "b": {"c": "Two"}}
          >>> document.get().to_dict() is None  # document exists
          False
          >>> document.set(document_data)
          >>> document.get().to_dict() == document_data  # exists
          True

        - To "merge" ``document_data`` with an existing document (creating if
          the document does not exist), pass ``merge`` as True``:

          >>> document_data = {"a": 1, "b": {"c": "Two"}}
          >>> document.get().to_dict() == {"d": "Three", "b": {}} # exists
          >>> document.set(document_data, merge=True)
          >>> document.get().to_dict() == {"a": 1, "d": "Three", "b": {"c": "Two"}}
          True

          In this case, existing documents with top-level keys which are
          not present in ``document_data`` (``"d"``) will preserve the values
          of those keys.


        - To merge only specific fields of ``document_data`` with existing
          documents (creating if the document does not exist), pass ``merge``
          as a list of field paths:


          >>> document_data = {"a": 1, "b": {"c": "Two"}}
          >>> document.get().to_dict() == {"b": {"c": "One", "d": "Four" }} # exists
          True
          >>> document.set(document_data, merge=["b.c"])
          >>> document.get().to_dict() == {"b": {"c": "Two", "d": "Four" }}
          True

          For more information on field paths, see
          :meth:`~google.cloud.firestore_v1.base_client.BaseClient.field_path`.

        Args:
            document_data (dict): Property names and values to use for
                replacing a document.
            merge (Optional[bool] or Optional[List<fieldpath>]):
                If True, apply merging instead of overwriting the state
                of the document.
            retry (google.api_core.retry.Retry): Designation of what errors, if any,
                should be retried.  Defaults to a system-specified policy.
            timeout (float): The timeout for this request.  Defaults to a
                system-specified value.

        Returns:
            :class:`~google.cloud.firestore_v1.types.WriteResult`:
            The write result corresponding to the committed document. A write
            result contains an ``update_time`` field.
        """
        batch, kwargs = self._prep_set(document_data, merge, retry, timeout)
        write_results = batch.commit(**kwargs)
        return _first_write_result(write_results)

    def update(
        self,
        field_updates: dict,
        option: _helpers.WriteOption = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> write.WriteResult:
        """Update an existing document in the Firestore database.

        By default, this method verifies that the document exists on the
        server before making updates. A write ``option`` can be specified to
        override these preconditions.

        Each key in ``field_updates`` can either be a field name or a
        **field path** (For more information on field paths, see
        :meth:`~google.cloud.firestore_v1.base_client.BaseClient.field_path`.)
        To illustrate this, consider a document with

        .. code-block:: python

           >>> snapshot = document.get()
           >>> snapshot.to_dict()
           {
               'foo': {
                   'bar': 'baz',
               },
               'other': True,
           }

        stored on the server. If the field name is used in the update:

        .. code-block:: python

           >>> field_updates = {
           ...     'foo': {
           ...         'quux': 800,
           ...     },
           ... }
           >>> document.update(field_updates)

        then all of ``foo`` will be overwritten on the server and the new
        value will be

        .. code-block:: python

           >>> snapshot = document.get()
           >>> snapshot.to_dict()
           {
               'foo': {
                   'quux': 800,
               },
               'other': True,
           }

        On the other hand, if a ``.``-delimited **field path** is used in the
        update:

        .. code-block:: python

           >>> field_updates = {
           ...     'foo.quux': 800,
           ... }
           >>> document.update(field_updates)

        then only ``foo.quux`` will be updated on the server and the
        field ``foo.bar`` will remain intact:

        .. code-block:: python

           >>> snapshot = document.get()
           >>> snapshot.to_dict()
           {
               'foo': {
                   'bar': 'baz',
                   'quux': 800,
               },
               'other': True,
           }

        .. warning::

           A **field path** can only be used as a top-level key in
           ``field_updates``.

        To delete / remove a field from an existing document, use the
        :attr:`~google.cloud.firestore_v1.transforms.DELETE_FIELD` sentinel.
        So with the example above, sending

        .. code-block:: python

           >>> field_updates = {
           ...     'other': firestore.DELETE_FIELD,
           ... }
           >>> document.update(field_updates)

        would update the value on the server to:

        .. code-block:: python

           >>> snapshot = document.get()
           >>> snapshot.to_dict()
           {
               'foo': {
                   'bar': 'baz',
               },
           }

        To set a field to the current time on the server when the
        update is received, use the
        :attr:`~google.cloud.firestore_v1.transforms.SERVER_TIMESTAMP`
        sentinel.
        Sending

        .. code-block:: python

           >>> field_updates = {
           ...     'foo.now': firestore.SERVER_TIMESTAMP,
           ... }
           >>> document.update(field_updates)

        would update the value on the server to:

        .. code-block:: python

           >>> snapshot = document.get()
           >>> snapshot.to_dict()
           {
               'foo': {
                   'bar': 'baz',
                   'now': datetime.datetime(2012, ...),
               },
               'other': True,
           }

        Args:
            field_updates (dict): Field names or paths to update and values
                to update with.
            option (Optional[:class:`~google.cloud.firestore_v1.client.WriteOption`]):
                A write option to make assertions / preconditions on the server
                state of the document before applying changes.
            retry (google.api_core.retry.Retry): Designation of what errors, if any,
                should be retried.  Defaults to a system-specified policy.
            timeout (float): The timeout for this request.  Defaults to a
                system-specified value.

        Returns:
            :class:`~google.cloud.firestore_v1.types.WriteResult`:
            The write result corresponding to the updated document. A write
            result contains an ``update_time`` field.

        Raises:
            :class:`google.cloud.exceptions.NotFound`:
                If the document does not exist.
        """
        batch, kwargs = self._prep_update(field_updates, option, retry, timeout)
        write_results = batch.commit(**kwargs)
        return _first_write_result(write_results)

    def delete(
        self,
        option: _helpers.WriteOption = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> Timestamp:
        """Delete the current document in the Firestore database.

        Args:
            option (Optional[:class:`~google.cloud.firestore_v1.client.WriteOption`]):
                A write option to make assertions / preconditions on the server
                state of the document before applying changes.
            retry (google.api_core.retry.Retry): Designation of what errors, if any,
                should be retried.  Defaults to a system-specified policy.
            timeout (float): The timeout for this request.  Defaults to a
                system-specified value.

        Returns:
            :class:`google.protobuf.timestamp_pb2.Timestamp`:
            The time that the delete request was received by the server.
            If the document did not exist when the delete was sent (i.e.
            nothing was deleted), this method will still succeed and will
            still return the time that the request was received by the server.
        """
        request, kwargs = self._prep_delete(option, retry, timeout)

        commit_response = self._client._firestore_api.commit(
            request=request, metadata=self._client._rpc_metadata, **kwargs,
        )

        return commit_response.commit_time

    def get(
        self,
        field_paths: Iterable[str] = None,
        transaction=None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> DocumentSnapshot:
        """Retrieve a snapshot of the current document.

        See :meth:`~google.cloud.firestore_v1.base_client.BaseClient.field_path` for
        more information on **field paths**.

        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).

        Args:
            field_paths (Optional[Iterable[str, ...]]): An iterable of field
                paths (``.``-delimited list of field names) to use as a
                projection of document fields in the returned results. If
                no value is provided, all fields will be returned.
            transaction (Optional[:class:`~google.cloud.firestore_v1.transaction.Transaction`]):
                An existing transaction that this reference
                will be retrieved in.
            retry (google.api_core.retry.Retry): Designation of what errors, if an                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      y,
                should be retried.  Defaults to a system-specified policy.
            timeout (float): The timeout for this request.  Defaults to a
                system-specified value.

        Returns:
            :class:`~google.cloud.firestore_v1.base_document.DocumentSnapshot`:
                A snapshot of the current document. If the document does not
                exist at the time of the snapshot is taken, the snapshot's
                :attr:`reference`, :attr:`data`, :attr:`update_time`, and
                :attr:`create_time` attributes will all be ``None`` and
                its :attr:`exists` attribute will be ``False``.
        """
        from google.cloud.firestore_v1.base_client import _parse_batch_get

        request, kwargs = self._prep_batch_get(field_paths, transaction, retry, timeout)

        response_iter = self._client._firestore_api.batch_get_documents(
            request=request, metadata=self._client._rpc_metadata, **kwargs,
        )

        get_doc_response = next(response_iter, None)

        if get_doc_response is not None:
            return _parse_batch_get(
                get_doc_response=get_doc_response,
                reference_map={self._document_path: self},
                client=self._client,
            )

        logger.warning(
            "`batch_get_documents` unexpectedly returned empty "
            "stream. Expected one object.",
        )

        return DocumentSnapshot(
            self,
            None,
            exists=False,
            read_time=_datetime_to_pb_timestamp(datetime.datetime.now()),
            create_time=None,
            update_time=None,
        )

    def collections(
        self,
        page_size: int = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> Generator[Any, Any, None]:
        """List subcollections of the current document.

        Args:
            page_size (Optional[int]]): The maximum number of collections
                in each page of results from this request. Non-positive values
                are ignored. Defaults to a sensible value set by the API.
            retry (google.api_core.retry.Retry): Designation of what errors, if any,
                should be retried.  Defaults to a system-specified policy.
            timeout (float): The timeout for this request.  Defaults to a
                system-specified value.

        Returns:
            Sequence[:class:`~google.cloud.firestore_v1.collection.CollectionReference`]:
                iterator of subcollections of the current document. If the
                document does not exist at the time of `snapshot`, the
                iterator will be empty
        """
        request, kwargs = self._prep_collections(page_size, retry, timeout)

        iterator = self._client._firestore_api.list_collection_ids(
            request=request, metadata=self._client._rpc_metadata, **kwargs,
        )

        for collection_id in iterator:
            yield self.collection(collection_id)

    def on_snapshot(self, callback: Callable) -> Watch:
        """Watch this document.

        This starts a watch on this document using a background thread. The
        provided callback is run on the snapshot.

        Args:
            callback(Callable[[:class:`~google.cloud.firestore_v1.base_document.DocumentSnapshot`], NoneType]):
                a callback to run when a change occurs

        Example:

        .. code-block:: python

            from google.cloud import firestore_v1

            db = firestore_v1.Client()
            collection_ref = db.collection(u'users')

            def on_snapshot(document_snapshot, changes, read_time):
                doc = document_snapshot
                print(u'{} => {}'.format(doc.id, doc.to_dict()))

            doc_ref = db.collection(u'users').document(
                u'alovelace' + unique_resource_id())

            # Watch this document
            doc_watch = doc_ref.on_snapshot(on_snapshot)

            # Terminate this watch
            doc_watch.unsubscribe()
        """
        return Watch.for_document(self, callback, DocumentSnapshot) _GRPC_ERROR_MAPPING: dict
BAD_PATH_TEMPLATE = "A path element must be a string. Received {}, which is a {}."
DOCUMENT_PATH_DELIMITER = "/"
INACTIVE_TXN = "Transaction not in progress, cannot be used in API requests."
READ_AFTER_WRITE_ERROR = "Attempted read after write in a transaction."
BAD_REFERENCE_ERROR = (
    "Reference value {!r} in unexpected format, expected to be of the form "
    "``projects/{{project}}/databases/{{database}}/"
    "documents/{{document_path}}``."
)
WRONG_APP_REFERENCE = (
    "Document {!r} does not correspond to the same database " "({!r}) as the client."
)
REQUEST_TIME_ENUM = DocumentTransform.FieldTransform.ServerValue.REQUEST_TIME
_GRPC_ERROR_MAPPING = {
    grpc.StatusCode.ALREADY_EXISTS: exceptions.Conflict,
    grpc.StatusCode.NOT_FOUND: exceptions.NotFound,
}
class GeoPoint(object):
class DocumentReferenceValue:
class DocumentExtractor(object):
class DocumentExtractorForMerge(DocumentExtractor):
class DocumentExtractorForUpdate(DocumentExtractor):
class ReadAfterWriteError(Exception):
class WriteOption(object):
class LastUpdateOption(WriteOption):
class ExistsOption(WriteOption):
    def __init__(self, latitude, longitude) -> None:
    def to_protobuf(self) -> latlng_pb2.LatLng:
    def __eq__(self, other):
    def __ne__(self, other):
def verify_path(path, is_collection) -> None:
def encode_value(value) -> types.document.Value:
def encode_dict(values_dict) -> dict:
def document_snapshot_to_protobuf(snapshot: "google.cloud.firestore_v1.base_document.DocumentSnapshot") -> Optional["google.cloud.firestore_v1.types.Document"]:  # type: ignore
    def __init__(self, reference_value: str):
    def full_key(self) -> str:
    def full_path(self) -> str:
def reference_value_to_document(reference_value, client) -> Any:
def decode_value(
    value, client
) -> Union[None, bool, int, float, list, datetime.datetime, str, bytes, dict, GeoPoint]:
def decode_dict(value_fields, client) -> dict:
def get_doc_id(document_pb, expected_prefix) -> str:
def extract_fields(
    document_data, prefix_path: FieldPath, expand_dots=False
) -> Generator[Tuple[Any, Any], Any, None]:
def set_field_value(document_data, field_path, value) -> None:
def get_field_value(document_data, field_path) -> Any:
    def __init__(self, document_data) -> None:
    def _get_document_iterator(
        self, prefix_path: FieldPath
    ) -> Generator[Tuple[Any, Any], Any, None]:
    def has_transforms(self):
    def transform_paths(self):
    def _get_update_mask(self, allow_empty_mask=False) -> None:
    def get_update_pb(
        self, document_path, exists=None, allow_empty_mask=False
    ) -> types.write.Write:
    def get_field_transform_pbs(
        self, document_path
    ) -> List[types.write.DocumentTransform.FieldTransform]:
        def make_array_value(values):
    def get_transform_pb(self, document_path, exists=None) -> types.write.Write:
def pbs_for_create(document_path, document_data) -> List[types.write.Write]:
def pbs_for_set_no_merge(document_path, document_data) -> List[types.write.Write]:
    def __init__(self, document_data) -> None:
    def _apply_merge_all(self) -> None:
    def _construct_merge_paths(self, merge) -> Generator[Any, Any, None]:
    def _normalize_merge_paths(self, merge) -> list:
    def _apply_merge_paths(self, merge) -> None:
    def apply_merge(self, merge) -> None:
    def _get_update_mask(
        self, allow_empty_mask=False
    ) -> Optional[types.common.DocumentMask]:
def pbs_for_set_with_merge(
    document_path, document_data, merge
) -> List[types.write.Write]:
    def __init__(self, document_data) -> None:
    def _get_document_iterator(
        self, prefix_path: FieldPath
    ) -> Generator[Tuple[Any, Any], Any, None]:
    def _get_update_mask(self, allow_empty_mask=False) -> types.common.DocumentMask:
def pbs_for_update(document_path, field_updates, option) -> List[types.write.Write]:
def pb_for_delete(document_path, option) -> types.write.Write:
def get_transaction_id(transaction, read_operation=True) -> Union[bytes, None]:
def metadata_with_prefix(prefix: str, **kw) -> List[Tuple[str, str]]:
    def modify_write(self, write, no_create_msg=None) -> NoReturn:
    def __init__(self, last_update_time) -> None:
    def __eq__(self, other):
    def modify_write(self, write, **unused_kwargs) -> None:
    def __init__(self, exists) -> None:
    def __eq__(self, other):
    def modify_write(self, write, **unused_kwargs) -> None:
def make_retry_timeout_kwargs(retry, timeout) -> dict:
def build_timestamp(
    dt: Optional[Union[DatetimeWithNanoseconds, datetime.datetime]] = None
) -> Timestamp:
def compare_timestamps(
    ts1: Union[Timestamp, datetime.datetime], ts2: Union[Timestamp, datetime.datetime],
) -> int:
def deserialize_bundle(
    serialized: Union[str, bytes],
    client: "google.cloud.firestore_v1.client.BaseClient",  # type: ignore
) -> "google.cloud.firestore_bundle.FirestoreBundle":  # type: ignore
def _parse_bundle_elements_data(serialized: Union[str, bytes]) -> Generator[Dict, None, None]:  # type: ignore
def _get_documents_from_bundle(
    bundle, *, query_name: Optional[str] = None
) -> Generator["google.cloud.firestore.DocumentSnapshot", None, None]:  # type: ignore
def _get_document_from_bundle(
    bundle, *, document_id: str,
) -> Optional["google.cloud.firestore.DocumentSnapshot"]:  # type: ignore
import datetime
import json
            import asyncio
from google.cloud.firestore_bundle.types.bundle import (
    BundledDocumentMetadata,
    BundledQuery,
    BundleElement,
    BundleMetadata,
    NamedQuery,
)
from google.cloud._helpers import _datetime_to_pb_timestamp, UTC  # type: ignore
from google.cloud.firestore_bundle._helpers import limit_type_of_query
from google.cloud.firestore_v1.async_query import AsyncQuery
from google.cloud.firestore_v1.base_client import BaseClient
from google.cloud.firestore_v1.base_document import DocumentSnapshot
from google.cloud.firestore_v1.base_query import BaseQuery
from google.cloud.firestore_v1.document import DocumentReference
from google.cloud.firestore_v1 import _helpers
from google.protobuf.timestamp_pb2 import Timestamp  # type: ignore
from google.protobuf import json_format  # type: ignore
from typing import (
    Dict,
    List,
    Optional,
    Union,
)
        from google.cloud.firestore_v1.types.document import Document
        # if `_ts` is greater than `self.latest_read_time`
        if _helpers.compare_timestamps(_ts, self.latest_read_time) == 1:
            self.latest_read_time = _ts

    def _add_bundle_element(self, bundle_element: BundleElement, *, client: BaseClient, type: str):  # type: ignore
        """Applies BundleElements to this FirestoreBundle instance as a part of
        deserializing a FirestoreBundle string.
        """

        if getattr(self, "_doc_metadata_map", None) is None:
            self._doc_metadata_map = {}
        if type == "metadata":
            self._deserialized_metadata = bundle_element.metadata  # type: ignore
        elif type == "namedQuery":
            self.named_queries[bundle_element.named_query.name] = bundle_element.named_query  # type: ignore
        elif type == "documentMetadata":
            self._doc_metadata_map[
                bundle_element.document_metadata.name
            ] = bundle_element.document_metadata
        elif type == "document":
            doc_ref_value = _helpers.DocumentReferenceValue(
                bundle_element.document.name
            )
            snapshot = DocumentSnapshot(
                data=_helpers.decode_dict(
                    Document(mapping=bundle_element.document).fields, client
                ),
                exists=True,
                reference=DocumentReference(
                    doc_ref_value.collection_name,
                    doc_ref_value.document_id,
                    client=client,
                ),
                read_time=self._doc_metadata_map[
                    bundle_element.document.name
                ].read_time,
                create_time=bundle_element.document.create_time,  # type: ignore
                update_time=bundle_element.document.update_time,  # type: ignore
            )
            self.add_document(snapshot)

            bundled_document = self.documents.get(snapshot.reference._document_path)
            for query_name in self._doc_metadata_map[
                bundle_element.document.name
            ].queries:
                bundled_document.metadata.queries.append(query_name)  # type: ignore
        else:
            raise ValueError(f"Unexpected type of BundleElement: {type}")

    def build(self) -> str:
        """Iterates over the bundle's stored documents and queries and produces
        a single length-prefixed json string suitable for long-term storage.

        Example:

        .. code-block:: python

            from google.cloud import firestore

            db = firestore.Client()
            collection_ref = db.collection(u'users')

            bundle = firestore.FirestoreBundle('my bundle')
            bundle.add_named_query('app-users', collection_ref._query())

            serialized_bundle: str = bundle.build()

            # Now upload `serialized_bundle` to Google Cloud Storage, store it
            # in Memorystore, or any other storage solution.

        Returns:
            str: The length-prefixed string representation of this bundle'
                contents.
        """
        buffer: str = ""

        named_query: NamedQuery
        for named_query in self.named_queries.values():
            buffer += self._compile_bundle_element(
                BundleElement(named_query=named_query)
            )

        bundled_document: "_BundledDocument"  # type: ignore
        document_count: int = 0
        for bundled_document in self.documents.values():
            buffer += self._compile_bundle_element(
                BundleElement(document_metadata=bundled_document.metadata)
            )
            document_count += 1
            buffer += self._compile_bundle_element(
                BundleElement(document=bundled_document.snapshot._to_protobuf()._pb,)
            )

        metadata: BundleElement = BundleElement(
            metadata=self._deserialized_metadata
            or BundleMetadata(
                id=self.name,
                create_time=_helpers.build_timestamp(),
                version=FirestoreBundle.BUNDLE_SCHEMA_VERSION,
                total_documents=document_count,
                total_bytes=len(buffer.encode("utf-8")),
            )
        )
        return f"{self._compile_bundle_element(metadata)}{buffer}"

    def _compile_bundle_element(self, bundle_element: BundleElement) -> str:
        serialized_be = json.dumps(json_format.MessageToDict(bundle_element._pb))
        return f"{len(serialized_be)}{serialized_be}"

    def _reset_metadata(self):
        """Hydrating bundles stores cached data we must reset anytime new
        queries or documents are added"""
        self._deserialized_metadata = None


class _BundledDocument:
    """Convenience class to hold both the metadata and the actual content
    of a document to be bundled."""

    def __init__(

        self, snapshot: DocumentSnapshot, metadata: BundledDocumentMetadata,