# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

# Code generated by the Google Gen AI SDK generator DO NOT EDIT.

from functools import cached_property
import io
import json
import logging
import os
from typing import Any, Optional, Union
from urllib.parse import urlencode

from . import _api_module
from . import _common
from . import _extra_utils
from . import types
from ._common import get_value_by_path as getv
from ._common import set_value_by_path as setv
from ._operations_converters import _UploadToFileSearchStoreOperation_from_mldev
from .documents import AsyncDocuments, Documents
from .pagers import AsyncPager, Pager


logger = logging.getLogger('google_genai.filesearchstores')


def _CreateFileSearchStoreConfig_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}

  if getv(from_object, ['display_name']) is not None:
    setv(parent_object, ['displayName'], getv(from_object, ['display_name']))

  return to_object


def _CreateFileSearchStoreParameters_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['config']) is not None:
    _CreateFileSearchStoreConfig_to_mldev(
        getv(from_object, ['config']), to_object
    )

  return to_object


def _DeleteFileSearchStoreConfig_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}

  if getv(from_object, ['force']) is not None:
    setv(parent_object, ['_query', 'force'], getv(from_object, ['force']))

  return to_object


def _DeleteFileSearchStoreParameters_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['name']) is not None:
    setv(to_object, ['_url', 'name'], getv(from_object, ['name']))

  if getv(from_object, ['config']) is not None:
    _DeleteFileSearchStoreConfig_to_mldev(
        getv(from_object, ['config']), to_object
    )

  return to_object


def _GetFileSearchStoreParameters_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['name']) is not None:
    setv(to_object, ['_url', 'name'], getv(from_object, ['name']))

  return to_object


def _ImportFileConfig_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}

  if getv(from_object, ['custom_metadata']) is not None:
    setv(
        parent_object,
        ['customMetadata'],
        [item for item in getv(from_object, ['custom_metadata'])],
    )

  if getv(from_object, ['chunking_config']) is not None:
    setv(
        parent_object,
        ['chunkingConfig'],
        getv(from_object, ['chunking_config']),
    )

  return to_object


def _ImportFileOperation_from_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['name']) is not None:
    setv(to_object, ['name'], getv(from_object, ['name']))

  if getv(from_object, ['metadata']) is not None:
    setv(to_object, ['metadata'], getv(from_object, ['metadata']))

  if getv(from_object, ['done']) is not None:
    setv(to_object, ['done'], getv(from_object, ['done']))

  if getv(from_object, ['error']) is not None:
    setv(to_object, ['error'], getv(from_object, ['error']))

  if getv(from_object, ['response']) is not None:
    setv(
        to_object,
        ['response'],
        _ImportFileResponse_from_mldev(
            getv(from_object, ['response']), to_object
        ),
    )

  return to_object


def _ImportFileParameters_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['file_search_store_name']) is not None:
    setv(
        to_object,
        ['_url', 'file_search_store_name'],
        getv(from_object, ['file_search_store_name']),
    )

  if getv(from_object, ['file_name']) is not None:
    setv(to_object, ['fileName'], getv(from_object, ['file_name']))

  if getv(from_object, ['config']) is not None:
    _ImportFileConfig_to_mldev(getv(from_object, ['config']), to_object)

  return to_object


def _ImportFileResponse_from_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['sdkHttpResponse']) is not None:
    setv(
        to_object, ['sdk_http_response'], getv(from_object, ['sdkHttpResponse'])
    )

  if getv(from_object, ['parent']) is not None:
    setv(to_object, ['parent'], getv(from_object, ['parent']))

  if getv(from_object, ['documentName']) is not None:
    setv(to_object, ['document_name'], getv(from_object, ['documentName']))

  return to_object


def _ListFileSearchStoresConfig_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}

  if getv(from_object, ['page_size']) is not None:
    setv(
        parent_object, ['_query', 'pageSize'], getv(from_object, ['page_size'])
    )

  if getv(from_object, ['page_token']) is not None:
    setv(
        parent_object,
        ['_query', 'pageToken'],
        getv(from_object, ['page_token']),
    )

  return to_object


def _ListFileSearchStoresParameters_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['config']) is not None:
    _ListFileSearchStoresConfig_to_mldev(
        getv(from_object, ['config']), to_object
    )

  return to_object


