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 DEFAULT_DATABASE = "(default)" _CLIENT_INFO: Any = client_info.ClientInfo(client_library_version=__version__) def _parse_batch_get(
    get_doc_response: types.BatchGetDocumentsResponse,
    reference_map: dict,
    client: BaseClient,
) -> DocumentSnapshot:
    """Parse a `BatchGetDocumentsResponse` protobuf.

    Args:
        get_doc_response (~google.cloud.proto.firestore.v1.\
            firestore.BatchGetDocumentsResponse): A single response (from
            a stream) containing the "get" response for a document.
        reference_map (Dict[str, .DocumentReference]): A mapping (produced
            by :func:`_reference_info`) of fully-qualified document paths to
            document references.
        client (:class:`~google.cloud.firestore_v1.client.Client`):
            A client that has a document factory.

    Returns:
       [.DocumentSnapshot]: The retrieved snapshot.

    Raises:
        ValueError: If the response has a ``result`` field (a oneof) other
            than ``found`` or ``missing``.
    """
    result_type = get_doc_response._pb.WhichOneof("result")
    if result_type == "found":
        reference = _get_reference(get_doc_response.found.name, reference_map)
        data = _helpers.decode_dict(get_doc_response.found.fields, client)
        snapshot = DocumentSnapshot(
            reference,
            data,
            exists=True,
            read_time=get_doc_response.read_time,
            create_time=get_doc_response.found.create_time,
            update_time=get_doc_response.found.update_time,
        )
    elif result_type == "missing":
        reference = _get_reference(get_doc_response.missing, reference_map)
        snapshot = DocumentSnapshot(
            reference,
            None,
            exists=False,
            read_time=get_doc_response.read_time,
            create_time=None,
            update_time=None,
        )
    else:
        raise ValueError(
            "`BatchGetDocumentsResponse.result` (a oneof) had a field other "
            "than `found` or `missing` set, or was unset"
        )
    return snapshot def _path_helper(path: tuple) -> Tuple[str]:
    """Standardize path into a tuple of path segments.

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

            * A single ``/``-delimited path
            * A tuple of path segments
    """
    if len(path) == 1:
        return path[0].split(_helpers.DOCUMENT_PATH_DELIMITER)
    else:
        return path class AsyncCollectionGroup(AsyncQuery, BaseCollectionGroup):
    """Represents a Collection Group in the Firestore API.

    This is a specialization of :class:`.AsyncQuery` that includes all documents in the
    database that are contained in a collection or subcollection of the given
    parent.

    Args:
        parent (:class:`~google.cloud.firestore_v1.collection.CollectionReference`):
            The collection that this query applies to.
    """

    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=True,
        recursive=False,
    ) -> None:
        super(AsyncCollectionGroup, 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,
        )

    @staticmethod
    def _get_query_class():
        return AsyncQuery

    async def get_partitions(
        self,
        partition_count,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> AsyncGenerator[QueryPartition, None]:
        """Partition a query for parallelization.

        Partitions a query by returning partition cursors that can be used to run the
        query in parallel. The returned partition cursors are split points that can be
        used as starting/end points for the query results.

        Args:
            partition_count (int): The desired maximum number of partition points. The
                number must be strictly positive. The actual number of partitions
                returned may be fewer.
            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.
        """
        request, kwargs = self._prep_get_partitions(partition_count, retry, timeout)
        pager = await self._client._firestore_api.partition_query(
            request=request, metadata=self._client._rpc_metadata, **kwargs,
        )

        start_at = None
        async for cursor_pb in pager:
            cursor = self._client.document(cursor_pb.values[0].reference_value)
            yield QueryPartition(self, start_at, cursor)
            start_at = cursor

        yield QueryPartition(self, start_at, None) class AsyncWriteBatch(BaseWriteBatch):
    """Accumulate write operations to be sent in a batch.

    This has the same set of methods for write operations that
    :class:`~google.cloud.firestore_v1.async_document.AsyncDocumentReference` does,
    e.g. :meth:`~google.cloud.firestore_v1.async_document.AsyncDocumentReference.create`.

    Args:
        client (:class:`~google.cloud.firestore_v1.async_client.AsyncClient`):
            The client that created this batch.
    """

    def __init__(self, client) -> None:
        super(AsyncWriteBatch, self).__init__(client=client)

    async def commit(
        self, retry: retries.Retry = gapic_v1.method.DEFAULT, timeout: float = None,
    ) -> list:
        """Commit the changes accumulated in this batch.

        Args:
            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:
            List[:class:`google.cloud.proto.firestore.v1.write.WriteResult`, ...]:
            The write results corresponding to the changes committed, returned
            in the same order as the changes were applied to this batch. A
            write result contains an ``update_time`` field.
        """
        request, kwargs = self._prep_commit(retry, timeout)

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

        self._write_pbs = []
        self.write_results = results = list(commit_response.write_results)
        self.commit_time = commit_response.commit_time

        return results

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        if exc_type is None:
            await self.commit() class AsyncCollectionReference(BaseCollectionReference):
    """A reference to a collection in a Firestore database.

    The collection may already exist or this class can facilitate creation
    of documents within the collection.

    Args:
        path (Tuple[str, ...]): The components in the collection 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.
        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` if provided. It
            represents the client that created this collection 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(AsyncCollectionReference, self).__init__(*path, **kwargs)

    def _query(self) -> async_query.AsyncQuery:
        """Query factory.

        Returns:
            :class:`~google.cloud.firestore_v1.query.Query`
        """
        return async_query.AsyncQuery(self)

    async def _chunkify(self, chunk_size: int):
        async for page in self._query()._chunkify(chunk_size):
            yield page

    async def add(
        self,
        document_data: dict,
        document_id: str = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> Tuple[Any, Any]:
        """Create a document in the Firestore database with the provided data.

        Args:
            document_data (dict): Property names and values to use for
                creating the document.
            document_id (Optional[str]): The document identifier within the
                current collection. If not provided, an ID will be
                automatically assigned by the server (the assigned ID will be
                a random 20 character string composed of digits,
                uppercase and lowercase letters).
            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:
            Tuple[:class:`google.protobuf.timestamp_pb2.Timestamp`, \
                :class:`~google.cloud.firestore_v1.async_document.AsyncDocumentReference`]:
                Pair of

                * The ``update_time`` when the document was created/overwritten.
                * A document reference for the created document.

        Raises:
            :class:`google.cloud.exceptions.Conflict`:
                If ``document_id`` is provided and the document already exists.
        """
        document_ref, kwargs = self._prep_add(
            document_data, document_id, retry, timeout,
        )
        write_result = await document_ref.create(document_data, **kwargs)
        return write_result.update_time, document_ref

    def document(
        self, document_id: str = None
    ) -> async_document.AsyncDocumentReference:
        """Create a sub-document underneath the current collection.

        Args:
            document_id (Optional[str]): The document identifier
                within the current collection. If not provided, will default
                to a random 20 character string composed of digits,
                uppercase and lowercase and letters.

        Returns:
            :class:`~google.cloud.firestore_v1.document.async_document.AsyncDocumentReference`:
            The child document.
        """
        return super(AsyncCollectionReference, self).document(document_id)

    async def list_documents(
        self,
        page_size: int = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> AsyncGenerator[DocumentReference, None]:
        """List all subdocuments of the current collection.

        Args:
            page_size (Optional[int]]): The maximum number of documents
                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.DocumentReference`]:
                iterator of subdocuments of the current collection. If the
                collection does not exist at the time of `snapshot`, the
                iterator will be empty
        """
        request, kwargs = self._prep_list_documents(page_size, retry, timeout)

        iterator = await self._client._firestore_api.list_documents(
            request=request, metadata=self._client._rpc_metadata, **kwargs,
        )
        async for i in iterator:
            yield _item_to_document_ref(self, i)

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

        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 this collection that match the query.
        """
        query, kwargs = self._prep_get_or_stream(retry, timeout)

        return await query.get(transaction=transaction, **kwargs)

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

        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 the 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.document.DocumentSnapshot`:
            The next document that fulfills the query.
        """
        query, kwargs = self._prep_get_or_stream(retry, timeout)

        async for d in query.stream(transaction=transaction, **kwargs):
            yield d  # pytype: disable=name-error class AsyncDocumentReference(BaseDocumentReference):
    def __init__(self, *path, **kwargs) -> None:
    async def create(
        self,
        document_data: dict,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> write.WriteResult:
    async def set(
        self,
        document_data: dict,
        merge: bool = False,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> write.WriteResult:
    async def update(
        self,
        field_updates: dict,
        option: _helpers.WriteOption = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> write.WriteResult:
    async def delete(
        self,
        option: _helpers.WriteOption = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> Timestamp:
    async def get(
        self,
        field_paths: Iterable[str] = None,
        transaction=None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> Union[DocumentSnapshot, Coroutine[Any, Any, DocumentSnapshot]]:
    async def collections(
        self,
        page_size: int = None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> AsyncGenerator: class AsyncTransaction(async_batch.AsyncWriteBatch, BaseTransaction):
    """Accumulate read-and-write operations to be sent in a transaction.

    Args:
        client (:class:`~google.cloud.firestore_v1.client.Client`):
            The client that created this transaction.
        max_attempts (Optional[int]): The maximum number of attempts for
            the transaction (i.e. allowing retries). Defaults to
            :attr:`~google.cloud.firestore_v1.transaction.MAX_ATTEMPTS`.
        read_only (Optional[bool]): Flag indicating if the transaction
            should be read-only or should allow writes. Defaults to
            :data:`False`.
    """

    def __init__(self, client, max_attempts=MAX_ATTEMPTS, read_only=False) -> None:
        super(AsyncTransaction, self).__init__(client)
        BaseTransaction.__init__(self, max_attempts, read_only)

    def _add_write_pbs(self, write_pbs: list) -> None:
        """Add `Write`` protobufs to this transaction.

        Args:
            write_pbs (List[google.cloud.proto.firestore.v1.\
                write.Write]): A list of write protobufs to be added.

        Raises:
            ValueError: If this transaction is read-only.
        """
        if self._read_only:
            raise ValueError(_WRITE_READ_ONLY)

        super(AsyncTransaction, self)._add_write_pbs(write_pbs)

    async def _begin(self, retry_id: bytes = None) -> None:
        """Begin the transaction.

        Args:
            retry_id (Optional[bytes]): Transaction ID of a transaction to be
                retried.

        Raises:
            ValueError: If the current transaction has already begun.
        """
        if self.in_progress:
            msg = _CANT_BEGIN.format(self._id)
            raise ValueError(msg)

        transaction_response = await self._client._firestore_api.begin_transaction(
            request={
                "database": self._client._database_string,
                "options": self._options_protobuf(retry_id),
            },
            metadata=self._client._rpc_metadata,
        )
        self._id = transaction_response.transaction

    async def _rollback(self) -> None:
        """Roll back the transaction.

        Raises:
            ValueError: If no transaction is in progress.
        """
        if not self.in_progress:
            raise ValueError(_CANT_ROLLBACK)

        try:
            # NOTE: The response is just ``google.protobuf.Empty``.
            await self._client._firestore_api.rollback(
                request={
                    "database": self._client._database_string,
                    "transaction": self._id,
                },
                metadata=self._client._rpc_metadata,
            )
        finally:
            self._clean_up()

    async def _commit(self) -> list:
        """Transactionally commit the changes accumulated.

        Returns:
            List[:class:`google.cloud.proto.firestore.v1.write.WriteResult`, ...]:
            The write results corresponding to the changes committed, returned
            in the same order as the changes were applied to this transaction.
            A write result contains an ``update_time`` field.

        Raises:
            ValueError: If no transaction is in progress.
        """
        if not self.in_progress:
            raise ValueError(_CANT_COMMIT)

        commit_response = await _commit_with_retry(
            self._client, self._write_pbs, self._id
        )

        self._clean_up()
        return list(commit_response.write_results)

    async def get_all(
        self,
        references: list,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> AsyncGenerator[DocumentSnapshot, Any]:
        """Retrieves multiple documents from Firestore.

        Args:
            references (List[.AsyncDocumentReference, ...]): Iterable of document
                references to be retrieved.
            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:
            .DocumentSnapshot: The next document snapshot that fulfills the
            query, or :data:`None` if the document does not exist.
        """
        kwargs = _helpers.make_retry_timeout_kwargs(retry, timeout)
        return await self._client.get_all(references, transaction=self, **kwargs)

    async def get(
        self,
        ref_or_query,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> AsyncGenerator[DocumentSnapshot, Any]:
        """
        Retrieve a document or a query result from the database.

        Args:
            ref_or_query The document references or query object to return.
            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:
            .DocumentSnapshot: The next document snapshot that fulfills the
            query, or :data:`None` if the document does not exist.
        """
        kwargs = _helpers.make_retry_timeout_kwargs(retry, timeout)
        if isinstance(ref_or_query, AsyncDocumentReference):
            return await self._client.get_all(
                [ref_or_query], transaction=self, **kwargs
            )
        elif isinstance(ref_or_query, AsyncQuery):
            return await ref_or_query.stream(transaction=self, **kwargs)
        else:
            raise ValueError(
                'Value for argument "ref_or_query" must be a AsyncDocumentReference or a AsyncQuery.'
            ) class FieldPath(object):
    """Field Path object for client use.

    A field path is a sequence of element keys, separated by periods.
    Each element key can be either a simple identifier, or a full unicode
    string.

    In the string representation of a field path, non-identifier elements
    must be quoted using backticks, with internal backticks and backslashes
    escaped with a backslash.

    Args:
        parts: (one or more strings)
            Indicating path of the key to be used.
    """

    def __init__(self, *parts):
        for part in parts:
            if not isinstance(part, str) or not part:
                error = "One or more components is not a string or is empty."
                raise ValueError(error)
        self.parts = tuple(parts)

    @classmethod
    def from_api_repr(cls, api_repr: str):
        """Factory: create a FieldPath from the string formatted per the API.

        Args:
            api_repr (str): a string path, with non-identifier elements quoted
            It cannot exceed 1500 characters, and cannot be empty.
        Returns:
            (:class:`FieldPath`) An instance parsed from ``api_repr``.
        Raises:
            ValueError if the parsing fails
        """
        api_repr = api_repr.strip()
        if not api_repr:
            raise ValueError("Field path API representation cannot be empty.")
        return cls(*parse_field_path(api_repr))

    @classmethod
    def from_string(cls, path_string: str):
        """Factory: create a FieldPath from a unicode string representation.

        This method splits on the character `.` and disallows the
        characters `~*/[]`. To create a FieldPath whose components have
        those characters, call the constructor.

        Args:
            path_string (str): A unicode string which cannot contain
            `~*/[]` characters, cannot exceed 1500 bytes, and cannot be empty.

        Returns:
            (:class:`FieldPath`) An instance parsed from ``path_string``.
        """
        try:
            return cls.from_api_repr(path_string)
        except ValueError:
            elements = path_string.split(".")
            for element in elements:
                if not element:
                    raise ValueError("Empty element")
                if _LEADING_ALPHA_INVALID.match(element):
                    raise ValueError(
                        "Non-alphanum char in element with leading alpha: {}".format(
                            element
                        )
                    )
            return FieldPath(*elements)

    def __repr__(self):
        paths = ""
        for part in self.parts:
            paths += "'" + part + "',"
        paths = paths[:-1]
        return "FieldPath({})".format(paths)

    def __hash__(self):
        return hash(self.to_api_repr())

    def __eq__(self, other):
        if isinstance(other, FieldPath):
            return self.parts == other.parts
        return NotImplemented

    def __lt__(self, other):
        if isinstance(other, FieldPath):
            return self.parts < other.parts
        return NotImplemented

    def __add__(self, other):
        """Adds `other` field path to end of this field path.

        Args:
            other (~google.cloud.firestore_v1._helpers.FieldPath, str):
                The field path to add to the end of this `FieldPath`.
        """
        if isinstance(other, FieldPath):
            parts = self.parts + other.parts
            return FieldPath(*parts)
        elif isinstance(other, str):
            parts = self.parts + FieldPath.from_string(other).parts
            return FieldPath(*parts)
        else:
            return NotImplemented

    def to_api_repr(self):
        """Render a quoted string representation of the FieldPath

        Returns:
            (str) Quoted string representation of the path stored
            within this FieldPath.
        """
        return render_field_path(self.parts)

    def eq_or_parent(self, other):
        """Check whether ``other`` is an ancestor.

        Returns:
            (bool) True IFF ``other`` is an ancestor or equal to ``self``,
            else False.
        """
        return self.parts[: len(other.parts)] == other.parts[: len(self.parts)]

    def lineage(self):
        """Return field paths for all parents.

        Returns: Set[:class:`FieldPath`]
        """
        indexes = range(1, len(self.parts))
        return {FieldPath(*self.parts[:index]) for index in indexes}

    @staticmethod
    def document_id():
        """A special FieldPath value to refer to the ID of a document. It can be used
           in queries to sort or filter by the document ID.

        Returns: A special sentinel value to refer to the ID of a document.
        """
        return "__name__" class FirestoreAsyncClient:
    DEFAULT_ENDPOINT = FirestoreClient.DEFAULT_ENDPOINT
    DEFAULT_MTLS_ENDPOINT = FirestoreClient.DEFAULT_MTLS_ENDPOINT
    DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
        gapic_version=pkg_resources.get_distribution("google-cloud-firestore",).version,
    )
    DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo()
    def from_service_account_info(cls, info: dict, *args, **kwargs):
    def from_service_account_file(cls, filename: str, *args, **kwargs):
    def get_mtls_endpoint_and_cert_source(
        cls, client_options: Optional[ClientOptions] = None
    ):
    def transport(self) -> FirestoreTransport:
    def __init__(
        self,
        *,
        credentials: ga_credentials.Credentials = None,
        transport: Union[str, FirestoreTransport] = "grpc_asyncio",
        client_options: ClientOptions = None,
        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
    ) -> None:
    async def get_document(
        self,
        request: Union[firestore.GetDocumentRequest, dict] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> document.Document:
    async def list_documents(
        self,
        request: Union[firestore.ListDocumentsRequest, dict] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> pagers.ListDocumentsAsyncPager:
    async def update_document(
        self,
        request: Union[firestore.UpdateDocumentRequest, dict] = None,
        *,
        document: gf_document.Document = None,
        update_mask: common.DocumentMask = None,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> gf_document.Document:
    async def delete_document(
        self,
        request: Union[firestore.DeleteDocumentRequest, dict] = None,
        *,
        name: str = None,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> None:
    def batch_get_documents(
        self,
        request: Union[firestore.BatchGetDocumentsRequest, dict] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> Awaitable[AsyncIterable[firestore.BatchGetDocumentsResponse]]:
    async def begin_transaction(
        self,
        request: Union[firestore.BeginTransactionRequest, dict] = None,
        *,
        database: str = None,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> firestore.BeginTransactionResponse:
    async def commit(
        self,
        request: Union[firestore.CommitRequest, dict] = None,
        *,
        database: str = None,
        writes: Sequence[gf_write.Write] = None,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> firestore.CommitResponse:
    async def rollback(
        self,
        request: Union[firestore.RollbackRequest, dict] = None,
        *,
        database: str = None,
        transaction: bytes = None,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> None:
    def run_query(
        self,
        request: Union[firestore.RunQueryRequest, dict] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> Awaitable[AsyncIterable[firestore.RunQueryResponse]]:
    async def partition_query(
        self,
        request: Union[firestore.PartitionQueryRequest, dict] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> pagers.PartitionQueryAsyncPager:
    def write(
        self,
        requests: AsyncIterator[firestore.WriteRequest] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> Awaitable[AsyncIterable[firestore.WriteResponse]]:
    def listen(
        self,
        requests: AsyncIterator[firestore.ListenRequest] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> Awaitable[AsyncIterable[firestore.ListenResponse]]:
    async def list_collection_ids(
        self,
        request: Union[firestore.ListCollectionIdsRequest, dict] = None,
        *,
        parent: str = None,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> pagers.ListCollectionIdsAsyncPager:
    async def batch_write(
        self,
        request: Union[firestore.BatchWriteRequest, dict] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> firestore.BatchWriteResponse:
    async def create_document(
        self,
        request: Union[firestore.CreateDocumentRequest, dict] = None,
        *,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: float = None,
        metadata: Sequence[Tuple[str, str]] = (),
    ) -> document.Document:
    async def __aenter__(self):
    async def __aexit__(self, exc_type, exc, tb): class FirestoreGrpcAsyncIOTransport(FirestoreTransport):
    def create_channel(
        cls,
        host: str = "firestore.googleapis.com",
        credentials: ga_credentials.Credentials = None,
        credentials_file: Optional[str] = None,
        scopes: Optional[Sequence[str]] = None,
        quota_project_id: Optional[str] = None,
        **kwargs,
    ) -> aio.Channel:
    def __init__(
        self,
        *,
        host: str = "firestore.googleapis.com",
        credentials: ga_credentials.Credentials = None,
        credentials_file: Optional[str] = None,
        scopes: Optional[Sequence[str]] = None,
        channel: aio.Channel = None,
        api_mtls_endpoint: str = None,
        client_cert_source: Callable[[], Tuple[bytes, bytes]] = None,
        ssl_channel_credentials: grpc.ChannelCredentials = None,
        client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
        quota_project_id=None,
        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
        always_use_jwt_access: Optional[bool] = False,
    ) -> None:
    def grpc_channel(self) -> aio.Channel:
    def get_document(
        self,
    ) -> Callable[[firestore.GetDocumentRequest], Awaitable[document.Document]]:
    def list_documents(
        self,
    ) -> Callable[
    def update_document(
        self,
    ) -> Callable[[firestore.UpdateDocumentRequest], Awaitable[gf_document.Document]]:
    def delete_document(
        self,
    ) -> Callable[[firestore.DeleteDocumentRequest], Awaitable[empty_pb2.Empty]]:
    def batch_get_documents(
        self,
    ) -> Callable[
    def begin_transaction(
        self,
    ) -> Callable[
    def commit(
        self,
    ) -> Callable[[firestore.CommitRequest], Awaitable[firestore.CommitResponse]]:
    def rollback(
        self,
    ) -> Callable[[firestore.RollbackRequest], Awaitable[empty_pb2.Empty]]:
    def run_query(
        self,
    ) -> Callable[[firestore.RunQueryRequest], Awaitable[firestore.RunQueryResponse]]:
    def partition_query(
        self,
    ) -> Callable[
    def write(
        self,
    ) -> Callable[[firestore.WriteRequest], Awaitable[firestore.WriteResponse]]:
    def listen(
        self,
    ) -> Callable[[firestore.ListenRequest], Awaitable[firestore.ListenResponse]]:
    def list_collection_ids(
        self,
    ) -> Callable[
    def batch_write(
        self,
    ) -> Callable[
    def create_document(
        self,
    ) -> Callable[[firestore.CreateDocumentRequest], Awaitable[document.Document]]:
    def close(self):
from google.api_core import gapic_v1
from google.api_core import retry as retries
from google.cloud.firestore_v1.base_client import (
    BaseClient,
    DEFAULT_DATABASE,
    _CLIENT_INFO,
    _parse_batch_get,  # type: ignore
    _path_helper,
)
from google.cloud.firestore_v1.async_query import AsyncCollectionGroup
from google.cloud.firestore_v1.async_batch import AsyncWriteBatch
from google.cloud.firestore_v1.async_collection import AsyncCollectionReference
from google.cloud.firestore_v1.async_document import (
    AsyncDocumentReference,
    DocumentSnapshot,
)
from google.cloud.firestore_v1.async_transaction import AsyncTransaction
from google.cloud.firestore_v1.field_path import FieldPath
from google.cloud.firestore_v1.services.firestore import (
    async_client as firestore_client,
)
from google.cloud.firestore_v1.services.firestore.transports import (
    grpc_asyncio as firestore_grpc_transport,
)
from typing import Any, AsyncGenerator, Iterable, List, Optional, Union, TYPE_CHECKING
    from google.cloud.firestore_v1.bulk_writer import BulkWriter  # pragma: NO COVER
        from google.cloud.firestore_v1.client import Client
        return AsyncDocumentReference(
            *self._document_path_helper(*document_path), client=self
        )

    async def get_all(
        self,
        references: List[AsyncDocumentReference],
        field_paths: Iterable[str] = None,
        transaction=None,
        retry: retries.Retry = gapic_v1.method.DEFAULT,
        timeout: float = None,
    ) -> AsyncGenerator[DocumentSnapshot, Any]:
        """Retrieve a batch of documents.

        .. note::

           Documents returned by this method are not guaranteed to be
           returned in the same order that they are given in ``references``.

        .. note::

           If multiple ``references`` refer to the same document, the server
           will only return one result.

        See :meth:`~google.cloud.firestore_v1.client.Client.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:
            references (List[.AsyncDocumentReference, ...]): Iterable of document
                references to be retrieved.
            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.async_transaction.AsyncTransaction`]):
                An existing transaction that these ``references`` will be
                retrieved 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:
            .DocumentSnapshot: The next document snapshot that fulfills the
            query, or :data:`None` if the document does not exist.
        """
        request, reference_map, kwargs = self._prep_get_all(
            references, field_paths, transaction, retry, timeout
        )

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

        async for get_doc_response in response_iterator:
            yield _parse_batch_get(get_doc_response, reference_map, self)

    async def collections(
        self, retry: retries.Retry = gapic_v1.method.DEFAULT, timeout: float = None,
    ) -> AsyncGenerator[AsyncCollectionReference, Any]:
        """List top-level collections of the client's database.

        Args:
            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.async_collection.AsyncCollectionReference`]:
                iterator of subcollections of the current document.
        """
        request, kwargs = self._prep_collections(retry, timeout)
        iterator = await self._firestore_api.list_collection_ids(
            request=request, metadata=self._rpc_metadata, **kwargs,
        )

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

    async def recursive_delete(
        self,
        reference: Union[AsyncCollectionReference, AsyncDocumentReference],
        *,
        bulk_writer: Optional["BulkWriter"] = None,
        chunk_size: Optional[int] = 5000,
    ):
        """Deletes documents and their subcollections, regardless of collection
        name.

        Passing an AsyncCollectionReference leads to each document in the
        collection getting deleted, as well as all of their descendents.

        Passing an AsyncDocumentReference deletes that one document and all of
        its descendents.

        Args:
            reference (Union[
                :class:`@google.cloud.firestore_v1.async_collection.CollectionReference`,
                :class:`@google.cloud.firestore_v1.async_document.DocumentReference`,
            ])
                The reference to be deleted.

            bulk_writer (Optional[:class:`@google.cloud.firestore_v1.bulk_writer.BulkWriter`])
                The BulkWriter used to delete all matching documents. Supply this
                if you want to override the default throttling behavior.
        """
        if bulk_writer is None:
            bulk_writer = self.bulk_writer()

        return await self._recursive_delete(
            reference, bulk_writer=bulk_writer, chunk_size=chunk_size,
        )

    async def _recursive_delete(
        self,

        reference: Union[AsyncCollectionReference, AsyncDocumentReference],