# 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. # # pyformat: disable # pylint: skip-file """Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT.""" from .basesdk import AsyncBaseSDK, BaseSDK from .sdkconfiguration import SDKConfiguration from . import errors, models, types, utils from ._hooks import AfterParseErrorContext, HookContext, ResponseContext from .files import AsyncFiles, Files from .types import OptionalNullable, UNSET, environments, interactions from .types.environments import ( createenvironmentrequest as environments_createenvironmentrequest, ) from .types.interactions import source as interactions_source from .utils import get_security_from_env, response_helpers from .utils.unmarshal_json_response import unmarshal_json_response import httpx from typing import Any, Iterable, List, Mapping, Optional, Union, cast class Environments(BaseSDK): @property def with_raw_response(self): return EnvironmentsWithRawResponse(self) @property def with_streaming_response(self): return EnvironmentsWithStreamingResponse(self) files: Files def __init__( self, sdk_config: SDKConfiguration, parent_ref: Optional[object] = None ) -> None: BaseSDK.__init__(self, sdk_config, parent_ref=parent_ref) self.sdk_configuration = sdk_config self._init_sdks() def _init_sdks(self): self.files = Files(self.sdk_configuration, parent_ref=self.parent_ref) def list_environments( self, *, api_version: Optional[str] = None, page_size: Optional[int] = None, page_token: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> environments.ListEnvironmentsResponse: r"""Lists environments. :param api_version: Which version of the API to use. :param page_size: Optional. Maximum number of environments to return.\nIf unspecified, defaults to 50. Maximum is 1000. :param page_token: Optional. Pagination token. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.ListEnvironmentsRequest( api_version=api_version, page_size=page_size, page_token=page_token, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="GET", path="/{api_version}/environments", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.ListEnvironmentsGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) def _speakeasy_parse_response(http_res): if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "default", "application/json"): return unmarshal_json_response( environments.ListEnvironmentsResponse, http_res, validate=False ) raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="ListEnvironments", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "list", "lang": "sh", "source": 'curl -X GET https://generativelanguage.googleapis.com/v1beta/environments \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "list", "lang": "python", "source": "from google import genai\n\nclient = genai.Client()\nresponse = client.environments.list()\nfor environment in response.environments or []:\n print(environment.id)\n", }, { "label": "list", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst response = await ai.environments.list();\nfor (const env of (response.environments ?? [])) {\n console.log(env.id);\n}\n", }, { "label": "list", "lang": "java", "source": 'import com.google.genai.Client;\nimport com.google.genai.gaos.models.environments.Environment;\nimport com.google.genai.gaos.models.operations.ListEnvironmentsResponse;\nimport java.util.List;\n\nClient client = new Client();\nListEnvironmentsResponse response = client.environments.listEnvironmentsDirect();\nfor (Environment env : response.listEnvironmentsResponse().flatMap(res -> res.environments()).orElse(List.of())) {\n System.out.println(env.id().orElse(""));\n}\n', }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="sync"), ) http_res = self.do_request( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): http_res.read() try: _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.StreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.APIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), ), ) try: return _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) def create_environment( self, *, api_version: Optional[str] = None, from_environment: Optional[str] = None, network: Optional[ Union[ environments_createenvironmentrequest.CreateEnvironmentRequestNetworkUnion, environments_createenvironmentrequest.CreateEnvironmentRequestNetworkUnionParam, ] ] = None, sources: Optional[ Union[ Iterable[interactions_source.Source], Iterable[interactions_source.SourceParam], ] ] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> environments.Environment: r"""Creates an environment. :param api_version: Which version of the API to use. :param from_environment: Optional. The source environment to copy/fork from. Format: `environments/{environment_id}` or `{environment_id}`. When specified, `sources` and `env` must be empty. :param network: Network configuration for the environment. :param sources: Sources to be mounted into the environment. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.CreateEnvironmentRequest( api_version=api_version, body=environments.CreateEnvironmentRequest( from_environment=from_environment, network=utils.get_pydantic_model( network, Optional[environments.CreateEnvironmentRequestNetworkUnion] ), sources=utils.get_pydantic_model( sources, Optional[List[interactions.Source]] ), ), ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="POST", path="/{api_version}/environments", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.CreateEnvironmentGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.body, False, False, "json", environments.CreateEnvironmentRequest, extra_body=extra_body, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) def _speakeasy_parse_response(http_res): if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "default", "application/json"): return unmarshal_json_response( environments.Environment, http_res, validate=False ) raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="CreateEnvironment", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "create", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/environments \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "sources": [{\n "type": "inline",\n "target": "main.py",\n "content": "print(\\"Hello, World!\\")"\n }]\n }\'\n', }, { "label": "create", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\nenvironment = client.environments.create(\n sources=[{\n "type": "inline",\n "target": "main.py",\n "content": "print(\'Hello, World!\')",\n }]\n)\nprint(environment.id)\n\n# [cleanup]\nclient.environments.delete(id=environment.id)\n# [/cleanup]\n', }, { "label": "create", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst environment = await ai.environments.create({\n sources: [{\n type: 'inline',\n target: 'main.py',\n content: \"print('Hello, World!')\",\n }],\n});\nif (!environment.id) {\n throw new Error('Environment creation failed: ID is undefined');\n}\nconsole.log(environment.id);\n\n// [cleanup]\nawait ai.environments.delete(environment.id);\n// [/cleanup]\n", }, { "label": "create", "lang": "java", "source": 'import com.google.genai.Client;\nimport com.google.genai.gaos.models.environments.CreateEnvironmentRequest;\nimport com.google.genai.gaos.models.environments.Environment;\nimport com.google.genai.gaos.models.interactions.Source;\nimport com.google.genai.gaos.models.interactions.SourceType;\nimport java.util.List;\n\nClient client = new Client();\nCreateEnvironmentRequest request = CreateEnvironmentRequest.builder()\n .sources(List.of(Source.builder()\n .type(SourceType.INLINE)\n .target("main.py")\n .content("print(\'Hello, World!\')")\n .build()))\n .build();\nEnvironment environment = client.environments.createEnvironment(request).environment().orElseThrow();\nSystem.out.println(environment.id().orElse(""));\n\n// [cleanup]\nenvironment.id().ifPresent(id -> client.environments.deleteEnvironment(id));\n// [/cleanup]\n', }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="sync"), ) http_res = self.do_request( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): http_res.read() try: _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.StreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.APIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), ), ) try: return _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) def delete_environment( self, id: str, *, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Empty: r"""Deletes an environment. :param id: Resource ID segment making up resource `name`. It identifies the resource within its parent collection as described in https://google.aip.dev/122. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.DeleteEnvironmentRequest( api_version=api_version, id=id, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="DELETE", path="/{api_version}/environments/{id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.DeleteEnvironmentGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) def _speakeasy_parse_response(http_res): if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "default", "application/json"): return unmarshal_json_response( interactions.Empty, http_res, validate=False ) raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="DeleteEnvironment", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "delete", "lang": "sh", "source": 'curl -X DELETE https://generativelanguage.googleapis.com/v1beta/environments/env_abc123 \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "delete", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# [setup]\ncreated = client.environments.create(\n sources=[{"type": "inline", "target": "main.py", "content": "print(\'Hello\')"}]\n)\n# [/setup]\n\nclient.environments.delete(id=created.id)\nprint("Environment deleted successfully.")\n', }, { "label": "delete", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// [setup]\nconst created = await ai.environments.create({\n sources: [{type: 'inline', target: 'main.py', content: \"print('Hello')\"}],\n});\nif (!created.id) throw new Error('Failed to create environment');\n// [/setup]\n\nawait ai.environments.delete(created.id);\nconsole.log('Environment deleted successfully.');\n", }, { "label": "delete", "lang": "java", "source": 'import com.google.genai.Client;\nimport com.google.genai.gaos.models.environments.CreateEnvironmentRequest;\nimport com.google.genai.gaos.models.environments.Environment;\nimport com.google.genai.gaos.models.interactions.Source;\nimport com.google.genai.gaos.models.interactions.SourceType;\nimport java.util.List;\n\nClient client = new Client();\n\n// [setup]\nCreateEnvironmentRequest request = CreateEnvironmentRequest.builder()\n .sources(List.of(Source.builder()\n .type(SourceType.INLINE)\n .target("main.py")\n .content("print(\'Hello\')")\n .build()))\n .build();\nEnvironment created = client.environments.createEnvironment(request).environment().orElseThrow();\nString envId = created.id().orElseThrow();\n// [/setup]\n\nclient.environments.deleteEnvironment(envId);\nSystem.out.println("Environment deleted successfully.");\n', }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="sync"), ) http_res = self.do_request( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): http_res.read() try: _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.StreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.APIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), ), ) try: return _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) def get_environment( self, id: str, *, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> environments.Environment: r"""Gets an environment. :param id: Resource ID segment making up resource `name`. It identifies the resource within its parent collection as described in https://google.aip.dev/122. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetEnvironmentRequest( api_version=api_version, id=id, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="GET", path="/{api_version}/environments/{id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.GetEnvironmentGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) def _speakeasy_parse_response(http_res): if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "default", "application/json"): return unmarshal_json_response( environments.Environment, http_res, validate=False ) raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="GetEnvironment", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "get", "lang": "sh", "source": 'curl -X GET https://generativelanguage.googleapis.com/v1beta/environments/env_abc123 \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "get", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# [setup]\ncreated = client.environments.create(\n sources=[{"type": "inline", "target": "main.py", "content": "print(\'Hello\')"}]\n)\n# [/setup]\n\nenvironment = client.environments.get(id=created.id)\nprint(environment.id)\n\n# [cleanup]\nclient.environments.delete(id=created.id)\n# [/cleanup]\n', }, { "label": "get", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// [setup]\nconst created = await ai.environments.create({\n sources: [{type: 'inline', target: 'main.py', content: \"print('Hello')\"}],\n});\nif (!created.id) throw new Error('Failed to create environment');\n// [/setup]\n\nconst environment = await ai.environments.get(created.id);\nconsole.log(environment.id);\n\n// [cleanup]\nawait ai.environments.delete(created.id);\n// [/cleanup]\n", }, { "label": "get", "lang": "java", "source": 'import com.google.genai.Client;\nimport com.google.genai.gaos.models.environments.CreateEnvironmentRequest;\nimport com.google.genai.gaos.models.environments.Environment;\nimport com.google.genai.gaos.models.interactions.Source;\nimport com.google.genai.gaos.models.interactions.SourceType;\nimport java.util.List;\n\nClient client = new Client();\n\n// [setup]\nCreateEnvironmentRequest request = CreateEnvironmentRequest.builder()\n .sources(List.of(Source.builder()\n .type(SourceType.INLINE)\n .target("main.py")\n .content("print(\'Hello\')")\n .build()))\n .build();\nEnvironment created = client.environments.createEnvironment(request).environment().orElseThrow();\nString envId = created.id().orElseThrow();\n// [/setup]\n\nEnvironment environment = client.environments.getEnvironment(envId).environment().orElseThrow();\nSystem.out.println(environment.id().orElse(""));\n\n// [cleanup]\nclient.environments.deleteEnvironment(envId);\n// [/cleanup]\n', }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="sync"), ) http_res = self.do_request( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): http_res.read() try: _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.StreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.APIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), ), ) try: return _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) class EnvironmentsWithRawResponse: def __init__(self, sdk: Environments) -> None: self._sdk = sdk self.list_environments = response_helpers.to_raw_response_wrapper( sdk.list_environments, "extra_headers" ) self.create_environment = response_helpers.to_raw_response_wrapper( sdk.create_environment, "extra_headers" ) self.delete_environment = response_helpers.to_raw_response_wrapper( sdk.delete_environment, "extra_headers" ) self.get_environment = response_helpers.to_raw_response_wrapper( sdk.get_environment, "extra_headers" ) @property def files(self): return self._sdk.files.with_raw_response class EnvironmentsWithStreamingResponse: def __init__(self, sdk: Environments) -> None: self._sdk = sdk self.list_environments = response_helpers.to_streamed_response_wrapper( sdk.list_environments, "extra_headers" ) self.create_environment = response_helpers.to_streamed_response_wrapper( sdk.create_environment, "extra_headers" ) self.delete_environment = response_helpers.to_streamed_response_wrapper( sdk.delete_environment, "extra_headers" ) self.get_environment = response_helpers.to_streamed_response_wrapper( sdk.get_environment, "extra_headers" ) @property def files(self): return self._sdk.files.with_streaming_response class AsyncEnvironments(AsyncBaseSDK): @property def with_raw_response(self): return AsyncEnvironmentsWithRawResponse(self) @property def with_streaming_response(self): return AsyncEnvironmentsWithStreamingResponse(self) files: AsyncFiles def __init__( self, sdk_config: SDKConfiguration, parent_ref: Optional[object] = None ) -> None: AsyncBaseSDK.__init__(self, sdk_config, parent_ref=parent_ref) self.sdk_configuration = sdk_config self._init_sdks() def _init_sdks(self): self.files = AsyncFiles(self.sdk_configuration, parent_ref=self.parent_ref) async def list_environments( self, *, api_version: Optional[str] = None, page_size: Optional[int] = None, page_token: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> environments.ListEnvironmentsResponse: r"""Lists environments. :param api_version: Which version of the API to use. :param page_size: Optional. Maximum number of environments to return.\nIf unspecified, defaults to 50. Maximum is 1000. :param page_token: Optional. Pagination token. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.ListEnvironmentsRequest( api_version=api_version, page_size=page_size, page_token=page_token, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="GET", path="/{api_version}/environments", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.ListEnvironmentsGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) async def _speakeasy_parse_response(http_res): if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "default", "application/json"): return unmarshal_json_response( environments.ListEnvironmentsResponse, http_res, validate=False ) raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="ListEnvironments", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "list", "lang": "sh", "source": 'curl -X GET https://generativelanguage.googleapis.com/v1beta/environments \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "list", "lang": "python", "source": "from google import genai\n\nclient = genai.Client()\nresponse = client.environments.list()\nfor environment in response.environments or []:\n print(environment.id)\n", }, { "label": "list", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst response = await ai.environments.list();\nfor (const env of (response.environments ?? [])) {\n console.log(env.id);\n}\n", }, { "label": "list", "lang": "java", "source": 'import com.google.genai.Client;\nimport com.google.genai.gaos.models.environments.Environment;\nimport com.google.genai.gaos.models.operations.ListEnvironmentsResponse;\nimport java.util.List;\n\nClient client = new Client();\nListEnvironmentsResponse response = client.environments.listEnvironmentsDirect();\nfor (Environment env : response.listEnvironmentsResponse().flatMap(res -> res.environments()).orElse(List.of())) {\n System.out.println(env.id().orElse(""));\n}\n', }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="async"), ) http_res = await self.do_request_async( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): await http_res.aread() try: await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.AsyncStreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.AsyncAPIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), async_hooks=self.sdk_configuration.__dict__.get("_async_hooks"), ), ) try: return await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) async def create_environment( self, *, api_version: Optional[str] = None, from_environment: Optional[str] = None, network: Optional[ Union[ environments_createenvironmentrequest.CreateEnvironmentRequestNetworkUnion, environments_createenvironmentrequest.CreateEnvironmentRequestNetworkUnionParam, ] ] = None, sources: Optional[ Union[ Iterable[interactions_source.Source], Iterable[interactions_source.SourceParam], ] ] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> environments.Environment: r"""Creates an environment. :param api_version: Which version of the API to use. :param from_environment: Optional. The source environment to copy/fork from. Format: `environments/{environment_id}` or `{environment_id}`. When specified, `sources` and `env` must be empty. :param network: Network configuration for the environment. :param sources: Sources to be mounted into the environment. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.CreateEnvironmentRequest( api_version=api_version, body=environments.CreateEnvironmentRequest( from_environment=from_environment, network=utils.get_pydantic_model( network, Optional[environments.CreateEnvironmentRequestNetworkUnion] ), sources=utils.get_pydantic_model( sources, Optional[List[interactions.Source]] ), ), ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="POST", path="/{api_version}/environments", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.CreateEnvironmentGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.body, False, False, "json", environments.CreateEnvironmentRequest, extra_body=extra_body, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) async def _speakeasy_parse_response(http_res): if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "default", "application/json"): return unmarshal_json_response( environments.Environment, http_res, validate=False ) raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="CreateEnvironment", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "create", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/environments \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "sources": [{\n "type": "inline",\n "target": "main.py",\n "content": "print(\\"Hello, World!\\")"\n }]\n }\'\n', }, { "label": "create", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\nenvironment = client.environments.create(\n sources=[{\n "type": "inline",\n "target": "main.py",\n "content": "print(\'Hello, World!\')",\n }]\n)\nprint(environment.id)\n\n# [cleanup]\nclient.environments.delete(id=environment.id)\n# [/cleanup]\n', }, { "label": "create", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst environment = await ai.environments.create({\n sources: [{\n type: 'inline',\n target: 'main.py',\n content: \"print('Hello, World!')\",\n }],\n});\nif (!environment.id) {\n throw new Error('Environment creation failed: ID is undefined');\n}\nconsole.log(environment.id);\n\n// [cleanup]\nawait ai.environments.delete(environment.id);\n// [/cleanup]\n", }, { "label": "create", "lang": "java", "source": 'import com.google.genai.Client;\nimport com.google.genai.gaos.models.environments.CreateEnvironmentRequest;\nimport com.google.genai.gaos.models.environments.Environment;\nimport com.google.genai.gaos.models.interactions.Source;\nimport com.google.genai.gaos.models.interactions.SourceType;\nimport java.util.List;\n\nClient client = new Client();\nCreateEnvironmentRequest request = CreateEnvironmentRequest.builder()\n .sources(List.of(Source.builder()\n .type(SourceType.INLINE)\n .target("main.py")\n .content("print(\'Hello, World!\')")\n .build()))\n .build();\nEnvironment environment = client.environments.createEnvironment(request).environment().orElseThrow();\nSystem.out.println(environment.id().orElse(""));\n\n// [cleanup]\nenvironment.id().ifPresent(id -> client.environments.deleteEnvironment(id));\n// [/cleanup]\n', }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="async"), ) http_res = await self.do_request_async( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): await http_res.aread() try: await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.AsyncStreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.AsyncAPIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), async_hooks=self.sdk_configuration.__dict__.get("_async_hooks"), ), ) try: return await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) async def delete_environment( self, id: str, *, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Empty: r"""Deletes an environment. :param id: Resource ID segment making up resource `name`. It identifies the resource within its parent collection as described in https://google.aip.dev/122. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.DeleteEnvironmentRequest( api_version=api_version, id=id, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="DELETE", path="/{api_version}/environments/{id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.DeleteEnvironmentGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) async def _speakeasy_parse_response(http_res): if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "default", "application/json"): return unmarshal_json_response( interactions.Empty, http_res, validate=False ) raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="DeleteEnvironment", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "delete", "lang": "sh", "source": 'curl -X DELETE https://generativelanguage.googleapis.com/v1beta/environments/env_abc123 \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "delete", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# [setup]\ncreated = client.environments.create(\n sources=[{"type": "inline", "target": "main.py", "content": "print(\'Hello\')"}]\n)\n# [/setup]\n\nclient.environments.delete(id=created.id)\nprint("Environment deleted successfully.")\n', }, { "label": "delete", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// [setup]\nconst created = await ai.environments.create({\n sources: [{type: 'inline', target: 'main.py', content: \"print('Hello')\"}],\n});\nif (!created.id) throw new Error('Failed to create environment');\n// [/setup]\n\nawait ai.environments.delete(created.id);\nconsole.log('Environment deleted successfully.');\n", }, { "label": "delete", "lang": "java", "source": 'import com.google.genai.Client;\nimport com.google.genai.gaos.models.environments.CreateEnvironmentRequest;\nimport com.google.genai.gaos.models.environments.Environment;\nimport com.google.genai.gaos.models.interactions.Source;\nimport com.google.genai.gaos.models.interactions.SourceType;\nimport java.util.List;\n\nClient client = new Client();\n\n// [setup]\nCreateEnvironmentRequest request = CreateEnvironmentRequest.builder()\n .sources(List.of(Source.builder()\n .type(SourceType.INLINE)\n .target("main.py")\n .content("print(\'Hello\')")\n .build()))\n .build();\nEnvironment created = client.environments.createEnvironment(request).environment().orElseThrow();\nString envId = created.id().orElseThrow();\n// [/setup]\n\nclient.environments.deleteEnvironment(envId);\nSystem.out.println("Environment deleted successfully.");\n', }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="async"), ) http_res = await self.do_request_async( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): await http_res.aread() try: await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.AsyncStreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.AsyncAPIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), async_hooks=self.sdk_configuration.__dict__.get("_async_hooks"), ), ) try: return await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) async def get_environment( self, id: str, *, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> environments.Environment: r"""Gets an environment. :param id: Resource ID segment making up resource `name`. It identifies the resource within its parent collection as described in https://google.aip.dev/122. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetEnvironmentRequest( api_version=api_version, id=id, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="GET", path="/{api_version}/environments/{id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.GetEnvironmentGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) async def _speakeasy_parse_response(http_res): if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "API error occurred", http_res, http_res_text ) if utils.match_response(http_res, "default", "application/json"): return unmarshal_json_response( environments.Environment, http_res, validate=False ) raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="GetEnvironment", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "get", "lang": "sh", "source": 'curl -X GET https://generativelanguage.googleapis.com/v1beta/environments/env_abc123 \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "get", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# [setup]\ncreated = client.environments.create(\n sources=[{"type": "inline", "target": "main.py", "content": "print(\'Hello\')"}]\n)\n# [/setup]\n\nenvironment = client.environments.get(id=created.id)\nprint(environment.id)\n\n# [cleanup]\nclient.environments.delete(id=created.id)\n# [/cleanup]\n', }, { "label": "get", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// [setup]\nconst created = await ai.environments.create({\n sources: [{type: 'inline', target: 'main.py', content: \"print('Hello')\"}],\n});\nif (!created.id) throw new Error('Failed to create environment');\n// [/setup]\n\nconst environment = await ai.environments.get(created.id);\nconsole.log(environment.id);\n\n// [cleanup]\nawait ai.environments.delete(created.id);\n// [/cleanup]\n", }, { "label": "get", "lang": "java", "source": 'import com.google.genai.Client;\nimport com.google.genai.gaos.models.environments.CreateEnvironmentRequest;\nimport com.google.genai.gaos.models.environments.Environment;\nimport com.google.genai.gaos.models.interactions.Source;\nimport com.google.genai.gaos.models.interactions.SourceType;\nimport java.util.List;\n\nClient client = new Client();\n\n// [setup]\nCreateEnvironmentRequest request = CreateEnvironmentRequest.builder()\n .sources(List.of(Source.builder()\n .type(SourceType.INLINE)\n .target("main.py")\n .content("print(\'Hello\')")\n .build()))\n .build();\nEnvironment created = client.environments.createEnvironment(request).environment().orElseThrow();\nString envId = created.id().orElseThrow();\n// [/setup]\n\nEnvironment environment = client.environments.getEnvironment(envId).environment().orElseThrow();\nSystem.out.println(environment.id().orElse(""));\n\n// [cleanup]\nclient.environments.deleteEnvironment(envId);\n// [/cleanup]\n', }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="async"), ) http_res = await self.do_request_async( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): await http_res.aread() try: await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.AsyncStreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.AsyncAPIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), async_hooks=self.sdk_configuration.__dict__.get("_async_hooks"), ), ) try: return await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) class AsyncEnvironmentsWithRawResponse: def __init__(self, sdk: AsyncEnvironments) -> None: self._sdk = sdk self.list_environments = response_helpers.async_to_raw_response_wrapper( sdk.list_environments, "extra_headers" ) self.create_environment = response_helpers.async_to_raw_response_wrapper( sdk.create_environment, "extra_headers" ) self.delete_environment = response_helpers.async_to_raw_response_wrapper( sdk.delete_environment, "extra_headers" ) self.get_environment = response_helpers.async_to_raw_response_wrapper( sdk.get_environment, "extra_headers" ) @property def files(self): return self._sdk.files.with_raw_response class AsyncEnvironmentsWithStreamingResponse: def __init__(self, sdk: AsyncEnvironments) -> None: self._sdk = sdk self.list_environments = response_helpers.async_to_streamed_response_wrapper( sdk.list_environments, "extra_headers" ) self.create_environment = response_helpers.async_to_streamed_response_wrapper( sdk.create_environment, "extra_headers" ) self.delete_environment = response_helpers.async_to_streamed_response_wrapper( sdk.delete_environment, "extra_headers" ) self.get_environment = response_helpers.async_to_streamed_response_wrapper( sdk.get_environment, "extra_headers" ) @property def files(self): return self._sdk.files.with_streaming_response