def _ListFileSearchStoresResponse_from_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['sdkHttpResponse']) is not None:
    setv(
        to_object, ['sdk_http_response'], getv(from_object, ['sdkHttpResponse'])
    )

  if getv(from_object, ['nextPageToken']) is not None:
    setv(to_object, ['next_page_token'], getv(from_object, ['nextPageToken']))

  if getv(from_object, ['fileSearchStores']) is not None:
    setv(
        to_object,
        ['file_search_stores'],
        [item for item in getv(from_object, ['fileSearchStores'])],
    )

  return to_object


def _UploadToFileSearchStoreConfig_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}

  if getv(from_object, ['mime_type']) is not None:
    setv(parent_object, ['mimeType'], getv(from_object, ['mime_type']))

  if getv(from_object, ['display_name']) is not None:
    setv(parent_object, ['displayName'], getv(from_object, ['display_name']))

  if getv(from_object, ['custom_metadata']) is not None:
    setv(
        parent_object,
        ['customMetadata'],
        [item for item in getv(from_object, ['custom_metadata'])],
    )

  if getv(from_object, ['chunking_config']) is not None:
    setv(
        parent_object,
        ['chunkingConfig'],
        getv(from_object, ['chunking_config']),
    )

  return to_object


def _UploadToFileSearchStoreParameters_to_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['file_search_store_name']) is not None:
    setv(
        to_object,
        ['_url', 'file_search_store_name'],
        getv(from_object, ['file_search_store_name']),
    )

  if getv(from_object, ['config']) is not None:
    _UploadToFileSearchStoreConfig_to_mldev(
        getv(from_object, ['config']), to_object
    )

  return to_object


