# This file was auto-generated by Fern from our API Definition. import datetime as dt 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.pydantic_utilities import parse_obj_as from ..core.request_options import RequestOptions from ..errors.bad_request_error import BadRequestError from ..errors.not_found_error import NotFoundError from ..errors.unprocessable_entity_error import UnprocessableEntityError from ..types.result import Result from ..types.workspace_configuration import WorkspaceConfiguration from ..types.workspace_metric_response import WorkspaceMetricResponse from ..types.workspace_metrics_summary_response import WorkspaceMetricsSummaryResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawWorkspacesClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def costs_summary( self, *, interval_start: dt.datetime, interval_end: dt.datetime, project_ids: typing.Optional[typing.Sequence[str]] = OMIT, start_before_end: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[Result]: """ Get costs summary Parameters ---------- interval_start : dt.datetime interval_end : dt.datetime project_ids : typing.Optional[typing.Sequence[str]] start_before_end : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Result] Workspace Metrics """ _response = self._client_wrapper.httpx_client.request( "v1/private/workspaces/costs/summaries", method="POST", json={ "project_ids": project_ids, "interval_start": interval_start, "interval_end": interval_end, "start_before_end": start_before_end, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Result, parse_obj_as( type_=Result, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_workspace_configuration( self, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[WorkspaceConfiguration]: """ Get workspace configuration Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[WorkspaceConfiguration] Workspace Configuration """ _response = self._client_wrapper.httpx_client.request( "v1/private/workspaces/configurations", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceConfiguration, parse_obj_as( type_=WorkspaceConfiguration, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def upsert_workspace_configuration( self, *, timeout_to_mark_thread_as_inactive: typing.Optional[str] = OMIT, truncation_on_tables: typing.Optional[bool] = OMIT, color_map: typing.Optional[typing.Dict[str, str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[WorkspaceConfiguration]: """ Upsert workspace configuration Parameters ---------- timeout_to_mark_thread_as_inactive : typing.Optional[str] Duration in ISO-8601 format (e.g., PT30M for 30 minutes, PT2H for 2 hours, P1D for 1 day). Minimum precision supported is seconds, please use a duration with seconds precision or higher. Also, the max duration allowed is 7 days. truncation_on_tables : typing.Optional[bool] Enable or disable data truncation in table views. When disabled, the frontend will limit pagination to prevent performance issues. Default: true (truncation enabled). color_map : typing.Optional[typing.Dict[str, str]] Workspace-level color map. Maps label names to hex color values (e.g. #FF0000). Max 10000 entries. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[WorkspaceConfiguration] Configuration Updated """ _response = self._client_wrapper.httpx_client.request( "v1/private/workspaces/configurations", method="PUT", json={ "timeout_to_mark_thread_as_inactive": timeout_to_mark_thread_as_inactive, "truncation_on_tables": truncation_on_tables, "color_map": color_map, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceConfiguration, parse_obj_as( type_=WorkspaceConfiguration, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete_workspace_configuration( self, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete workspace configuration Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/workspaces/configurations", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_cost( self, *, interval_start: dt.datetime, interval_end: dt.datetime, project_ids: typing.Optional[typing.Sequence[str]] = OMIT, start_before_end: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[WorkspaceMetricResponse]: """ Get cost daily data Parameters ---------- interval_start : dt.datetime interval_end : dt.datetime project_ids : typing.Optional[typing.Sequence[str]] start_before_end : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[WorkspaceMetricResponse] Workspace cost data by days """ _response = self._client_wrapper.httpx_client.request( "v1/private/workspaces/costs", method="POST", json={ "project_ids": project_ids, "interval_start": interval_start, "interval_end": interval_end, "start_before_end": start_before_end, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceMetricResponse, parse_obj_as( type_=WorkspaceMetricResponse, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_metric( self, *, interval_start: dt.datetime, interval_end: dt.datetime, project_ids: typing.Optional[typing.Sequence[str]] = OMIT, start_before_end: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[WorkspaceMetricResponse]: """ Get metric daily data Parameters ---------- interval_start : dt.datetime interval_end : dt.datetime project_ids : typing.Optional[typing.Sequence[str]] start_before_end : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[WorkspaceMetricResponse] Workspace metric data by days """ _response = self._client_wrapper.httpx_client.request( "v1/private/workspaces/metrics", method="POST", json={ "project_ids": project_ids, "interval_start": interval_start, "interval_end": interval_end, "start_before_end": start_before_end, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceMetricResponse, parse_obj_as( type_=WorkspaceMetricResponse, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def metrics_summary( self, *, interval_start: dt.datetime, interval_end: dt.datetime, project_ids: typing.Optional[typing.Sequence[str]] = OMIT, start_before_end: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[WorkspaceMetricsSummaryResponse]: """ Get metrics summary Parameters ---------- interval_start : dt.datetime interval_end : dt.datetime project_ids : typing.Optional[typing.Sequence[str]] start_before_end : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[WorkspaceMetricsSummaryResponse] Workspace Metrics """ _response = self._client_wrapper.httpx_client.request( "v1/private/workspaces/metrics/summaries", method="POST", json={ "project_ids": project_ids, "interval_start": interval_start, "interval_end": interval_end, "start_before_end": start_before_end, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceMetricsSummaryResponse, parse_obj_as( type_=WorkspaceMetricsSummaryResponse, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawWorkspacesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def costs_summary( self, *, interval_start: dt.datetime, interval_end: dt.datetime, project_ids: typing.Optional[typing.Sequence[str]] = OMIT, start_before_end: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[Result]: """ Get costs summary Parameters ---------- interval_start : dt.datetime interval_end : dt.datetime project_ids : typing.Optional[typing.Sequence[str]] start_before_end : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Result] Workspace Metrics """ _response = await self._client_wrapper.httpx_client.request( "v1/private/workspaces/costs/summaries", method="POST", json={ "project_ids": project_ids, "interval_start": interval_start, "interval_end": interval_end, "start_before_end": start_before_end, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Result, parse_obj_as( type_=Result, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_workspace_configuration( self, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[WorkspaceConfiguration]: """ Get workspace configuration Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[WorkspaceConfiguration] Workspace Configuration """ _response = await self._client_wrapper.httpx_client.request( "v1/private/workspaces/configurations", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceConfiguration, parse_obj_as( type_=WorkspaceConfiguration, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def upsert_workspace_configuration( self, *, timeout_to_mark_thread_as_inactive: typing.Optional[str] = OMIT, truncation_on_tables: typing.Optional[bool] = OMIT, color_map: typing.Optional[typing.Dict[str, str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[WorkspaceConfiguration]: """ Upsert workspace configuration Parameters ---------- timeout_to_mark_thread_as_inactive : typing.Optional[str] Duration in ISO-8601 format (e.g., PT30M for 30 minutes, PT2H for 2 hours, P1D for 1 day). Minimum precision supported is seconds, please use a duration with seconds precision or higher. Also, the max duration allowed is 7 days. truncation_on_tables : typing.Optional[bool] Enable or disable data truncation in table views. When disabled, the frontend will limit pagination to prevent performance issues. Default: true (truncation enabled). color_map : typing.Optional[typing.Dict[str, str]] Workspace-level color map. Maps label names to hex color values (e.g. #FF0000). Max 10000 entries. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[WorkspaceConfiguration] Configuration Updated """ _response = await self._client_wrapper.httpx_client.request( "v1/private/workspaces/configurations", method="PUT", json={ "timeout_to_mark_thread_as_inactive": timeout_to_mark_thread_as_inactive, "truncation_on_tables": truncation_on_tables, "color_map": color_map, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceConfiguration, parse_obj_as( type_=WorkspaceConfiguration, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete_workspace_configuration( self, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete workspace configuration Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/workspaces/configurations", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_cost( self, *, interval_start: dt.datetime, interval_end: dt.datetime, project_ids: typing.Optional[typing.Sequence[str]] = OMIT, start_before_end: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[WorkspaceMetricResponse]: """ Get cost daily data Parameters ---------- interval_start : dt.datetime interval_end : dt.datetime project_ids : typing.Optional[typing.Sequence[str]] start_before_end : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[WorkspaceMetricResponse] Workspace cost data by days """ _response = await self._client_wrapper.httpx_client.request( "v1/private/workspaces/costs", method="POST", json={ "project_ids": project_ids, "interval_start": interval_start, "interval_end": interval_end, "start_before_end": start_before_end, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceMetricResponse, parse_obj_as( type_=WorkspaceMetricResponse, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_metric( self, *, interval_start: dt.datetime, interval_end: dt.datetime, project_ids: typing.Optional[typing.Sequence[str]] = OMIT, start_before_end: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[WorkspaceMetricResponse]: """ Get metric daily data Parameters ---------- interval_start : dt.datetime interval_end : dt.datetime project_ids : typing.Optional[typing.Sequence[str]] start_before_end : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[WorkspaceMetricResponse] Workspace metric data by days """ _response = await self._client_wrapper.httpx_client.request( "v1/private/workspaces/metrics", method="POST", json={ "project_ids": project_ids, "interval_start": interval_start, "interval_end": interval_end, "start_before_end": start_before_end, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceMetricResponse, parse_obj_as( type_=WorkspaceMetricResponse, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def metrics_summary( self, *, interval_start: dt.datetime, interval_end: dt.datetime, project_ids: typing.Optional[typing.Sequence[str]] = OMIT, start_before_end: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[WorkspaceMetricsSummaryResponse]: """ Get metrics summary Parameters ---------- interval_start : dt.datetime interval_end : dt.datetime project_ids : typing.Optional[typing.Sequence[str]] start_before_end : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[WorkspaceMetricsSummaryResponse] Workspace Metrics """ _response = await self._client_wrapper.httpx_client.request( "v1/private/workspaces/metrics/summaries", method="POST", json={ "project_ids": project_ids, "interval_start": interval_start, "interval_end": interval_end, "start_before_end": start_before_end, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceMetricsSummaryResponse, parse_obj_as( type_=WorkspaceMetricsSummaryResponse, # 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.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)