diff --git a/tests/unit/vertexai/genai/test_sandbox_snapshots.py b/tests/unit/vertexai/genai/test_sandbox_snapshots.py new file mode 100644 index 0000000000..bca4a65140 --- /dev/null +++ b/tests/unit/vertexai/genai/test_sandbox_snapshots.py @@ -0,0 +1,157 @@ +# Copyright 2026 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. + +import datetime +import importlib +from unittest import mock + +from google import auth +from google.auth import credentials as auth_credentials +from google.cloud import aiplatform +import vertexai +from google.cloud.aiplatform import initializer +from google.genai import types as genai_types +import pytest + +_TEST_CREDENTIALS = mock.Mock(spec=auth_credentials.AnonymousCredentials()) +_TEST_LOCATION = "us-central1" +_TEST_PROJECT = "test-project" +_TEST_RESOURCE_ID = "1028944691210842416" +_TEST_SANDBOX_ID = "sandbox-123" +_TEST_SNAPSHOT_ID = "snapshot-456" +_TEST_PARENT = f"projects/{_TEST_PROJECT}/locations/{_TEST_LOCATION}" +_TEST_AGENT_ENGINE_RESOURCE_NAME = ( + f"{_TEST_PARENT}/reasoningEngines/{_TEST_RESOURCE_ID}" +) +_TEST_SANDBOX_RESOURCE_NAME = ( + f"{_TEST_AGENT_ENGINE_RESOURCE_NAME}/sandboxes/{_TEST_SANDBOX_ID}" +) +_TEST_SNAPSHOT_RESOURCE_NAME = ( + f"{_TEST_SANDBOX_RESOURCE_NAME}/sandboxEnvironmentSnapshots/{_TEST_SNAPSHOT_ID}" +) + + +@pytest.fixture(scope="module") +def google_auth_mock(): + with mock.patch.object(auth, "default") as google_auth_mock: + google_auth_mock.return_value = ( + auth_credentials.AnonymousCredentials(), + _TEST_PROJECT, + ) + yield google_auth_mock + + +@pytest.mark.usefixtures("google_auth_mock") +class TestSandboxSnapshots: + + def setup_method(self): + importlib.reload(initializer) + importlib.reload(aiplatform) + importlib.reload(vertexai) + self.client = vertexai.Client( + project=_TEST_PROJECT, + location=_TEST_LOCATION, + credentials=_TEST_CREDENTIALS, + ) + + def teardown_method(self): + initializer.global_pool.shutdown(wait=True) + + def test_create_snapshot(self): + with mock.patch.object( + self.client.agent_engines.sandboxes._api_client, "request" + ) as request_mock: + request_mock.return_value = genai_types.HttpResponse( + body=b'{"expireTime": "2026-04-22T00:00:00Z"}', headers={} + ) + + snapshot = self.client.agent_engines.sandboxes.snapshots.create( + name=_TEST_SANDBOX_RESOURCE_NAME, + sandbox_environment_snapshot={}, + ) + + assert isinstance(snapshot.expire_time, datetime.datetime) + request_mock.assert_called_once() + args, _ = request_mock.call_args + assert args[0] == "post" + assert args[1] == f"{_TEST_SANDBOX_RESOURCE_NAME}:snapshot" + + def test_delete_snapshot(self): + with mock.patch.object( + self.client.agent_engines.sandboxes._api_client, "request" + ) as request_mock: + request_mock.return_value = genai_types.HttpResponse( + body=b'{"name": "operation-name"}', headers={} + ) + + operation = self.client.agent_engines.sandboxes.snapshots.delete( + name=_TEST_SNAPSHOT_RESOURCE_NAME, + ) + + assert operation.name == "operation-name" + request_mock.assert_called_once() + args, _ = request_mock.call_args + assert args[0] == "delete" + assert args[1] == _TEST_SNAPSHOT_RESOURCE_NAME + + def test_get_snapshot(self): + with mock.patch.object( + self.client.agent_engines.sandboxes._api_client, "request" + ) as request_mock: + request_mock.return_value = genai_types.HttpResponse( + body=b'{"expireTime": "2026-04-22T00:00:00Z"}', headers={} + ) + + snapshot = self.client.agent_engines.sandboxes.snapshots.get( + name=_TEST_SNAPSHOT_RESOURCE_NAME, + ) + + assert isinstance(snapshot.expire_time, datetime.datetime) + request_mock.assert_called_once() + args, _ = request_mock.call_args + assert args[0] == "get" + assert args[1] == _TEST_SNAPSHOT_RESOURCE_NAME + + def test_list_snapshots(self): + with mock.patch.object( + self.client.agent_engines.sandboxes._api_client, "request" + ) as request_mock: + request_mock.return_value = genai_types.HttpResponse( + body=( + b'{"sandboxEnvironmentSnapshots": [{"expireTime":' + b' "2026-04-22T00:00:00Z"}, {"expireTime":' + b' "2026-04-22T01:00:00Z"}]}' + ), + headers={}, + ) + + response = self.client.agent_engines.sandboxes.snapshots.list( + name=_TEST_SANDBOX_RESOURCE_NAME, + ) + + assert len(response.sandbox_environment_snapshots) == 2 + assert isinstance( + response.sandbox_environment_snapshots[0].expire_time, + datetime.datetime, + ) + assert isinstance( + response.sandbox_environment_snapshots[1].expire_time, + datetime.datetime, + ) + request_mock.assert_called_once() + args, _ = request_mock.call_args + assert args[0] == "get" + assert ( + args[1] == f"{_TEST_SANDBOX_RESOURCE_NAME}/sandboxEnvironmentSnapshots" + ) diff --git a/vertexai/_genai/sandbox_snapshots.py b/vertexai/_genai/sandbox_snapshots.py new file mode 100644 index 0000000000..ba59c1908f --- /dev/null +++ b/vertexai/_genai/sandbox_snapshots.py @@ -0,0 +1,744 @@ +# 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 json +import logging +from typing import Any, Optional, Union +from urllib.parse import urlencode + +from google.genai import _api_module +from google.genai import _common +from google.genai._common import get_value_by_path as getv +from google.genai._common import set_value_by_path as setv + +from . import types + +logger = logging.getLogger("vertexai_genai.sandboxsnapshots") + + +def _CreateAgentEngineSandboxSnapshotConfig_to_vertex( + from_object: Union[dict[str, Any], object], + parent_object: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + to_object: dict[str, Any] = {} + + if getv(from_object, ["display_name"]) is not None: + setv(parent_object, ["displayName"], getv(from_object, ["display_name"])) + + if getv(from_object, ["owner"]) is not None: + setv(parent_object, ["owner"], getv(from_object, ["owner"])) + + if getv(from_object, ["ttl"]) is not None: + setv(parent_object, ["ttl"], getv(from_object, ["ttl"])) + + return to_object + + +def _CreateSandboxEnvironmentSnapshotRequestParameters_to_vertex( + 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, ["source_sandbox_environment_name"]) is not None: + setv( + to_object, + ["_url", "name"], + getv(from_object, ["source_sandbox_environment_name"]), + ) + + if getv(from_object, ["config"]) is not None: + setv( + to_object, + ["config"], + _CreateAgentEngineSandboxSnapshotConfig_to_vertex( + getv(from_object, ["config"]), to_object + ), + ) + + return to_object + + +def _DeleteSandboxEnvironmentSnapshotRequestParameters_to_vertex( + 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 _GetSandboxEnvironmentSnapshotRequestParameters_to_vertex( + 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 _ListSandboxEnvironmentSnapshotsConfig_to_vertex( + 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"])) + + if getv(from_object, ["filter"]) is not None: + setv(parent_object, ["_query", "filter"], getv(from_object, ["filter"])) + + return to_object + + +def _ListSandboxEnvironmentSnapshotsRequestParameters_to_vertex( + 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: + _ListSandboxEnvironmentSnapshotsConfig_to_vertex( + getv(from_object, ["config"]), to_object + ) + + return to_object + + +class SandboxSnapshots(_api_module.BaseModule): + """Sandbox environment snapshot commands.""" + + def create( + self, + *, + source_sandbox_environment_name: str, + config: Optional[types.CreateAgentEngineSandboxSnapshotConfigOrDict] = None, + ) -> types.SandboxEnvironmentSnapshot: + """ + Snapshots a sandbox environment. + + """ + + parameter_model = types._CreateSandboxEnvironmentSnapshotRequestParameters( + source_sandbox_environment_name=source_sandbox_environment_name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _CreateSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}:snapshot".format_map(request_url_dict) + else: + path = "{name}:snapshot" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = self._api_client.request("post", path, request_dict, http_options) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.SandboxEnvironmentSnapshot._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.DeleteSandboxEnvironmentSnapshotConfigOrDict] = None, + ) -> types.DeleteSandboxEnvironmentSnapshotOperation: + """ + Deletes a sandbox environment snapshot. + + """ + + parameter_model = types._DeleteSandboxEnvironmentSnapshotRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _DeleteSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}".format_map(request_url_dict) + else: + path = "{name}" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = self._api_client.request("delete", path, request_dict, http_options) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.DeleteSandboxEnvironmentSnapshotOperation._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.GetSandboxEnvironmentSnapshotConfigOrDict] = None, + ) -> types.SandboxEnvironmentSnapshot: + """ + Gets a sandbox environment snapshot. + + """ + + parameter_model = types._GetSandboxEnvironmentSnapshotRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _GetSandboxEnvironmentSnapshotRequestParameters_to_vertex( + 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.SandboxEnvironmentSnapshot._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, + *, + name: str, + config: Optional[types.ListSandboxEnvironmentSnapshotsConfigOrDict] = None, + ) -> types.ListSandboxEnvironmentSnapshotsResponse: + """ + Lists sandbox environment snapshots. + + """ + + parameter_model = types._ListSandboxEnvironmentSnapshotsRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _ListSandboxEnvironmentSnapshotsRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}/sandboxEnvironmentSnapshots".format_map(request_url_dict) + else: + path = "{name}/sandboxEnvironmentSnapshots" + + 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.ListSandboxEnvironmentSnapshotsResponse._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 + + +class AsyncSandboxSnapshots(_api_module.BaseModule): + """Sandbox environment snapshot commands.""" + + async def create( + self, + *, + source_sandbox_environment_name: str, + config: Optional[types.CreateAgentEngineSandboxSnapshotConfigOrDict] = None, + ) -> types.SandboxEnvironmentSnapshot: + """ + Snapshots a sandbox environment. + + """ + + parameter_model = types._CreateSandboxEnvironmentSnapshotRequestParameters( + source_sandbox_environment_name=source_sandbox_environment_name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _CreateSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}:snapshot".format_map(request_url_dict) + else: + path = "{name}:snapshot" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = await self._api_client.async_request( + "post", path, request_dict, http_options + ) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.SandboxEnvironmentSnapshot._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.DeleteSandboxEnvironmentSnapshotConfigOrDict] = None, + ) -> types.DeleteSandboxEnvironmentSnapshotOperation: + """ + Deletes a sandbox environment snapshot. + + """ + + parameter_model = types._DeleteSandboxEnvironmentSnapshotRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _DeleteSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}".format_map(request_url_dict) + else: + path = "{name}" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = await self._api_client.async_request( + "delete", path, request_dict, http_options + ) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.DeleteSandboxEnvironmentSnapshotOperation._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.GetSandboxEnvironmentSnapshotConfigOrDict] = None, + ) -> types.SandboxEnvironmentSnapshot: + """ + Gets a sandbox environment snapshot. + + """ + + parameter_model = types._GetSandboxEnvironmentSnapshotRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _GetSandboxEnvironmentSnapshotRequestParameters_to_vertex( + 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.SandboxEnvironmentSnapshot._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, + *, + name: str, + config: Optional[types.ListSandboxEnvironmentSnapshotsConfigOrDict] = None, + ) -> types.ListSandboxEnvironmentSnapshotsResponse: + """ + Lists sandbox environment snapshots. + + """ + + parameter_model = types._ListSandboxEnvironmentSnapshotsRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _ListSandboxEnvironmentSnapshotsRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}/sandboxEnvironmentSnapshots".format_map(request_url_dict) + else: + path = "{name}/sandboxEnvironmentSnapshots" + + 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.ListSandboxEnvironmentSnapshotsResponse._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 diff --git a/vertexai/_genai/sandboxes.py b/vertexai/_genai/sandboxes.py index 565e1d80d8..510479bb7d 100644 --- a/vertexai/_genai/sandboxes.py +++ b/vertexai/_genai/sandboxes.py @@ -629,6 +629,23 @@ def _get_sandbox_operation( self._api_client._verify_response(return_value) return return_value + _snapshots = None + + @property + def snapshots(self) -> Any: + if self._snapshots is None: + try: + self._snapshots = __import__("importlib").import_module( + ".sandbox_snapshots", __package__ + ) + except ImportError as e: + raise ImportError( + "The 'agent_engines.sandboxes.sandbox_snapshots' module requires " + "additional packages. Please install them using pip install " + "google-cloud-aiplatform[sandbox_snapshots]" + ) from e + return self._snapshots.SandboxSnapshots(self._api_client) # type: ignore[attr-defined, no-any-return] + def create( self, *, diff --git a/vertexai/_genai/types/__init__.py b/vertexai/_genai/types/__init__.py index d278f1cce3..56fe3dcb01 100644 --- a/vertexai/_genai/types/__init__.py +++ b/vertexai/_genai/types/__init__.py @@ -40,6 +40,7 @@ from .common import _CreateEvaluationRunParameters from .common import _CreateEvaluationSetParameters from .common import _CreateMultimodalDatasetParameters +from .common import _CreateSandboxEnvironmentSnapshotRequestParameters from .common import _CustomJobParameters from .common import _CustomJobParameters from .common import _DeleteAgentEngineMemoryRequestParameters @@ -51,6 +52,7 @@ from .common import _DeleteEvaluationMetricParameters from .common import _DeleteMultimodalDatasetRequestParameters from .common import _DeletePromptVersionRequestParameters +from .common import _DeleteSandboxEnvironmentSnapshotRequestParameters from .common import _EvaluateInstancesRequestParameters from .common import _ExecuteCodeAgentEngineSandboxRequestParameters from .common import _GenerateAgentEngineMemoriesRequestParameters @@ -79,6 +81,7 @@ from .common import _GetEvaluationSetParameters from .common import _GetMultimodalDatasetOperationParameters from .common import _GetMultimodalDatasetParameters +from .common import _GetSandboxEnvironmentSnapshotRequestParameters from .common import _IngestEventsRequestParameters from .common import _ListAgentEngineMemoryRequestParameters from .common import _ListAgentEngineMemoryRevisionsRequestParameters @@ -92,6 +95,7 @@ from .common import _ListDatasetVersionsRequestParameters from .common import _ListEvaluationMetricsParameters from .common import _ListMultimodalDatasetsRequestParameters +from .common import _ListSandboxEnvironmentSnapshotsRequestParameters from .common import _OptimizeRequestParameters from .common import _OptimizeRequestParameters from .common import _PurgeAgentEngineMemoriesRequestParameters @@ -238,6 +242,9 @@ from .common import CreateAgentEngineSandboxConfig from .common import CreateAgentEngineSandboxConfigDict from .common import CreateAgentEngineSandboxConfigOrDict +from .common import CreateAgentEngineSandboxSnapshotConfig +from .common import CreateAgentEngineSandboxSnapshotConfigDict +from .common import CreateAgentEngineSandboxSnapshotConfigOrDict from .common import CreateAgentEngineSessionConfig from .common import CreateAgentEngineSessionConfigDict from .common import CreateAgentEngineSessionConfigOrDict @@ -331,6 +338,12 @@ from .common import DeletePromptVersionOperation from .common import DeletePromptVersionOperationDict from .common import DeletePromptVersionOperationOrDict +from .common import DeleteSandboxEnvironmentSnapshotConfig +from .common import DeleteSandboxEnvironmentSnapshotConfigDict +from .common import DeleteSandboxEnvironmentSnapshotConfigOrDict +from .common import DeleteSandboxEnvironmentSnapshotOperation +from .common import DeleteSandboxEnvironmentSnapshotOperationDict +from .common import DeleteSandboxEnvironmentSnapshotOperationOrDict from .common import DiskSpec from .common import DiskSpecDict from .common import DiskSpecOrDict @@ -550,6 +563,9 @@ from .common import GetPromptConfig from .common import GetPromptConfigDict from .common import GetPromptConfigOrDict +from .common import GetSandboxEnvironmentSnapshotConfig +from .common import GetSandboxEnvironmentSnapshotConfigDict +from .common import GetSandboxEnvironmentSnapshotConfigOrDict from .common import IdentityType from .common import Importance from .common import IngestEventsConfig @@ -641,6 +657,12 @@ from .common import ListReasoningEnginesSessionsResponse from .common import ListReasoningEnginesSessionsResponseDict from .common import ListReasoningEnginesSessionsResponseOrDict +from .common import ListSandboxEnvironmentSnapshotsConfig +from .common import ListSandboxEnvironmentSnapshotsConfigDict +from .common import ListSandboxEnvironmentSnapshotsConfigOrDict +from .common import ListSandboxEnvironmentSnapshotsResponse +from .common import ListSandboxEnvironmentSnapshotsResponseDict +from .common import ListSandboxEnvironmentSnapshotsResponseOrDict from .common import LLMMetric from .common import LossAnalysisConfig from .common import LossAnalysisConfigDict @@ -804,6 +826,7 @@ from .common import PointwiseMetricInstance from .common import PointwiseMetricInstanceDict from .common import PointwiseMetricInstanceOrDict +from .common import PostSnapshotAction from .common import Prompt from .common import PromptData from .common import PromptDataDict @@ -1033,6 +1056,9 @@ from .common import SandboxEnvironmentConnectionInfoOrDict from .common import SandboxEnvironmentDict from .common import SandboxEnvironmentOrDict +from .common import SandboxEnvironmentSnapshot +from .common import SandboxEnvironmentSnapshotDict +from .common import SandboxEnvironmentSnapshotOrDict from .common import SandboxEnvironmentSpec from .common import SandboxEnvironmentSpecCodeExecutionEnvironment from .common import SandboxEnvironmentSpecCodeExecutionEnvironmentDict @@ -2064,6 +2090,27 @@ "ListAgentEngineSandboxesResponse", "ListAgentEngineSandboxesResponseDict", "ListAgentEngineSandboxesResponseOrDict", + "CreateAgentEngineSandboxSnapshotConfig", + "CreateAgentEngineSandboxSnapshotConfigDict", + "CreateAgentEngineSandboxSnapshotConfigOrDict", + "SandboxEnvironmentSnapshot", + "SandboxEnvironmentSnapshotDict", + "SandboxEnvironmentSnapshotOrDict", + "DeleteSandboxEnvironmentSnapshotConfig", + "DeleteSandboxEnvironmentSnapshotConfigDict", + "DeleteSandboxEnvironmentSnapshotConfigOrDict", + "DeleteSandboxEnvironmentSnapshotOperation", + "DeleteSandboxEnvironmentSnapshotOperationDict", + "DeleteSandboxEnvironmentSnapshotOperationOrDict", + "GetSandboxEnvironmentSnapshotConfig", + "GetSandboxEnvironmentSnapshotConfigDict", + "GetSandboxEnvironmentSnapshotConfigOrDict", + "ListSandboxEnvironmentSnapshotsConfig", + "ListSandboxEnvironmentSnapshotsConfigDict", + "ListSandboxEnvironmentSnapshotsConfigOrDict", + "ListSandboxEnvironmentSnapshotsResponse", + "ListSandboxEnvironmentSnapshotsResponseDict", + "ListSandboxEnvironmentSnapshotsResponseOrDict", "CreateAgentEngineSessionConfig", "CreateAgentEngineSessionConfigDict", "CreateAgentEngineSessionConfigOrDict", @@ -2371,6 +2418,7 @@ "Operator", "Language", "MachineConfig", + "PostSnapshotAction", "Framework", "EvaluationItemType", "SamplingMethod", @@ -2456,6 +2504,10 @@ "_GetAgentEngineSandboxRequestParameters", "_ListAgentEngineSandboxesRequestParameters", "_GetAgentEngineSandboxOperationParameters", + "_CreateSandboxEnvironmentSnapshotRequestParameters", + "_DeleteSandboxEnvironmentSnapshotRequestParameters", + "_GetSandboxEnvironmentSnapshotRequestParameters", + "_ListSandboxEnvironmentSnapshotsRequestParameters", "_CreateAgentEngineSessionRequestParameters", "_DeleteAgentEngineSessionRequestParameters", "_GetAgentEngineSessionRequestParameters", diff --git a/vertexai/_genai/types/common.py b/vertexai/_genai/types/common.py index 126eeec663..fc96c561ca 100644 --- a/vertexai/_genai/types/common.py +++ b/vertexai/_genai/types/common.py @@ -324,6 +324,17 @@ class MachineConfig(_common.CaseInSensitiveEnum): """The default value: milligcu 4000, memory 4 Gib""" +class PostSnapshotAction(_common.CaseInSensitiveEnum): + """Input only. Action to take on the source SandboxEnvironment after the snapshot is taken. This field is only used in CreateSandboxEnvironmentSnapshotRequest and it is not stored in the resource.""" + + POST_SNAPSHOT_ACTION_UNSPECIFIED = "POST_SNAPSHOT_ACTION_UNSPECIFIED" + """The default value. This value is unused.""" + RUNNING = "RUNNING" + """Sandbox environment will continue to run after snapshot is taken.""" + PAUSE = "PAUSE" + """Sandbox environment will be paused after snapshot is taken.""" + + class Framework(_common.CaseInSensitiveEnum): """Framework used to build the application.""" @@ -7502,7 +7513,7 @@ class ReasoningEngineSpecDeploymentSpec(_common.BaseModel): ) min_instances: Optional[int] = Field( default=None, - description="""Optional. The minimum number of application instances that will be kept running at all times. Defaults to 1. Range: [0, 10].""", + description="""Optional. The minimum number of application instances that will be kept running at all times. Defaults to 1. Range: [0, 75].""", ) psc_interface_config: Optional[PscInterfaceConfig] = Field( default=None, description="""Optional. Configuration for PSC-I.""" @@ -7543,7 +7554,7 @@ class ReasoningEngineSpecDeploymentSpecDict(TypedDict, total=False): """Optional. The maximum number of application instances that can be launched to handle increased traffic. Defaults to 100. Range: [1, 1000]. If VPC-SC or PSC-I is enabled, the acceptable range is [1, 100].""" min_instances: Optional[int] - """Optional. The minimum number of application instances that will be kept running at all times. Defaults to 1. Range: [0, 10].""" + """Optional. The minimum number of application instances that will be kept running at all times. Defaults to 1. Range: [0, 75].""" psc_interface_config: Optional[PscInterfaceConfigDict] """Optional. Configuration for PSC-I.""" @@ -11788,6 +11799,10 @@ class SandboxEnvironmentConnectionInfo(_common.BaseModel): default=None, description="""Output only. The hostname of the SandboxEnvironment.""", ) + routing_token: Optional[str] = Field( + default=None, + description="""Output only. The routing token for the SandboxEnvironment.""", + ) class SandboxEnvironmentConnectionInfoDict(TypedDict, total=False): @@ -11805,6 +11820,9 @@ class SandboxEnvironmentConnectionInfoDict(TypedDict, total=False): sandbox_hostname: Optional[str] """Output only. The hostname of the SandboxEnvironment.""" + routing_token: Optional[str] + """Output only. The routing token for the SandboxEnvironment.""" + SandboxEnvironmentConnectionInfoOrDict = Union[ SandboxEnvironmentConnectionInfo, SandboxEnvironmentConnectionInfoDict @@ -11850,6 +11868,22 @@ class SandboxEnvironment(_common.BaseModel): default=None, description="""Output only. The timestamp when this SandboxEnvironment was most recently updated.""", ) + latest_sandbox_environment_snapshot: Optional[str] = Field( + default=None, + description="""Output only. The resource name of the latest snapshot taken for this SandboxEnvironment.""", + ) + owner: Optional[str] = Field( + default=None, + description="""Optional. Owner information for this sandbox environment. A Sandbox can only be restored from a snapshot that belongs to the same owner. If not set, sandbox will be created as the default owner.""", + ) + sandbox_environment_snapshot: Optional[str] = Field( + default=None, + description="""Optional. The resource name of the SandboxEnvironmentSnapshot to use for creating this SandboxEnvironment. Format: `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine}/sandboxEnvironmentSnapshots/{sandbox_environment_snapshot}`""", + ) + sandbox_environment_template: Optional[str] = Field( + default=None, + description="""Optional. The name of the SandboxEnvironmentTemplate specified in the parent Agent Engine resource that this SandboxEnvironment is created from. Only one of `sandbox_environment_template` and `spec` should be set.""", + ) class SandboxEnvironmentDict(TypedDict, total=False): @@ -11883,6 +11917,18 @@ class SandboxEnvironmentDict(TypedDict, total=False): update_time: Optional[datetime.datetime] """Output only. The timestamp when this SandboxEnvironment was most recently updated.""" + latest_sandbox_environment_snapshot: Optional[str] + """Output only. The resource name of the latest snapshot taken for this SandboxEnvironment.""" + + owner: Optional[str] + """Optional. Owner information for this sandbox environment. A Sandbox can only be restored from a snapshot that belongs to the same owner. If not set, sandbox will be created as the default owner.""" + + sandbox_environment_snapshot: Optional[str] + """Optional. The resource name of the SandboxEnvironmentSnapshot to use for creating this SandboxEnvironment. Format: `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine}/sandboxEnvironmentSnapshots/{sandbox_environment_snapshot}`""" + + sandbox_environment_template: Optional[str] + """Optional. The name of the SandboxEnvironmentTemplate specified in the parent Agent Engine resource that this SandboxEnvironment is created from. Only one of `sandbox_environment_template` and `spec` should be set.""" + SandboxEnvironmentOrDict = Union[SandboxEnvironment, SandboxEnvironmentDict] @@ -12316,6 +12362,396 @@ class _GetAgentEngineSandboxOperationParametersDict(TypedDict, total=False): ] +class CreateAgentEngineSandboxSnapshotConfig(_common.BaseModel): + """Config for creating a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptions] = Field( + default=None, description="""Used to override HTTP request options.""" + ) + display_name: Optional[str] = Field( + default=None, description="""The display name of the sandbox snapshot.""" + ) + owner: Optional[str] = Field( + default=None, description="""The owner of the sandbox snapshot.""" + ) + ttl: Optional[str] = Field( + default=None, + description="""The TTL for this resource. The expiration time is computed: now + TTL.""", + ) + + +class CreateAgentEngineSandboxSnapshotConfigDict(TypedDict, total=False): + """Config for creating a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptionsDict] + """Used to override HTTP request options.""" + + display_name: Optional[str] + """The display name of the sandbox snapshot.""" + + owner: Optional[str] + """The owner of the sandbox snapshot.""" + + ttl: Optional[str] + """The TTL for this resource. The expiration time is computed: now + TTL.""" + + +CreateAgentEngineSandboxSnapshotConfigOrDict = Union[ + CreateAgentEngineSandboxSnapshotConfig, CreateAgentEngineSandboxSnapshotConfigDict +] + + +class _CreateSandboxEnvironmentSnapshotRequestParameters(_common.BaseModel): + """Parameters for creating a sandbox environment snapshot.""" + + source_sandbox_environment_name: Optional[str] = Field( + default=None, description="""Name of the sandbox environment to snapshot.""" + ) + config: Optional[CreateAgentEngineSandboxSnapshotConfig] = Field( + default=None, description="""""" + ) + + +class _CreateSandboxEnvironmentSnapshotRequestParametersDict(TypedDict, total=False): + """Parameters for creating a sandbox environment snapshot.""" + + source_sandbox_environment_name: Optional[str] + """Name of the sandbox environment to snapshot.""" + + config: Optional[CreateAgentEngineSandboxSnapshotConfigDict] + """""" + + +_CreateSandboxEnvironmentSnapshotRequestParametersOrDict = Union[ + _CreateSandboxEnvironmentSnapshotRequestParameters, + _CreateSandboxEnvironmentSnapshotRequestParametersDict, +] + + +class SandboxEnvironmentSnapshot(_common.BaseModel): + """A sandbox environment snapshot.""" + + display_name: Optional[str] = Field( + default=None, + description="""The display name of the sandbox environment snapshot.""", + ) + expire_time: Optional[datetime.datetime] = Field( + default=None, + description="""Expiration time of the sandbox environment snapshot. + """, + ) + create_time: Optional[datetime.datetime] = Field( + default=None, + description="""Output only. The timestamp when this SandboxEnvironmentSnapshot was created.""", + ) + name: Optional[str] = Field( + default=None, + description="""Identifier. The resource name of the SandboxEnvironmentSnapshot. Format: `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine}/sandboxEnvironmentSnapshots/{sandbox_environment_snapshot}`""", + ) + owner: Optional[str] = Field( + default=None, + description="""Optional. Owner information for this sandbox snapshot. Different owners will have isolations on snapshot storage and identity. If not set, snapshot will be created as the default owner.""", + ) + parent_snapshot: Optional[str] = Field( + default=None, + description="""Output only. The resource name of the parent SandboxEnvironmentSnapshot. Empty if this is a root Snapshot (the first snapshot from a newly created sandbox). Can be used to reconstruct the whole ancestry tree of snapshots.""", + ) + post_snapshot_action: Optional[PostSnapshotAction] = Field( + default=None, + description="""Optional. Input only. Action to take on the source SandboxEnvironment after the snapshot is taken. This field is only used in CreateSandboxEnvironmentSnapshotRequest and it is not stored in the resource.""", + ) + size_bytes: Optional[int] = Field( + default=None, + description="""Optional. Output only. Size of the snapshot data in bytes.""", + ) + source_sandbox_environment: Optional[str] = Field( + default=None, + description="""Required. The resource name of the source SandboxEnvironment this snapshot was taken from.""", + ) + ttl: Optional[str] = Field( + default=None, + description="""Optional. Input only. The TTL for the sandbox environment snapshot. The expiration time is computed: now + TTL.""", + ) + update_time: Optional[datetime.datetime] = Field( + default=None, + description="""Output only. The timestamp when this SandboxEnvironment was most recently updated.""", + ) + + +class SandboxEnvironmentSnapshotDict(TypedDict, total=False): + """A sandbox environment snapshot.""" + + display_name: Optional[str] + """The display name of the sandbox environment snapshot.""" + + expire_time: Optional[datetime.datetime] + """Expiration time of the sandbox environment snapshot. + """ + + create_time: Optional[datetime.datetime] + """Output only. The timestamp when this SandboxEnvironmentSnapshot was created.""" + + name: Optional[str] + """Identifier. The resource name of the SandboxEnvironmentSnapshot. Format: `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine}/sandboxEnvironmentSnapshots/{sandbox_environment_snapshot}`""" + + owner: Optional[str] + """Optional. Owner information for this sandbox snapshot. Different owners will have isolations on snapshot storage and identity. If not set, snapshot will be created as the default owner.""" + + parent_snapshot: Optional[str] + """Output only. The resource name of the parent SandboxEnvironmentSnapshot. Empty if this is a root Snapshot (the first snapshot from a newly created sandbox). Can be used to reconstruct the whole ancestry tree of snapshots.""" + + post_snapshot_action: Optional[PostSnapshotAction] + """Optional. Input only. Action to take on the source SandboxEnvironment after the snapshot is taken. This field is only used in CreateSandboxEnvironmentSnapshotRequest and it is not stored in the resource.""" + + size_bytes: Optional[int] + """Optional. Output only. Size of the snapshot data in bytes.""" + + source_sandbox_environment: Optional[str] + """Required. The resource name of the source SandboxEnvironment this snapshot was taken from.""" + + ttl: Optional[str] + """Optional. Input only. The TTL for the sandbox environment snapshot. The expiration time is computed: now + TTL.""" + + update_time: Optional[datetime.datetime] + """Output only. The timestamp when this SandboxEnvironment was most recently updated.""" + + +SandboxEnvironmentSnapshotOrDict = Union[ + SandboxEnvironmentSnapshot, SandboxEnvironmentSnapshotDict +] + + +class DeleteSandboxEnvironmentSnapshotConfig(_common.BaseModel): + """Config for deleting a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptions] = Field( + default=None, description="""Used to override HTTP request options.""" + ) + + +class DeleteSandboxEnvironmentSnapshotConfigDict(TypedDict, total=False): + """Config for deleting a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptionsDict] + """Used to override HTTP request options.""" + + +DeleteSandboxEnvironmentSnapshotConfigOrDict = Union[ + DeleteSandboxEnvironmentSnapshotConfig, DeleteSandboxEnvironmentSnapshotConfigDict +] + + +class _DeleteSandboxEnvironmentSnapshotRequestParameters(_common.BaseModel): + """Parameters for deleting sandbox environment snapshots.""" + + name: Optional[str] = Field( + default=None, + description="""Name of the sandbox environment snapshot to delete.""", + ) + config: Optional[DeleteSandboxEnvironmentSnapshotConfig] = Field( + default=None, description="""""" + ) + + +class _DeleteSandboxEnvironmentSnapshotRequestParametersDict(TypedDict, total=False): + """Parameters for deleting sandbox environment snapshots.""" + + name: Optional[str] + """Name of the sandbox environment snapshot to delete.""" + + config: Optional[DeleteSandboxEnvironmentSnapshotConfigDict] + """""" + + +_DeleteSandboxEnvironmentSnapshotRequestParametersOrDict = Union[ + _DeleteSandboxEnvironmentSnapshotRequestParameters, + _DeleteSandboxEnvironmentSnapshotRequestParametersDict, +] + + +class DeleteSandboxEnvironmentSnapshotOperation(_common.BaseModel): + """Operation for deleting sandbox environment snapshots.""" + + name: Optional[str] = Field( + default=None, + description="""The server-assigned name, which is only unique within the same service that originally returns it. If you use the default HTTP mapping, the `name` should be a resource name ending with `operations/{unique_id}`.""", + ) + metadata: Optional[dict[str, Any]] = Field( + default=None, + description="""Service-specific metadata associated with the operation. It typically contains progress information and common metadata such as create time. Some services might not provide such metadata. Any method that returns a long-running operation should document the metadata type, if any.""", + ) + done: Optional[bool] = Field( + default=None, + description="""If the value is `false`, it means the operation is still in progress. If `true`, the operation is completed, and either `error` or `response` is available.""", + ) + error: Optional[dict[str, Any]] = Field( + default=None, + description="""The error result of the operation in case of failure or cancellation.""", + ) + + +class DeleteSandboxEnvironmentSnapshotOperationDict(TypedDict, total=False): + """Operation for deleting sandbox environment snapshots.""" + + name: Optional[str] + """The server-assigned name, which is only unique within the same service that originally returns it. If you use the default HTTP mapping, the `name` should be a resource name ending with `operations/{unique_id}`.""" + + metadata: Optional[dict[str, Any]] + """Service-specific metadata associated with the operation. It typically contains progress information and common metadata such as create time. Some services might not provide such metadata. Any method that returns a long-running operation should document the metadata type, if any.""" + + done: Optional[bool] + """If the value is `false`, it means the operation is still in progress. If `true`, the operation is completed, and either `error` or `response` is available.""" + + error: Optional[dict[str, Any]] + """The error result of the operation in case of failure or cancellation.""" + + +DeleteSandboxEnvironmentSnapshotOperationOrDict = Union[ + DeleteSandboxEnvironmentSnapshotOperation, + DeleteSandboxEnvironmentSnapshotOperationDict, +] + + +class GetSandboxEnvironmentSnapshotConfig(_common.BaseModel): + """Config for getting a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptions] = Field( + default=None, description="""Used to override HTTP request options.""" + ) + + +class GetSandboxEnvironmentSnapshotConfigDict(TypedDict, total=False): + """Config for getting a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptionsDict] + """Used to override HTTP request options.""" + + +GetSandboxEnvironmentSnapshotConfigOrDict = Union[ + GetSandboxEnvironmentSnapshotConfig, GetSandboxEnvironmentSnapshotConfigDict +] + + +class _GetSandboxEnvironmentSnapshotRequestParameters(_common.BaseModel): + """Parameters for getting a sandbox environment snapshot.""" + + name: Optional[str] = Field( + default=None, description="""Name of the sandbox environment snapshot.""" + ) + config: Optional[GetSandboxEnvironmentSnapshotConfig] = Field( + default=None, description="""""" + ) + + +class _GetSandboxEnvironmentSnapshotRequestParametersDict(TypedDict, total=False): + """Parameters for getting a sandbox environment snapshot.""" + + name: Optional[str] + """Name of the sandbox environment snapshot.""" + + config: Optional[GetSandboxEnvironmentSnapshotConfigDict] + """""" + + +_GetSandboxEnvironmentSnapshotRequestParametersOrDict = Union[ + _GetSandboxEnvironmentSnapshotRequestParameters, + _GetSandboxEnvironmentSnapshotRequestParametersDict, +] + + +class ListSandboxEnvironmentSnapshotsConfig(_common.BaseModel): + """Config for listing sandbox environment snapshots.""" + + http_options: Optional[genai_types.HttpOptions] = Field( + default=None, description="""Used to override HTTP request options.""" + ) + page_size: Optional[int] = Field(default=None, description="""""") + page_token: Optional[str] = Field(default=None, description="""""") + filter: Optional[str] = Field( + default=None, + description="""An expression for filtering the results of the request.""", + ) + + +class ListSandboxEnvironmentSnapshotsConfigDict(TypedDict, total=False): + """Config for listing sandbox environment snapshots.""" + + http_options: Optional[genai_types.HttpOptionsDict] + """Used to override HTTP request options.""" + + page_size: Optional[int] + """""" + + page_token: Optional[str] + """""" + + filter: Optional[str] + """An expression for filtering the results of the request.""" + + +ListSandboxEnvironmentSnapshotsConfigOrDict = Union[ + ListSandboxEnvironmentSnapshotsConfig, ListSandboxEnvironmentSnapshotsConfigDict +] + + +class _ListSandboxEnvironmentSnapshotsRequestParameters(_common.BaseModel): + """Parameters for listing sandbox environment snapshots.""" + + name: Optional[str] = Field( + default=None, + description="""Name of the reasoning engine to list snapshots from.""", + ) + config: Optional[ListSandboxEnvironmentSnapshotsConfig] = Field( + default=None, description="""""" + ) + + +class _ListSandboxEnvironmentSnapshotsRequestParametersDict(TypedDict, total=False): + """Parameters for listing sandbox environment snapshots.""" + + name: Optional[str] + """Name of the reasoning engine to list snapshots from.""" + + config: Optional[ListSandboxEnvironmentSnapshotsConfigDict] + """""" + + +_ListSandboxEnvironmentSnapshotsRequestParametersOrDict = Union[ + _ListSandboxEnvironmentSnapshotsRequestParameters, + _ListSandboxEnvironmentSnapshotsRequestParametersDict, +] + + +class ListSandboxEnvironmentSnapshotsResponse(_common.BaseModel): + """Response for listing sandbox environment snapshots.""" + + sdk_http_response: Optional[genai_types.HttpResponse] = Field( + default=None, description="""Used to retain the full HTTP response.""" + ) + next_page_token: Optional[str] = Field(default=None, description="""""") + sandbox_environment_snapshots: Optional[list[SandboxEnvironmentSnapshot]] = Field( + default=None, description="""List of sandbox environment snapshots.""" + ) + + +class ListSandboxEnvironmentSnapshotsResponseDict(TypedDict, total=False): + """Response for listing sandbox environment snapshots.""" + + sdk_http_response: Optional[genai_types.HttpResponseDict] + """Used to retain the full HTTP response.""" + + next_page_token: Optional[str] + """""" + + sandbox_environment_snapshots: Optional[list[SandboxEnvironmentSnapshotDict]] + """List of sandbox environment snapshots.""" + + +ListSandboxEnvironmentSnapshotsResponseOrDict = Union[ + ListSandboxEnvironmentSnapshotsResponse, ListSandboxEnvironmentSnapshotsResponseDict +] + + class CreateAgentEngineSessionConfig(_common.BaseModel): """Config for creating a Session."""