# 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: """Lists all FileSearchStore owned by the user. Args: config (ListFileSearchStoresConfig | None): Optional parameters for the request, such as page_size. Returns: ListFileSearchStoresResponse: A paginated list of FileSearchStore. """ 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}") """ return Pager( 'file_search_stores', self._list, 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: """Lists all FileSearchStore owned by the user. Args: config (ListFileSearchStoresConfig | None): Optional parameters for the request, such as page_size. Returns: ListFileSearchStoresResponse: A paginated list of FileSearchStore. """ 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 configuration for the list request. 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.file_search_stores.list(): print(f"file search store: {file_search_store.name} - {file_search_store.display_name}") """ return AsyncPager( 'file_search_stores', self._list, await self._list(config=config), config, )