# 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. import builtins import io import json import logging import os from typing import Any, Optional, Union, overload from urllib.parse import urlencode import google.auth from . import _api_client from . import _api_module from . import _common from . import _extra_utils from . import _transformers as t from . import types from ._common import get_value_by_path as getv from ._common import set_value_by_path as setv from .pagers import AsyncPager, Pager logger = logging.getLogger('google_genai.files') def _CreateFileParameters_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']) is not None: setv(to_object, ['file'], getv(from_object, ['file'])) return to_object def _CreateFileResponse_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 def _DeleteFileParameters_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', 'file'], t.t_file_name(getv(from_object, ['name'])) ) return to_object def _DeleteFileResponse_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 def _GetFileParameters_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', 'file'], t.t_file_name(getv(from_object, ['name'])) ) return to_object def _InternalRegisterFilesParameters_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, ['uris']) is not None: setv(to_object, ['uris'], getv(from_object, ['uris'])) return to_object def _ListFilesConfig_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 _ListFilesParameters_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: _ListFilesConfig_to_mldev(getv(from_object, ['config']), to_object) return to_object def _ListFilesResponse_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, ['files']) is not None: setv(to_object, ['files'], [item for item in getv(from_object, ['files'])]) return to_object def _RegisterFilesResponse_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, ['files']) is not None: setv(to_object, ['files'], [item for item in getv(from_object, ['files'])]) return to_object class Files(_api_module.BaseModule): def _list( self, *, config: Optional[types.ListFilesConfigOrDict] = None ) -> types.ListFilesResponse: parameter_model = types._ListFilesParameters( config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _ListFilesParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'files'.format_map(request_url_dict) else: path = 'files' 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 = _ListFilesResponse_from_mldev(response_dict) return_value = types.ListFilesResponse._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value def _create( self, *, file: types.FileOrDict, config: Optional[types.CreateFileConfigOrDict] = None, ) -> types.CreateFileResponse: parameter_model = types._CreateFileParameters( file=file, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _CreateFileParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'upload/v1beta/files'.format_map(request_url_dict) else: path = 'upload/v1beta/files' 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.CreateFileResponse(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 = _CreateFileResponse_from_mldev(response_dict) return_value = types.CreateFileResponse._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) self._api_client._verify_response(return_value) return return_value def get( self, *, name: str, config: Optional[types.GetFileConfigOrDict] = None ) -> types.File: """Retrieves the file information from the service. Args: name (str): The name identifier for the file to retrieve. config (GetFileConfig): Optional, configuration for the get method. Returns: File: The file information. Usage: .. code-block:: python file = client.files.get(name='files/...') print(file.uri) """ parameter_model = types._GetFileParameters( name=name, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _GetFileParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'files/{file}'.format_map(request_url_dict) else: path = 'files/{file}' 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.File._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) self._api_client._verify_response(return_value) return return_value def delete( self, *, name: str, config: Optional[types.DeleteFileConfigOrDict] = None ) -> types.DeleteFileResponse: """Deletes a remotely stored file. Args: name (str): The name identifier for the file to delete. config (DeleteFileConfig): Optional, configuration for the delete method. Returns: DeleteFileResponse: The response for the delete method Usage: .. code-block:: python client.files.delete(name='files/...') """ parameter_model = types._DeleteFileParameters( name=name, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _DeleteFileParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'files/{file}'.format_map(request_url_dict) else: path = 'files/{file}' 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 ) response_dict = {} if not response.body else json.loads(response.body) if not self._api_client.vertexai: response_dict = _DeleteFileResponse_from_mldev(response_dict) return_value = types.DeleteFileResponse._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value def _register_files( self, *, uris: builtins.list[str], config: Optional[types.RegisterFilesConfigOrDict] = None, ) -> types.RegisterFilesResponse: parameter_model = types._InternalRegisterFilesParameters( uris=uris, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _InternalRegisterFilesParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'files:register'.format_map(request_url_dict) else: path = 'files:register' 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.RegisterFilesResponse(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 = _RegisterFilesResponse_from_mldev(response_dict) return_value = types.RegisterFilesResponse._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) self._api_client._verify_response(return_value) return return_value def upload( self, *, file: Union[str, os.PathLike[str], io.IOBase], config: Optional[types.UploadFileConfigOrDict] = None, ) -> types.File: """Calls the API to upload a file using a supported file service. Args: 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 `display_name`, `mime_type`, and `name`. """ if self._api_client.vertexai: raise ValueError( 'This method is only supported in the Gemini Developer client.' ) config_model = types.UploadFileConfig() if config: if isinstance(config, dict): config_model = types.UploadFileConfig(**config) else: config_model = config file_obj = types.File( mime_type=config_model.mime_type, name=config_model.name, display_name=config_model.display_name, ) else: # if not config file_obj = types.File() if file_obj.name is not None and not file_obj.name.startswith('files/'): file_obj.name = f'files/{file_obj.name}' http_options, size_bytes, mime_type = _extra_utils.prepare_resumable_upload( file, user_http_options=config_model.http_options, user_mime_type=config_model.mime_type, ) file_obj.size_bytes = size_bytes file_obj.mime_type = mime_type response = self._create( file=file_obj, config=types.CreateFileConfig( http_options=http_options, should_return_http_response=True ), ) 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 ): raise KeyError( 'Failed to create file. Upload URL was not returned from the create' ' file request.' ) upload_url = response.sdk_http_response.headers['x-goog-upload-url'] if isinstance(file, io.IOBase): return_file = self._api_client.upload_file( file, upload_url, file_obj.size_bytes, http_options=http_options ) else: fs_path = os.fspath(file) return_file = self._api_client.upload_file( fs_path, upload_url, file_obj.size_bytes, http_options=http_options ) return types.File._from_response( response=return_file.json['file'], kwargs=config_model.model_dump() if config else {}, ) @overload def download( self, *, file: Union[str, types.File, types.Video, types.GeneratedVideo], destination: None = None, config: Optional[types.DownloadFileConfigOrDict] = None, ) -> bytes: ... @overload def download( self, *, file: Union[str, types.File, types.Video, types.GeneratedVideo], destination: Union[str, os.PathLike[str], io.IOBase], config: Optional[types.DownloadFileConfigOrDict] = None, ) -> None: ... def download( self, *, file: Union[str, types.File, types.Video, types.GeneratedVideo], destination: Optional[Union[str, os.PathLike[str], io.IOBase]] = None, config: Optional[types.DownloadFileConfigOrDict] = None, ) -> Optional[bytes]: """Downloads a file's data from storage. Files created by `upload` can't be downloaded. You can tell which files are downloadable by checking the `source` or `download_uri` property. Note: When destination is None, this method returns the data as bytes. For `Video` and `GeneratedVideo` objects, if destination is None, it also sets the `video_bytes` property on the `Video` object. When destination is provided, chunks are streamed directly to disk or the writable stream without loading the full file into memory, and None is returned. Args: file: A file name, uri, or file object identifying which file to download. destination: Optional, a local file path or writable binary stream to write the file directly to in chunks, avoiding holding the full file in memory. config: Optional, configuration for the get method. Returns: bytes: The file data as bytes if destination is None; otherwise None. Usage: .. code-block:: python for file in client.files.list(): if file.download_uri is not None: break else: raise ValueError('No files found with a `download_uri`.') # Download to memory: data = client.files.download(file=file) # Stream directly to a file path (avoids memory spike): client.files.download(file=file, destination="output.mp4") # Download video object: video = types.Video(uri=file.uri) video_bytes = client.files.download(file=video) video.video_bytes """ if self._api_client.vertexai: raise ValueError( 'This method is only supported in the Gemini Developer client.' ) config_model = None if config: if isinstance(config, dict): config_model = types.DownloadFileConfig(**config) else: config_model = config if isinstance(file, types.File) and file.download_uri is None: raise ValueError( "Only generated files can be downloaded, uploaded files can't be " 'downloaded. You can tell which files are downloadable by checking ' 'the `source` or `download_uri` property.' ) name = t.t_file_name(file) path = f'files/{name}:download' query_params = {'alt': 'media'} path = f'{path}?{urlencode(query_params)}' http_options = None if getv(config_model, ['http_options']) is not None: http_options = getv(config_model, ['http_options']) data = self._api_client.download_file( path, http_options=http_options, destination=destination, ) if destination is None: if isinstance(file, types.Video): file.video_bytes = data elif isinstance(file, types.GeneratedVideo) and file.video is not None: file.video.video_bytes = data return data def register_files( self, *, auth: google.auth.credentials.Credentials, uris: builtins.list[str], config: Optional[types.RegisterFilesConfigOrDict] = None, ) -> types.RegisterFilesResponse: """Registers gcs files with the file service.""" if not isinstance(auth, google.auth.credentials.Credentials): raise ValueError( 'auth must be a google.auth.credentials.Credentials object.' ) if config is None: config = types.RegisterFilesConfig() else: config = types.RegisterFilesConfig.model_validate(config) config = config.model_copy(deep=True) http_options = config.http_options or types.HttpOptions() headers = http_options.headers or {} headers = {k.lower(): v for k, v in headers.items()} token = _api_client.get_token_from_credentials(self._api_client, auth) headers['authorization'] = f'Bearer {token}' if auth.quota_project_id: headers['x-goog-user-project'] = auth.quota_project_id http_options.headers = headers config.http_options = http_options return self._register_files(uris=uris, config=config) def list( self, *, config: Optional[types.ListFilesConfigOrDict] = None ) -> Pager[types.File]: """Lists all files from the service. Args: config (ListFilesConfig): Optional, configuration for the list method. Returns: A Pager object that contains one page of files. When iterating over the pager, it automatically fetches the next page if there are more. Usage: .. code-block:: python for file in client.files.list(config={'page_size': 10}): print(file.name) """ list_request = self._list return Pager( 'files', list_request, self._list(config=config), config, ) class AsyncFiles(_api_module.BaseModule): async def _list( self, *, config: Optional[types.ListFilesConfigOrDict] = None ) -> types.ListFilesResponse: parameter_model = types._ListFilesParameters( config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _ListFilesParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'files'.format_map(request_url_dict) else: path = 'files' 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 = _ListFilesResponse_from_mldev(response_dict) return_value = types.ListFilesResponse._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value async def _create( self, *, file: types.FileOrDict, config: Optional[types.CreateFileConfigOrDict] = None, ) -> types.CreateFileResponse: parameter_model = types._CreateFileParameters( file=file, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _CreateFileParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'upload/v1beta/files'.format_map(request_url_dict) else: path = 'upload/v1beta/files' 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.CreateFileResponse(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 = _CreateFileResponse_from_mldev(response_dict) return_value = types.CreateFileResponse._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) self._api_client._verify_response(return_value) return return_value async def get( self, *, name: str, config: Optional[types.GetFileConfigOrDict] = None ) -> types.File: """Retrieves the file information from the service. Args: name (str): The name identifier for the file to retrieve. config (GetFileConfig): Optional, configuration for the get method. Returns: File: The file information. Usage: .. code-block:: python file = await client.aio.files.get(name='files/...') print(file.uri) """ parameter_model = types._GetFileParameters( name=name, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _GetFileParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'files/{file}'.format_map(request_url_dict) else: path = 'files/{file}' 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.File._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) self._api_client._verify_response(return_value) return return_value async def delete( self, *, name: str, config: Optional[types.DeleteFileConfigOrDict] = None ) -> types.DeleteFileResponse: """Deletes a remotely stored file. Args: name (str): The name identifier for the file to delete. config (DeleteFileConfig): Optional, configuration for the delete method. Returns: DeleteFileResponse: The response for the delete method Usage: .. code-block:: python await client.aio.files.delete(name='files/...') """ parameter_model = types._DeleteFileParameters( name=name, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _DeleteFileParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'files/{file}'.format_map(request_url_dict) else: path = 'files/{file}' 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 ) response_dict = {} if not response.body else json.loads(response.body) if not self._api_client.vertexai: response_dict = _DeleteFileResponse_from_mldev(response_dict) return_value = types.DeleteFileResponse._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value async def _register_files( self, *, uris: builtins.list[str], config: Optional[types.RegisterFilesConfigOrDict] = None, ) -> types.RegisterFilesResponse: parameter_model = types._InternalRegisterFilesParameters( uris=uris, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _InternalRegisterFilesParameters_to_mldev(parameter_model) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'files:register'.format_map(request_url_dict) else: path = 'files:register' 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.RegisterFilesResponse(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 = _RegisterFilesResponse_from_mldev(response_dict) return_value = types.RegisterFilesResponse._from_response( response=response_dict, kwargs={ 'config': { 'response_schema': getattr( parameter_model.config, 'response_schema', None ), 'response_json_schema': getattr( parameter_model.config, 'response_json_schema', None ), 'include_all_fields': getattr( parameter_model.config, 'include_all_fields', None ), } } if getattr(parameter_model, 'config', None) else {}, ) self._api_client._verify_response(return_value) return return_value async def upload( self, *, file: Union[str, os.PathLike[str], io.IOBase], config: Optional[types.UploadFileConfigOrDict] = None, ) -> types.File: """Calls the API to upload a file asynchronously using a supported file service. Args: 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 `display_name`, `mime_type`, and `name`. """ if self._api_client.vertexai: raise ValueError( 'This method is only supported in the Gemini Developer client.' ) config_model = types.UploadFileConfig() if config: if isinstance(config, dict): config_model = types.UploadFileConfig(**config) else: config_model = config file_obj = types.File( mime_type=config_model.mime_type, name=config_model.name, display_name=config_model.display_name, ) else: # if not config file_obj = types.File() if file_obj.name is not None and not file_obj.name.startswith('files/'): file_obj.name = f'files/{file_obj.name}' http_options, size_bytes, mime_type = _extra_utils.prepare_resumable_upload( file, user_http_options=config_model.http_options, user_mime_type=config_model.mime_type, ) file_obj.size_bytes = size_bytes file_obj.mime_type = mime_type response = await self._create( file=file_obj, config=types.CreateFileConfig( http_options=http_options, should_return_http_response=True ), ) 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 create file. Upload URL was not returned from the create' ' file 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): return_file = await self._api_client.async_upload_file( file, upload_url, file_obj.size_bytes, http_options=http_options ) else: fs_path = os.fspath(file) return_file = await self._api_client.async_upload_file( fs_path, upload_url, file_obj.size_bytes, http_options=http_options ) return types.File._from_response( response=return_file.json['file'], kwargs=config_model.model_dump() if config else {}, ) @overload async def download( self, *, file: Union[str, types.File, types.Video, types.GeneratedVideo], destination: None = None, config: Optional[types.DownloadFileConfigOrDict] = None, ) -> bytes: ... @overload async def download( self, *, file: Union[str, types.File, types.Video, types.GeneratedVideo], destination: Union[str, os.PathLike[str], io.IOBase], config: Optional[types.DownloadFileConfigOrDict] = None, ) -> None: ... async def download( self, *, file: Union[str, types.File, types.Video, types.GeneratedVideo], destination: Optional[Union[str, os.PathLike[str], io.IOBase]] = None, config: Optional[types.DownloadFileConfigOrDict] = None, ) -> Optional[bytes]: """Downloads a file's data from the file service asynchronously. The Gemini Enterprise Agent Platform implementation of the API does not include the file service. Files created by `upload` can't be downloaded. You can tell which files are downloadable by checking the `download_uri` property. Note: When destination is None, this method returns the data as bytes. For `Video` and `GeneratedVideo` objects, if destination is None, it also sets the `video_bytes` property on the `Video` object. When destination is provided, chunks are streamed directly to disk or the writable stream without loading the full file into memory, and None is returned. Args: file: A file name, uri, or file object identifying which file to download. destination: Optional, a local file path or writable binary stream to write the file directly to in chunks, avoiding holding the full file in memory. config: Optional, configuration for the get method. Returns: bytes: The file data as bytes if destination is None; otherwise None. Usage: .. code-block:: python async for file in await client.aio.files.list(): if file.download_uri is not None: break else: raise ValueError('No files found with a `download_uri`.') # Download to memory: data = await client.aio.files.download(file=file) # Stream directly to a file path (avoids memory spike): await client.aio.files.download(file=file, destination="output.mp4") """ if self._api_client.vertexai: raise ValueError( 'This method is only supported in the Gemini Developer client.' ) config_model = None if config: if isinstance(config, dict): config_model = types.DownloadFileConfig(**config) else: config_model = config if isinstance(file, types.File) and file.download_uri is None: raise ValueError( "Only generated files can be downloaded, uploaded files can't be " 'downloaded. You can tell which files are downloadable by checking ' 'the `source` or `download_uri` property.' ) name = t.t_file_name(file) path = f'files/{name}:download' http_options = None if getv(config_model, ['http_options']) is not None: http_options = getv(config_model, ['http_options']) query_params = {'alt': 'media'} if query_params: path = f'{path}?{urlencode(query_params)}' data = await self._api_client.async_download_file( path, http_options=http_options, destination=destination, ) if destination is None: if isinstance(file, types.Video): file.video_bytes = data elif isinstance(file, types.GeneratedVideo) and file.video is not None: file.video.video_bytes = data return data async def register_files( self, *, auth: google.auth.credentials.Credentials, uris: builtins.list[str], config: Optional[types.RegisterFilesConfigOrDict] = None, ) -> types.RegisterFilesResponse: """Registers gcs files with the file service.""" if not isinstance(auth, google.auth.credentials.Credentials): raise ValueError( 'auth must be a google.auth.credentials.Credentials object.' ) if config is None: config = types.RegisterFilesConfig() else: config = types.RegisterFilesConfig.model_validate(config) config = config.model_copy(deep=True) http_options = config.http_options or types.HttpOptions() headers = http_options.headers or {} headers = {k.lower(): v for k, v in headers.items()} token = await _api_client.async_get_token_from_credentials( self._api_client, auth ) headers['authorization'] = f'Bearer {token}' if auth.quota_project_id: headers['x-goog-user-project'] = auth.quota_project_id http_options.headers = headers config.http_options = http_options return await self._register_files(uris=uris, config=config) async def list( self, *, config: Optional[types.ListFilesConfigOrDict] = None ) -> AsyncPager[types.File]: """Lists all files from the service asynchronously. Args: config (ListFilesConfig): Optional, configuration for the list method. Returns: A Pager object that contains one page of files. When iterating over the pager, it automatically fetches the next page if there are more. Usage: .. code-block:: python async for file in await client.aio.files.list(config={'page_size': 10}): print(file.name) """ list_request = self._list return AsyncPager( 'files', list_request, await self._list(config=config), config, )