def _UploadToFileSearchStoreResumableResponse_from_mldev(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
  to_object: dict[str, Any] = {}
  if getv(from_object, ['sdkHttpResponse']) is not None:
    setv(
        to_object, ['sdk_http_response'], getv(from_object, ['sdkHttpResponse'])
    )

  return to_object


class FileSearchStores(_api_module.BaseModule):

  @cached_property
  def documents(self) -> Documents:
    return Documents(self._api_client)

  def create(
      self, *, config: Optional[types.CreateFileSearchStoreConfigOrDict] = None
  ) -> types.FileSearchStore:
    """Creates a File Search Store.

    Args:
      config (CreateFileSearchStoreConfig | None): Optional parameters for the
        request.

    Returns:
      FileSearchStore
    """

    parameter_model = types._CreateFileSearchStoreParameters(
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _CreateFileSearchStoreParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = 'fileSearchStores'.format_map(request_url_dict)
      else:
        path = 'fileSearchStores'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = self._api_client.request(
        'post', path, request_dict, http_options
    )

    response_dict = {} if not response.body else json.loads(response.body)

    return_value = types.FileSearchStore._from_response(
        response=response_dict, kwargs=parameter_model.model_dump()
    )

    self._api_client._verify_response(return_value)
    return return_value

  def get(
      self,
      *,
      name: str,
      config: Optional[types.GetFileSearchStoreConfigOrDict] = None,
  ) -> types.FileSearchStore:
    """Gets metadata about a FileSearchStore.

    Args:
      name (str): The resource name of the FileSearchStore. Example:
        `FileSearchStores/my-file-search-store-123`
      config (GetFileSearchStoreConfig | None): Optional parameters for the
        request.

    Returns:
      A FileSearchStore object containing the metadata.
    """

    parameter_model = types._GetFileSearchStoreParameters(
        name=name,
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _GetFileSearchStoreParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = '{name}'.format_map(request_url_dict)
      else:
        path = '{name}'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = self._api_client.request('get', path, request_dict, http_options)

    response_dict = {} if not response.body else json.loads(response.body)

    return_value = types.FileSearchStore._from_response(
        response=response_dict, kwargs=parameter_model.model_dump()
    )

    self._api_client._verify_response(return_value)
    return return_value

  def delete(
      self,
      *,
      name: str,
      config: Optional[types.DeleteFileSearchStoreConfigOrDict] = None,
  ) -> None:
    """Deletes a FileSearchStore.

    Args:
      name (str): The resource name of the FileSearchStore. Example:
        `FileSearchStores/my-file-search-store-123`
      config (DeleteFileSearchStoreConfig | None): Optional parameters for the
        request.

    Returns:
      None
    """

    parameter_model = types._DeleteFileSearchStoreParameters(
        name=name,
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _DeleteFileSearchStoreParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = '{name}'.format_map(request_url_dict)
      else:
        path = '{name}'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = self._api_client.request(
        'delete', path, request_dict, http_options
    )

  def _list(
      self, *, config: Optional[types.ListFileSearchStoresConfigOrDict] = None
  ) -> types.ListFileSearchStoresResponse:
    parameter_model = types._ListFileSearchStoresParameters(
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _ListFileSearchStoresParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = 'fileSearchStores'.format_map(request_url_dict)
      else:
        path = 'fileSearchStores'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = self._api_client.request('get', path, request_dict, http_options)

    response_dict = {} if not response.body else json.loads(response.body)

    if not self._api_client.vertexai:
      response_dict = _ListFileSearchStoresResponse_from_mldev(response_dict)

    return_value = types.ListFileSearchStoresResponse._from_response(
        response=response_dict, kwargs=parameter_model.model_dump()
    )

    self._api_client._verify_response(return_value)
    return return_value

  def _upload_to_file_search_store(
      self,
      *,
      file_search_store_name: str,
      config: Optional[types.UploadToFileSearchStoreConfigOrDict] = None,
  ) -> types.UploadToFileSearchStoreResumableResponse:
    parameter_model = types._UploadToFileSearchStoreParameters(
        file_search_store_name=file_search_store_name,
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _UploadToFileSearchStoreParameters_to_mldev(
          parameter_model
      )
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = (
            'upload/v1beta/{file_search_store_name}:uploadToFileSearchStore'
            .format_map(request_url_dict)
        )
      else:
        path = 'upload/v1beta/{file_search_store_name}:uploadToFileSearchStore'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = self._api_client.request(
        'post', path, request_dict, http_options
    )

    if config is not None and getattr(
        config, 'should_return_http_response', None
    ):
      return_value = types.UploadToFileSearchStoreResumableResponse(
          sdk_http_response=response
      )
      self._api_client._verify_response(return_value)
      return return_value

    response_dict = {} if not response.body else json.loads(response.body)

    if not self._api_client.vertexai:
      response_dict = _UploadToFileSearchStoreResumableResponse_from_mldev(
          response_dict
      )

    return_value = (
        types.UploadToFileSearchStoreResumableResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )
    )

    self._api_client._verify_response(return_value)
    return return_value

  def import_file(
      self,
      *,
      file_search_store_name: str,
      file_name: str,
      config: Optional[types.ImportFileConfigOrDict] = None,
  ) -> types.ImportFileOperation:
    """Imports a File from File Service to a FileSearchStore.

    This is a long-running operation, see aip.dev/151

    Args:
      file_search_store_name (str): The resource name of the FileSearchStore.
        Example: `fileSearchStores/my-file-search-store-123`
      file_name (str): The resource name of the File to import. Example:
        `files/abc-123`
      config (ImportFileConfig | None): Optional parameters for the request.

    Returns:
      ImportFileOperation.
    """

    parameter_model = types._ImportFileParameters(
        file_search_store_name=file_search_store_name,
        file_name=file_name,
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _ImportFileParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = '{file_search_store_name}:importFile'.format_map(
            request_url_dict
        )
      else:
        path = '{file_search_store_name}:importFile'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = self._api_client.request(
        'post', path, request_dict, http_options
    )

    response_dict = {} if not response.body else json.loads(response.body)

    if not self._api_client.vertexai:
      response_dict = _ImportFileOperation_from_mldev(response_dict)

    return_value = types.ImportFileOperation._from_response(
        response=response_dict, kwargs=parameter_model.model_dump()
    )

    self._api_client._verify_response(return_value)
    return return_value

  def upload_to_file_search_store(
      self,
      *,
      file_search_store_name: str,
      file: Union[str, os.PathLike[str], io.IOBase],
      config: Optional[types.UploadToFileSearchStoreConfigOrDict] = None,
  ) -> types.UploadToFileSearchStoreOperation:
    """Calls the API to upload a file to the given file search store.

    Args:
      file_search_store_name: The resource name of the FileSearchStore. Example:
        `fileSearchStores/file-search-store-123`
      file: A path to the file or an `IOBase` object to be uploaded. If it's an
        IOBase object, it must be opened in blocking (the default) mode and
        binary mode. In other words, do not use non-blocking mode or text mode.
        The given stream must be seekable, that is, it must be able to call
        `seek()` on 'path'.
      config: Optional parameters to set `diplay_name`, `mime_type`, and others.
    """
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    if config is None:
      user_config_dict: dict[str, Any] = {}
    elif isinstance(config, dict):
      user_config_dict = dict(config)
    else:
      user_config_dict = config.model_dump(exclude_unset=True)

    http_options, size_bytes, _ = _extra_utils.prepare_resumable_upload(
        file,
        user_http_options=user_config_dict.get('http_options'),
        user_mime_type=user_config_dict.get('mime_type'),
    )
    final_config_dict = {
        **user_config_dict,
        'http_options': http_options,
        'should_return_http_response': True,
    }
    response = self._upload_to_file_search_store(
        file_search_store_name=file_search_store_name,
        config=types.UploadToFileSearchStoreConfig(**final_config_dict),
    )
    if (
        response.sdk_http_response is None
        or response.sdk_http_response.headers is None
        or (
            'x-goog-upload-url' not in response.sdk_http_response.headers
            and 'X-Goog-Upload-URL' not in response.sdk_http_response.headers
        )
    ):
      raise KeyError(
          'Failed to upload file to file search store. Upload URL did not'
          ' returned from the upload request.'
      )
    elif 'x-goog-upload-url' in response.sdk_http_response.headers:
      upload_url = response.sdk_http_response.headers['x-goog-upload-url']
    else:
      upload_url = response.sdk_http_response.headers['X-Goog-Upload-URL']
    if isinstance(file, io.IOBase):
      upload_response = self._api_client.upload_file(
          file, upload_url, size_bytes, http_options=http_options
      )
    else:
      fs_path = os.fspath(file)
      upload_response = self._api_client.upload_file(
          fs_path, upload_url, size_bytes, http_options=http_options
      )
    response_dict = _UploadToFileSearchStoreOperation_from_mldev(
        upload_response.json
    )
    return types.UploadToFileSearchStoreOperation._from_response(
        response=response_dict, kwargs={}
    )

  def list(
      self, *, config: Optional[types.ListFileSearchStoresConfigOrDict] = None
  ) -> Pager[types.FileSearchStore]:
    """Lists FileSearchStores.

    Args:
      config (ListFileSearchStoresConfig): Optional configuration for the list
        request.

    Returns:
      A Pager object that contains one page of file search stores. When
      iterating over
      the pager, it automatically fetches the next page if there are more.

    Usage:

    .. code-block:: python
      for file_search_store in client.file_search_stores.list():
        print(f"file search store: {file_search_store.name} -
        {file_search_store.display_name}")
    """

    list_request = self._list
    return Pager(
        'file_search_stores',
        list_request,
        self._list(config=config),
        config,
    )


class AsyncFileSearchStores(_api_module.BaseModule):

  @cached_property
  def documents(self) -> AsyncDocuments:
    return AsyncDocuments(self._api_client)

  async def create(
      self, *, config: Optional[types.CreateFileSearchStoreConfigOrDict] = None
  ) -> types.FileSearchStore:
    """Creates a File Search Store.

    Args:
      config (CreateFileSearchStoreConfig | None): Optional parameters for the
        request.

    Returns:
      FileSearchStore
    """

    parameter_model = types._CreateFileSearchStoreParameters(
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _CreateFileSearchStoreParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = 'fileSearchStores'.format_map(request_url_dict)
      else:
        path = 'fileSearchStores'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = await self._api_client.async_request(
        'post', path, request_dict, http_options
    )

    response_dict = {} if not response.body else json.loads(response.body)

    return_value = types.FileSearchStore._from_response(
        response=response_dict, kwargs=parameter_model.model_dump()
    )

    self._api_client._verify_response(return_value)
    return return_value

  async def get(
      self,
      *,
      name: str,
      config: Optional[types.GetFileSearchStoreConfigOrDict] = None,
  ) -> types.FileSearchStore:
    """Gets metadata about a FileSearchStore.

    Args:
      name (str): The resource name of the FileSearchStore. Example:
        `FileSearchStores/my-file-search-store-123`
      config (GetFileSearchStoreConfig | None): Optional parameters for the
        request.

    Returns:
      A FileSearchStore object containing the metadata.
    """

    parameter_model = types._GetFileSearchStoreParameters(
        name=name,
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _GetFileSearchStoreParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = '{name}'.format_map(request_url_dict)
      else:
        path = '{name}'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = await self._api_client.async_request(
        'get', path, request_dict, http_options
    )

    response_dict = {} if not response.body else json.loads(response.body)

    return_value = types.FileSearchStore._from_response(
        response=response_dict, kwargs=parameter_model.model_dump()
    )

    self._api_client._verify_response(return_value)
    return return_value

  async def delete(
      self,
      *,
      name: str,
      config: Optional[types.DeleteFileSearchStoreConfigOrDict] = None,
  ) -> None:
    """Deletes a FileSearchStore.

    Args:
      name (str): The resource name of the FileSearchStore. Example:
        `FileSearchStores/my-file-search-store-123`
      config (DeleteFileSearchStoreConfig | None): Optional parameters for the
        request.

    Returns:
      None
    """

    parameter_model = types._DeleteFileSearchStoreParameters(
        name=name,
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _DeleteFileSearchStoreParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = '{name}'.format_map(request_url_dict)
      else:
        path = '{name}'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = await self._api_client.async_request(
        'delete', path, request_dict, http_options
    )

  async def _list(
      self, *, config: Optional[types.ListFileSearchStoresConfigOrDict] = None
  ) -> types.ListFileSearchStoresResponse:
    parameter_model = types._ListFileSearchStoresParameters(
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _ListFileSearchStoresParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = 'fileSearchStores'.format_map(request_url_dict)
      else:
        path = 'fileSearchStores'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = await self._api_client.async_request(
        'get', path, request_dict, http_options
    )

    response_dict = {} if not response.body else json.loads(response.body)

    if not self._api_client.vertexai:
      response_dict = _ListFileSearchStoresResponse_from_mldev(response_dict)

    return_value = types.ListFileSearchStoresResponse._from_response(
        response=response_dict, kwargs=parameter_model.model_dump()
    )

    self._api_client._verify_response(return_value)
    return return_value

  async def _upload_to_file_search_store(
      self,
      *,
      file_search_store_name: str,
      config: Optional[types.UploadToFileSearchStoreConfigOrDict] = None,
  ) -> types.UploadToFileSearchStoreResumableResponse:
    parameter_model = types._UploadToFileSearchStoreParameters(
        file_search_store_name=file_search_store_name,
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _UploadToFileSearchStoreParameters_to_mldev(
          parameter_model
      )
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = (
            'upload/v1beta/{file_search_store_name}:uploadToFileSearchStore'
            .format_map(request_url_dict)
        )
      else:
        path = 'upload/v1beta/{file_search_store_name}:uploadToFileSearchStore'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = await self._api_client.async_request(
        'post', path, request_dict, http_options
    )

    if config is not None and getattr(
        config, 'should_return_http_response', None
    ):
      return_value = types.UploadToFileSearchStoreResumableResponse(
          sdk_http_response=response
      )
      self._api_client._verify_response(return_value)
      return return_value

    response_dict = {} if not response.body else json.loads(response.body)

    if not self._api_client.vertexai:
      response_dict = _UploadToFileSearchStoreResumableResponse_from_mldev(
          response_dict
      )

    return_value = (
        types.UploadToFileSearchStoreResumableResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )
    )

    self._api_client._verify_response(return_value)
    return return_value

  async def import_file(
      self,
      *,
      file_search_store_name: str,
      file_name: str,
      config: Optional[types.ImportFileConfigOrDict] = None,
  ) -> types.ImportFileOperation:
    """Imports a File from File Service to a FileSearchStore.

    This is a long-running operation, see aip.dev/151

    Args:
      file_search_store_name (str): The resource name of the FileSearchStore.
        Example: `fileSearchStores/my-file-search-store-123`
      file_name (str): The resource name of the File to import. Example:
        `files/abc-123`
      config (ImportFileConfig | None): Optional parameters for the request.

    Returns:
      ImportFileOperation.
    """

    parameter_model = types._ImportFileParameters(
        file_search_store_name=file_search_store_name,
        file_name=file_name,
        config=config,
    )

    request_url_dict: Optional[dict[str, str]]
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    else:
      request_dict = _ImportFileParameters_to_mldev(parameter_model)
      request_url_dict = request_dict.get('_url')
      if request_url_dict:
        path = '{file_search_store_name}:importFile'.format_map(
            request_url_dict
        )
      else:
        path = '{file_search_store_name}:importFile'

    query_params = request_dict.get('_query')
    if query_params:
      path = f'{path}?{urlencode(query_params)}'
    # TODO: remove the hack that pops config.
    request_dict.pop('config', None)

    http_options: Optional[types.HttpOptions] = None
    if (
        parameter_model.config is not None
        and parameter_model.config.http_options is not None
    ):
      http_options = parameter_model.config.http_options

    request_dict = _common.convert_to_dict(request_dict)
    request_dict = _common.encode_unserializable_types(request_dict)

    response = await self._api_client.async_request(
        'post', path, request_dict, http_options
    )

    response_dict = {} if not response.body else json.loads(response.body)

    if not self._api_client.vertexai:
      response_dict = _ImportFileOperation_from_mldev(response_dict)

    return_value = types.ImportFileOperation._from_response(
        response=response_dict, kwargs=parameter_model.model_dump()
    )

    self._api_client._verify_response(return_value)
    return return_value

  async def upload_to_file_search_store(
      self,
      *,
      file_search_store_name: str,
      file: Union[str, os.PathLike[str], io.IOBase],
      config: Optional[types.UploadToFileSearchStoreConfigOrDict] = None,
  ) -> types.UploadToFileSearchStoreOperation:
    """Calls the API to upload a file to the given file search store.

    Args:
      file_search_store_name: The resource name of the FileSearchStore. Example:
        `fileSearchStores/file-search-store-123`
      file: A path to the file or an `IOBase` object to be uploaded. If it's an
        IOBase object, it must be opened in blocking (the default) mode and
        binary mode. In other words, do not use non-blocking mode or text mode.
        The given stream must be seekable, that is, it must be able to call
        `seek()` on 'path'.
      config: Optional parameters to set `diplay_name`, `mime_type` and others.
    """
    if self._api_client.vertexai:
      raise ValueError(
          'This method is only supported in the Gemini Developer client.'
      )
    if config is None:
      user_config_dict: dict[str, Any] = {}
    elif isinstance(config, dict):
      user_config_dict = dict(config)
    else:
      user_config_dict = config.model_dump(exclude_unset=True)

    http_options, size_bytes, _ = _extra_utils.prepare_resumable_upload(
        file,
        user_http_options=user_config_dict.get('http_options'),
        user_mime_type=user_config_dict.get('mime_type'),
    )
    final_config_dict = {
        **user_config_dict,
        'http_options': http_options,
        'should_return_http_response': True,
    }
    response = await self._upload_to_file_search_store(
        file_search_store_name=file_search_store_name,
        config=types.UploadToFileSearchStoreConfig(**final_config_dict),
    )
    if (
        response.sdk_http_response is None
        or response.sdk_http_response.headers is None
        or (
            'x-goog-upload-url' not in response.sdk_http_response.headers
            and 'X-Goog-Upload-URL' not in response.sdk_http_response.headers
        )
    ):
      raise KeyError(
          'Failed to upload file to file search store. Upload URL did not'
          ' returned from the upload request.'
      )
    elif 'x-goog-upload-url' in response.sdk_http_response.headers:
      upload_url = response.sdk_http_response.headers['x-goog-upload-url']
    else:
      upload_url = response.sdk_http_response.headers['X-Goog-Upload-URL']
    if isinstance(file, io.IOBase):
      upload_response = await self._api_client.async_upload_file(
          file, upload_url, size_bytes, http_options=http_options
      )
    else:
      fs_path = os.fspath(file)
      upload_response = await self._api_client.async_upload_file(
          fs_path, upload_url, size_bytes, http_options=http_options
      )
    response_dict = _UploadToFileSearchStoreOperation_from_mldev(
        upload_response.json
    )
    return types.UploadToFileSearchStoreOperation._from_response(
        response=response_dict, kwargs={}
    )

  async def list(
      self, *, config: Optional[types.ListFileSearchStoresConfigOrDict] = None
  ) -> AsyncPager[types.FileSearchStore]:
    """Lists FileSearchStores asynchronously.

    Args:
      config (ListFileSearchStoresConfig): Optional parameters for the request,
        such as page_size.

    Returns:
      A Pager object that contains one page of FileSearchStores. When iterating
      over
      the pager, it automatically fetches the next page if there are more.

    Usage:

    .. code-block:: python
      async for file_search_store in await client.aio.file_search_stores.list():
        print(f"file search store: {file_search_store.name} -
        {file_search_store.display_name}")
    """

    list_request = self._list
    return AsyncPager(
        'file_search_stores',
        list_request,
        await self._list(config=config),
        config,
    )
