# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ...core.api_error import ApiError from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ...core.http_response import AsyncHttpResponse, HttpResponse from ...core.jsonable_encoder import encode_path_param from ...core.parse_error import ParsingError from ...core.request_options import RequestOptions from ...core.unchecked_base_model import construct_type from ...errors.bad_request_error import BadRequestError from ...types.agent_variable_v1 import AgentVariableV1 from ...types.delete_agent_variable_v1response import DeleteAgentVariableV1Response from ...types.list_agent_variables_v1response import ListAgentVariablesV1Response from pydantic import ValidationError # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawVariablesClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, project_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[ListAgentVariablesV1Response]: """ Returns all template variables for the specified project Parameters ---------- project_id : str The unique identifier of the project request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ListAgentVariablesV1Response] A list of agent variables """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListAgentVariablesV1Response, construct_type( type_=ListAgentVariablesV1Response, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def create( self, project_id: str, *, key: str, value: typing.Any, api_version: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentVariableV1]: """ Creates a new template variable. Variables follow the `DG_` naming format and can substitute any JSON value in an agent configuration. Parameters ---------- project_id : str The unique identifier of the project key : str The variable name, following the DG_ format value : typing.Any api_version : typing.Optional[int] API version. Defaults to 1 request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentVariableV1] Agent variable created successfully """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables", base_url=self._client_wrapper.get_environment().base, method="POST", json={ "key": key, "value": value, "api_version": api_version, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentVariableV1, construct_type( type_=AgentVariableV1, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get( self, project_id: str, variable_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[AgentVariableV1]: """ Returns the specified template variable Parameters ---------- project_id : str The unique identifier of the project variable_id : str The unique identifier of the agent variable request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentVariableV1] An agent variable """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables/{encode_path_param(variable_id)}", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentVariableV1, construct_type( type_=AgentVariableV1, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete( self, project_id: str, variable_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DeleteAgentVariableV1Response]: """ Deletes the specified template variable Parameters ---------- project_id : str The unique identifier of the project variable_id : str The unique identifier of the agent variable request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DeleteAgentVariableV1Response] Agent variable deleted """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables/{encode_path_param(variable_id)}", base_url=self._client_wrapper.get_environment().base, method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteAgentVariableV1Response, construct_type( type_=DeleteAgentVariableV1Response, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def update( self, project_id: str, variable_id: str, *, value: typing.Any, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentVariableV1]: """ Updates the value of an existing template variable Parameters ---------- project_id : str The unique identifier of the project variable_id : str The unique identifier of the agent variable value : typing.Any request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentVariableV1] Agent variable updated """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables/{encode_path_param(variable_id)}", base_url=self._client_wrapper.get_environment().base, method="PATCH", json={ "value": value, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentVariableV1, construct_type( type_=AgentVariableV1, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawVariablesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, project_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[ListAgentVariablesV1Response]: """ Returns all template variables for the specified project Parameters ---------- project_id : str The unique identifier of the project request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ListAgentVariablesV1Response] A list of agent variables """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListAgentVariablesV1Response, construct_type( type_=ListAgentVariablesV1Response, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def create( self, project_id: str, *, key: str, value: typing.Any, api_version: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentVariableV1]: """ Creates a new template variable. Variables follow the `DG_` naming format and can substitute any JSON value in an agent configuration. Parameters ---------- project_id : str The unique identifier of the project key : str The variable name, following the DG_ format value : typing.Any api_version : typing.Optional[int] API version. Defaults to 1 request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentVariableV1] Agent variable created successfully """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables", base_url=self._client_wrapper.get_environment().base, method="POST", json={ "key": key, "value": value, "api_version": api_version, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentVariableV1, construct_type( type_=AgentVariableV1, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get( self, project_id: str, variable_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[AgentVariableV1]: """ Returns the specified template variable Parameters ---------- project_id : str The unique identifier of the project variable_id : str The unique identifier of the agent variable request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentVariableV1] An agent variable """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables/{encode_path_param(variable_id)}", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentVariableV1, construct_type( type_=AgentVariableV1, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete( self, project_id: str, variable_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DeleteAgentVariableV1Response]: """ Deletes the specified template variable Parameters ---------- project_id : str The unique identifier of the project variable_id : str The unique identifier of the agent variable request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DeleteAgentVariableV1Response] Agent variable deleted """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables/{encode_path_param(variable_id)}", base_url=self._client_wrapper.get_environment().base, method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteAgentVariableV1Response, construct_type( type_=DeleteAgentVariableV1Response, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def update( self, project_id: str, variable_id: str, *, value: typing.Any, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentVariableV1]: """ Updates the value of an existing template variable Parameters ---------- project_id : str The unique identifier of the project variable_id : str The unique identifier of the agent variable value : typing.Any request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentVariableV1] Agent variable updated """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agent-variables/{encode_path_param(variable_id)}", base_url=self._client_wrapper.get_environment().base, method="PATCH", json={ "value": value, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentVariableV1, construct_type( type_=AgentVariableV1, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)