From 77d2daaacb18c62e5f5117622789de73226797c7 Mon Sep 17 00:00:00 2001 From: Martin Brodeur <63083086+brodmart@users.noreply.github.com> Date: Fri, 24 Apr 2026 09:11:32 -0400 Subject: [PATCH 1/2] fix: validate resource name to prevent path traversal in Documents operations Documents.get and Documents.delete (sync and async) pass the caller-supplied `name` parameter directly into `format_map` with no sanitization. A name containing `..` sequences can traverse outside the intended resource path. Add a lightweight `_validate_resource_name()` guard that raises ValueError on any name containing `..` before the URL is constructed. Fixes a path traversal reported via Google VRP (report 503098362). --- google/genai/documents.py | 1164 +++++++++++++++++++------------------ 1 file changed, 588 insertions(+), 576 deletions(-) diff --git a/google/genai/documents.py b/google/genai/documents.py index d542fea52..7500af8ba 100644 --- a/google/genai/documents.py +++ b/google/genai/documents.py @@ -1,589 +1,601 @@ -# 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 partial -import json -import logging -from typing import Any, Optional, Union -from urllib.parse import urlencode - -from . import _api_module -from . import _common -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.documents') - - -def _DeleteDocumentConfig_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 _DeleteDocumentParameters_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: - _DeleteDocumentConfig_to_mldev(getv(from_object, ['config']), to_object) - - return to_object - - -def _GetDocumentParameters_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 _ListDocumentsConfig_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 _ListDocumentsParameters_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, ['parent']) is not None: - setv(to_object, ['_url', 'parent'], getv(from_object, ['parent'])) - - if getv(from_object, ['config']) is not None: - _ListDocumentsConfig_to_mldev(getv(from_object, ['config']), to_object) - - return to_object - - -def _ListDocumentsResponse_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, ['documents']) is not None: - setv( - to_object, - ['documents'], - [item for item in getv(from_object, ['documents'])], - ) - - return to_object - - -class Documents(_api_module.BaseModule): - - def get( - self, *, name: str, config: Optional[types.GetDocumentConfigOrDict] = None - ) -> types.Document: - """Gets metadata about a Document. - - Args: - name (str): The resource name of the Document. - Example: ragStores/rag-store-foo/documents/documents-bar - config (GetDocumentConfig | None): Optional parameters for the request. - - Returns: - The Document. - """ - - parameter_model = types._GetDocumentParameters( - name=name, - config=config, - ) - - request_url_dict: Optional[dict[str, str]] - if self._api_client.vertexai: + # 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 partial + import json + import logging + from typing import Any, Optional, Union + from urllib.parse import urlencode + + from . import _api_module + from . import _common + 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.documents') + + + def _validate_resource_name(name: str) -> None: + if '..' in name: raise ValueError( - 'This method is only supported in the Gemini Developer client.' + f'Invalid resource name {name!r}: must not contain path traversal sequences' + ) + + + def _DeleteDocumentConfig_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 _DeleteDocumentParameters_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: + _DeleteDocumentConfig_to_mldev(getv(from_object, ['config']), to_object) + + return to_object + + + def _GetDocumentParameters_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 _ListDocumentsConfig_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']), ) - else: - request_dict = _GetDocumentParameters_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.Document._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.DeleteDocumentConfigOrDict] = None, - ) -> None: - """Deletes a Document. - - Args: - name (str): The resource name of the Document. - Example: ragStores/rag-store-foo/documents/documents-bar - config (DeleteDocumentConfig | None): Optional parameters for the request. - - Returns: - None - """ - - parameter_model = types._DeleteDocumentParameters( - 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.' + return to_object + + + def _ListDocumentsParameters_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, ['parent']) is not None: + setv(to_object, ['_url', 'parent'], getv(from_object, ['parent'])) + + if getv(from_object, ['config']) is not None: + _ListDocumentsConfig_to_mldev(getv(from_object, ['config']), to_object) + + return to_object + + + def _ListDocumentsResponse_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, ['documents']) is not None: + setv( + to_object, + ['documents'], + [item for item in getv(from_object, ['documents'])], + ) + + return to_object + + + class Documents(_api_module.BaseModule): + + def get( + self, *, name: str, config: Optional[types.GetDocumentConfigOrDict] = None + ) -> types.Document: + """Gets metadata about a Document. + + Args: + name (str): The resource name of the Document. + Example: ragStores/rag-store-foo/documents/documents-bar + config (GetDocumentConfig | None): Optional parameters for the request. + + Returns: + The Document. + """ + + parameter_model = types._GetDocumentParameters( + name=name, + config=config, ) - else: - request_dict = _DeleteDocumentParameters_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) - - self._api_client.request('delete', path, request_dict, http_options) - - def _list( - self, - *, - parent: str, - config: Optional[types.ListDocumentsConfigOrDict] = None, - ) -> types.ListDocumentsResponse: - parameter_model = types._ListDocumentsParameters( - parent=parent, - 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 = _ListDocumentsParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - path = '{parent}/documents'.format_map(request_url_dict) + 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: - path = '{parent}/documents' - - 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 = _ListDocumentsResponse_from_mldev(response_dict) - - return_value = types.ListDocumentsResponse._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 list( - self, - *, - parent: str, - config: Optional[types.ListDocumentsConfigOrDict] = None, - ) -> Pager[types.Document]: - """Lists documents. - - Args: - parent (str): The name of the RagStore containing the Documents. - config (ListDocumentsConfig): Optional configuration for the list request. - - Returns: - A Pager object that contains one page of documents. When iterating over - the pager, it automatically fetches the next page if there are more. - Usage: - .. code-block:: python - for document in client.documents.list(parent='rag_store_name'): - print(f"document: {document.name} - {document.display_name}") - """ - - list_request = partial(self._list, parent=parent) - return Pager( - 'documents', - list_request, - self._list(parent=parent, config=config), - config, - ) - - -class AsyncDocuments(_api_module.BaseModule): - - async def get( - self, *, name: str, config: Optional[types.GetDocumentConfigOrDict] = None - ) -> types.Document: - """Gets metadata about a Document. - - Args: - name (str): The resource name of the Document. - Example: ragStores/rag-store-foo/documents/documents-bar - config (GetDocumentConfig | None): Optional parameters for the request. - - Returns: - The Document. - """ - - parameter_model = types._GetDocumentParameters( - 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.' + request_dict = _GetDocumentParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + _validate_resource_name(request_url_dict.get('name', '')) + 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.Document._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 {}, ) - else: - request_dict = _GetDocumentParameters_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.Document._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.DeleteDocumentConfigOrDict] = None, - ) -> None: - """Deletes a Document. - - Args: - name (str): The resource name of the Document. - Example: ragStores/rag-store-foo/documents/documents-bar - config (DeleteDocumentConfig | None): Optional parameters for the request. - - Returns: - None - """ - - parameter_model = types._DeleteDocumentParameters( - 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.' + + self._api_client._verify_response(return_value) + return return_value + + def delete( + self, + *, + name: str, + config: Optional[types.DeleteDocumentConfigOrDict] = None, + ) -> None: + """Deletes a Document. + + Args: + name (str): The resource name of the Document. + Example: ragStores/rag-store-foo/documents/documents-bar + config (DeleteDocumentConfig | None): Optional parameters for the request. + + Returns: + None + """ + + parameter_model = types._DeleteDocumentParameters( + name=name, + config=config, ) - else: - request_dict = _DeleteDocumentParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - path = '{name}'.format_map(request_url_dict) + + 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: - 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) - - await self._api_client.async_request( - 'delete', path, request_dict, http_options - ) - - async def _list( - self, - *, - parent: str, - config: Optional[types.ListDocumentsConfigOrDict] = None, - ) -> types.ListDocumentsResponse: - parameter_model = types._ListDocumentsParameters( - parent=parent, - 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.' + request_dict = _DeleteDocumentParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + _validate_resource_name(request_url_dict.get('name', '')) + 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) + + self._api_client.request('delete', path, request_dict, http_options) + + def _list( + self, + *, + parent: str, + config: Optional[types.ListDocumentsConfigOrDict] = None, + ) -> types.ListDocumentsResponse: + parameter_model = types._ListDocumentsParameters( + parent=parent, + config=config, ) - else: - request_dict = _ListDocumentsParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - path = '{parent}/documents'.format_map(request_url_dict) - else: - path = '{parent}/documents' - - 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 = _ListDocumentsResponse_from_mldev(response_dict) - - return_value = types.ListDocumentsResponse._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 list( - self, - *, - parent: str, - config: Optional[types.ListDocumentsConfigOrDict] = None, - ) -> AsyncPager[types.Document]: - """Lists documents asynchronously. - - Args: - parent (str): The name of the RagStore containing the Documents. - config (ListDocumentsConfig): Optional configuration for the list request. + 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 = _ListDocumentsParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + path = '{parent}/documents'.format_map(request_url_dict) + else: + path = '{parent}/documents' + + 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 = _ListDocumentsResponse_from_mldev(response_dict) + + return_value = types.ListDocumentsResponse._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 list( + self, + *, + parent: str, + config: Optional[types.ListDocumentsConfigOrDict] = None, + ) -> Pager[types.Document]: + """Lists documents. + + Args: + parent (str): The name of the RagStore containing the Documents. + config (ListDocumentsConfig): Optional configuration for the list request. + + Returns: + A Pager object that contains one page of documents. When iterating over + the pager, it automatically fetches the next page if there are more. + Usage: + .. code-block:: python + for document in client.documents.list(parent='rag_store_name'): + print(f"document: {document.name} - {document.display_name}") + """ + + list_request = partial(self._list, parent=parent) + return Pager( + 'documents', + list_request, + self._list(parent=parent, config=config), + config, + ) + + + class AsyncDocuments(_api_module.BaseModule): + + async def get( + self, *, name: str, config: Optional[types.GetDocumentConfigOrDict] = None + ) -> types.Document: + """Gets metadata about a Document. + + Args: + name (str): The resource name of the Document. + Example: ragStores/rag-store-foo/documents/documents-bar + config (GetDocumentConfig | None): Optional parameters for the request. + + Returns: + The Document. + """ + + parameter_model = types._GetDocumentParameters( + name=name, + config=config, + ) - Returns: - A Pager object that contains one page of documents. When iterating over - the pager, it automatically fetches the next page if there are more. - Usage: - .. code-block:: python - async for document in await - client.aio.documents.list(parent='rag_store_name'): - print(f"document: {document.name} - {document.display_name}") - """ + 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 = _GetDocumentParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + _validate_resource_name(request_url_dict.get('name', '')) + 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.Document._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.DeleteDocumentConfigOrDict] = None, + ) -> None: + """Deletes a Document. + + Args: + name (str): The resource name of the Document. + Example: ragStores/rag-store-foo/documents/documents-bar + config (DeleteDocumentConfig | None): Optional parameters for the request. + + Returns: + None + """ + + parameter_model = types._DeleteDocumentParameters( + name=name, + config=config, + ) - list_request = partial(self._list, parent=parent) - return AsyncPager( - 'documents', - list_request, - await self._list(parent=parent, config=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 = _DeleteDocumentParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + _validate_resource_name(request_url_dict.get('name', '')) + 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) + + await self._api_client.async_request( + 'delete', path, request_dict, http_options + ) + + async def _list( + self, + *, + parent: str, + config: Optional[types.ListDocumentsConfigOrDict] = None, + ) -> types.ListDocumentsResponse: + parameter_model = types._ListDocumentsParameters( + parent=parent, + 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 = _ListDocumentsParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + path = '{parent}/documents'.format_map(request_url_dict) + else: + path = '{parent}/documents' + + 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 = _ListDocumentsResponse_from_mldev(response_dict) + + return_value = types.ListDocumentsResponse._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 list( + self, + *, + parent: str, + config: Optional[types.ListDocumentsConfigOrDict] = None, + ) -> AsyncPager[types.Document]: + """Lists documents asynchronously. + + Args: + parent (str): The name of the RagStore containing the Documents. + config (ListDocumentsConfig): Optional configuration for the list request. + + Returns: + A Pager object that contains one page of documents. When iterating over + the pager, it automatically fetches the next page if there are more. + Usage: + .. code-block:: python + async for document in await + client.aio.documents.list(parent='rag_store_name'): + print(f"document: {document.name} - {document.display_name}") + """ + + list_request = partial(self._list, parent=parent) + return AsyncPager( + 'documents', + list_request, + await self._list(parent=parent, config=config), + config, + ) + From 770aeb670397394a93a6a2962591f9265954992e Mon Sep 17 00:00:00 2001 From: Martin Brodeur <63083086+brodmart@users.noreply.github.com> Date: Fri, 24 Apr 2026 11:43:21 -0400 Subject: [PATCH 2/2] fix: validate resource name to prevent path traversal in Documents operations --- google/genai/documents.py | 1074 ++++++++++++++++++------------------- 1 file changed, 524 insertions(+), 550 deletions(-) diff --git a/google/genai/documents.py b/google/genai/documents.py index 7500af8ba..ec03e3d7a 100644 --- a/google/genai/documents.py +++ b/google/genai/documents.py @@ -1,601 +1,575 @@ - # 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 partial - import json - import logging - from typing import Any, Optional, Union - from urllib.parse import urlencode - - from . import _api_module - from . import _common - 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.documents') - - - def _validate_resource_name(name: str) -> None: +# 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 partial +import json +import logging +from typing import Any, Optional, Union +from urllib.parse import urlencode + +from . import _api_module +from . import _common +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.documents') + + +def _validate_resource_name(name: str) -> None: if '..' in name: - raise ValueError( - f'Invalid resource name {name!r}: must not contain path traversal sequences' - ) - - - def _DeleteDocumentConfig_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 _DeleteDocumentParameters_to_mldev( - from_object: Union[dict[str, Any], object], - parent_object: Optional[dict[str, Any]] = None, - ) -> dict[str, Any]: + raise ValueError( + f'Invalid resource name {name!r}: must not contain path traversal sequences' + ) + + +def _DeleteDocumentConfig_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, ['force']) is not None: + setv(parent_object, ['_query', 'force'], getv(from_object, ['force'])) + + return to_object + + +def _DeleteDocumentParameters_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: - _DeleteDocumentConfig_to_mldev(getv(from_object, ['config']), to_object) - + _DeleteDocumentConfig_to_mldev(getv(from_object, ['config']), to_object) + return to_object - - - def _GetDocumentParameters_to_mldev( - from_object: Union[dict[str, Any], object], - parent_object: Optional[dict[str, Any]] = None, - ) -> dict[str, Any]: - to_object: dict[str, Any] = {} + + +def _GetDocumentParameters_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 _ListDocumentsConfig_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 _ListDocumentsParameters_to_mldev( - from_object: Union[dict[str, Any], object], - parent_object: Optional[dict[str, Any]] = None, - ) -> dict[str, Any]: - to_object: dict[str, Any] = {} + setv(to_object, ['_url', 'name'], getv(from_object, ['name'])) + + return to_object + + +def _ListDocumentsConfig_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 _ListDocumentsParameters_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, ['parent']) is not None: - setv(to_object, ['_url', 'parent'], getv(from_object, ['parent'])) - - if getv(from_object, ['config']) is not None: - _ListDocumentsConfig_to_mldev(getv(from_object, ['config']), to_object) - - return to_object - - - def _ListDocumentsResponse_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'])) - + setv(to_object, ['_url', 'parent'], getv(from_object, ['parent'])) + + if getv(from_object, ['config']) is not None: + _ListDocumentsConfig_to_mldev(getv(from_object, ['config']), to_object) + + return to_object + + +def _ListDocumentsResponse_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, ['documents']) is not None: - setv( - to_object, - ['documents'], - [item for item in getv(from_object, ['documents'])], - ) + setv( + to_object, + ['documents'], + [item for item in getv(from_object, ['documents'])], + ) return to_object - - class Documents(_api_module.BaseModule): - - def get( + +class Documents(_api_module.BaseModule): + + def get( self, *, name: str, config: Optional[types.GetDocumentConfigOrDict] = None - ) -> types.Document: - """Gets metadata about a Document. - - Args: - name (str): The resource name of the Document. - Example: ragStores/rag-store-foo/documents/documents-bar - config (GetDocumentConfig | None): Optional parameters for the request. - - Returns: - The Document. - """ - - parameter_model = types._GetDocumentParameters( - 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.' + ) -> types.Document: + """Gets metadata about a Document. + + Args: + name (str): The resource name of the Document. + Example: ragStores/rag-store-foo/documents/documents-bar + config (GetDocumentConfig | None): Optional parameters for the request. + + Returns: + The Document. + """ + parameter_model = types._GetDocumentParameters( + name=name, + config=config, ) - else: - request_dict = _GetDocumentParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - _validate_resource_name(request_url_dict.get('name', '')) - 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.Document._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 + 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 = _GetDocumentParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + _validate_resource_name(request_url_dict.get('name', '')) + 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.Document._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.DeleteDocumentConfigOrDict] = None, + config: Optional[types.DeleteDocumentConfigOrDict] = None, ) -> None: - """Deletes a Document. - - Args: - name (str): The resource name of the Document. - Example: ragStores/rag-store-foo/documents/documents-bar - config (DeleteDocumentConfig | None): Optional parameters for the request. - - Returns: - None - """ - - parameter_model = types._DeleteDocumentParameters( - 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.' + """Deletes a Document. + Args: + name (str): The resource name of the Document. + Example: ragStores/rag-store-foo/documents/documents-bar + config (DeleteDocumentConfig | None): Optional parameters for the request. + + Returns: + None + """ + parameter_model = types._DeleteDocumentParameters( + name=name, + config=config, ) - else: - request_dict = _DeleteDocumentParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - _validate_resource_name(request_url_dict.get('name', '')) - path = '{name}'.format_map(request_url_dict) + 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: - 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) - - self._api_client.request('delete', path, request_dict, http_options) - - def _list( + request_dict = _DeleteDocumentParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + _validate_resource_name(request_url_dict.get('name', '')) + 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) + + self._api_client.request('delete', path, request_dict, http_options) + + def _list( self, *, parent: str, config: Optional[types.ListDocumentsConfigOrDict] = None, - ) -> types.ListDocumentsResponse: - parameter_model = types._ListDocumentsParameters( - parent=parent, - 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 = _ListDocumentsParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - path = '{parent}/documents'.format_map(request_url_dict) - else: - path = '{parent}/documents' - - 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 = _ListDocumentsResponse_from_mldev(response_dict) - - return_value = types.ListDocumentsResponse._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 + ) -> types.ListDocumentsResponse: + parameter_model = types._ListDocumentsParameters( + parent=parent, + 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 = _ListDocumentsParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + path = '{parent}/documents'.format_map(request_url_dict) + else: + path = '{parent}/documents' + + 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 = _ListDocumentsResponse_from_mldev(response_dict) + + return_value = types.ListDocumentsResponse._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 list( self, *, parent: str, config: Optional[types.ListDocumentsConfigOrDict] = None, - ) -> Pager[types.Document]: - """Lists documents. - - Args: - parent (str): The name of the RagStore containing the Documents. - config (ListDocumentsConfig): Optional configuration for the list request. - - Returns: - A Pager object that contains one page of documents. When iterating over - the pager, it automatically fetches the next page if there are more. - Usage: - .. code-block:: python - for document in client.documents.list(parent='rag_store_name'): - print(f"document: {document.name} - {document.display_name}") - """ - - list_request = partial(self._list, parent=parent) - return Pager( - 'documents', - list_request, - self._list(parent=parent, config=config), - config, - ) - - - class AsyncDocuments(_api_module.BaseModule): - + ) -> Pager[types.Document]: + """Lists documents. + + Args: + parent (str): The name of the RagStore containing the Documents. + config (ListDocumentsConfig): Optional configuration for the list request. + + Returns: + A Pager object that contains one page of documents. When iterating over + the pager, it automatically fetches the next page if there are more. + Usage: + .. code-block:: python + for document in client.documents.list(parent='rag_store_name'): + print(f"document: {document.name} - {document.display_name}") + """ + + list_request = partial(self._list, parent=parent) + return Pager( + 'documents', + list_request, + self._list(parent=parent, config=config), + config, + ) + + +class AsyncDocuments(_api_module.BaseModule): + async def get( self, *, name: str, config: Optional[types.GetDocumentConfigOrDict] = None - ) -> types.Document: - """Gets metadata about a Document. - - Args: - name (str): The resource name of the Document. - Example: ragStores/rag-store-foo/documents/documents-bar - config (GetDocumentConfig | None): Optional parameters for the request. - - Returns: - The Document. - """ - - parameter_model = types._GetDocumentParameters( - 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.' + ) -> types.Document: + """Gets metadata about a Document. + Args: + name (str): The resource name of the Document. + Example: ragStores/rag-store-foo/documents/documents-bar + config (GetDocumentConfig | None): Optional parameters for the request. + + Returns: + The Document. + """ + parameter_model = types._GetDocumentParameters( + 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 = _GetDocumentParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + _validate_resource_name(request_url_dict.get('name', '')) + 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.Document._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 {}, ) - else: - request_dict = _GetDocumentParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - _validate_resource_name(request_url_dict.get('name', '')) - 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.Document._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 + self._api_client._verify_response(return_value) + return return_value async def delete( self, *, name: str, config: Optional[types.DeleteDocumentConfigOrDict] = None, - ) -> None: - """Deletes a Document. - - Args: - name (str): The resource name of the Document. - Example: ragStores/rag-store-foo/documents/documents-bar - config (DeleteDocumentConfig | None): Optional parameters for the request. - - Returns: - None - """ - - parameter_model = types._DeleteDocumentParameters( - 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 = _DeleteDocumentParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - _validate_resource_name(request_url_dict.get('name', '')) - 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) - - await self._api_client.async_request( - 'delete', path, request_dict, http_options - ) - - async def _list( - self, + ) -> None: + """Deletes a Document. + Args: + name (str): The resource name of the Document. + Example: ragStores/rag-store-foo/documents/documents-bar + config (DeleteDocumentConfig | None): Optional parameters for the request. + + Returns: + None + """ + parameter_model = types._DeleteDocumentParameters( + 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 = _DeleteDocumentParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + _validate_resource_name(request_url_dict.get('name', '')) + 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) + + await self._api_client.async_request( + 'delete', path, request_dict, http_options + ) + + async def _list( + self, *, parent: str, config: Optional[types.ListDocumentsConfigOrDict] = None, ) -> types.ListDocumentsResponse: - parameter_model = types._ListDocumentsParameters( - parent=parent, - 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.' + parameter_model = types._ListDocumentsParameters( + parent=parent, + config=config, ) - else: - request_dict = _ListDocumentsParameters_to_mldev(parameter_model) - request_url_dict = request_dict.get('_url') - if request_url_dict: - path = '{parent}/documents'.format_map(request_url_dict) + + 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: - path = '{parent}/documents' - - 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 = _ListDocumentsResponse_from_mldev(response_dict) - - return_value = types.ListDocumentsResponse._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 + request_dict = _ListDocumentsParameters_to_mldev(parameter_model) + request_url_dict = request_dict.get('_url') + if request_url_dict: + path = '{parent}/documents'.format_map(request_url_dict) + else: + path = '{parent}/documents' + + 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 = _ListDocumentsResponse_from_mldev(response_dict) + + return_value = types.ListDocumentsResponse._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 list( self, *, parent: str, config: Optional[types.ListDocumentsConfigOrDict] = None, - ) -> AsyncPager[types.Document]: - """Lists documents asynchronously. - - Args: - parent (str): The name of the RagStore containing the Documents. - config (ListDocumentsConfig): Optional configuration for the list request. - - Returns: - A Pager object that contains one page of documents. When iterating over - the pager, it automatically fetches the next page if there are more. - Usage: - .. code-block:: python - async for document in await - client.aio.documents.list(parent='rag_store_name'): - print(f"document: {document.name} - {document.display_name}") - """ - - list_request = partial(self._list, parent=parent) - return AsyncPager( - 'documents', - list_request, - await self._list(parent=parent, config=config), - config, - ) - + ) -> AsyncPager[types.Document]: + """Lists documents asynchronously. + + Args: + parent (str): The name of the RagStore containing the Documents. + config (ListDocumentsConfig): Optional configuration for the list request. + Returns: + A Pager object that contains one page of documents. When iterating over + the pager, it automatically fetches the next page if there are more. + Usage: + .. code-block:: python + async for document in await + client.aio.documents.list(parent='rag_store_name'): + print(f"document: {document.name} - {document.display_name}") + """ + + list_request = partial(self._list, parent=parent) + return AsyncPager( + 'documents', + list_request, + await self._list(parent=parent, config=config), + config, + )