# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ...commons.errors.access_denied_error import ( AccessDeniedError as commons_errors_access_denied_error_AccessDeniedError, ) from ...commons.errors.error import Error from ...commons.errors.method_not_allowed_error import ( MethodNotAllowedError as commons_errors_method_not_allowed_error_MethodNotAllowedError, ) from ...commons.errors.not_found_error import ( NotFoundError as commons_errors_not_found_error_NotFoundError, ) from ...commons.errors.unauthorized_error import ( UnauthorizedError as commons_errors_unauthorized_error_UnauthorizedError, ) 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 jsonable_encoder from ...core.pydantic_utilities import parse_obj_as from ...core.request_options import RequestOptions from ...core.serialization import convert_and_respect_annotation_metadata from ..errors.errors.access_denied_error import ( AccessDeniedError as unstable_errors_errors_access_denied_error_AccessDeniedError, ) from ..errors.errors.bad_request_error import BadRequestError from ..errors.errors.conflict_error import ConflictError from ..errors.errors.internal_server_error import InternalServerError from ..errors.errors.method_not_allowed_error import ( MethodNotAllowedError as unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError, ) from ..errors.errors.not_found_error import ( NotFoundError as unstable_errors_errors_not_found_error_NotFoundError, ) from ..errors.errors.too_many_requests_error import TooManyRequestsError from ..errors.errors.unauthorized_error import ( UnauthorizedError as unstable_errors_errors_unauthorized_error_UnauthorizedError, ) from ..errors.types.public_api_error import PublicApiError from .types.dashboard_widget import DashboardWidget from .types.dashboard_widget_chart_config_input import DashboardWidgetChartConfigInput from .types.dashboard_widget_chart_type import DashboardWidgetChartType from .types.dashboard_widget_dimension import DashboardWidgetDimension from .types.dashboard_widget_filter import DashboardWidgetFilter from .types.dashboard_widget_list import DashboardWidgetList from .types.dashboard_widget_metric import DashboardWidgetMetric from .types.dashboard_widget_view import DashboardWidgetView from .types.delete_dashboard_widget_response import DeleteDashboardWidgetResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawDashboardWidgetsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, *, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[DashboardWidgetList]: """ List dashboard widgets in the project, ordered by most recently updated first. Responses may include legacy `traces` widgets created before this API existed. New widgets cannot be created with `view: traces`. Parameters ---------- page : typing.Optional[int] 1-based page number. Defaults to `1`. limit : typing.Optional[int] Maximum number of items per page. Defaults to `50`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DashboardWidgetList] """ _response = self._client_wrapper.httpx_client.request( "api/public/unstable/dashboard-widgets", method="GET", params={ "page": page, "limit": limit, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DashboardWidgetList, parse_obj_as( type_=DashboardWidgetList, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) def create( self, *, name: str, view: DashboardWidgetView, dimensions: typing.Sequence[DashboardWidgetDimension], metrics: typing.Sequence[DashboardWidgetMetric], filters: typing.Sequence[DashboardWidgetFilter], chart_type: DashboardWidgetChartType, description: typing.Optional[str] = OMIT, chart_config: typing.Optional[DashboardWidgetChartConfigInput] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[DashboardWidget]: """ Create a dashboard widget (a standalone chart definition you place on any dashboard). This endpoint creates the widget only; place it on a dashboard via `POST /dashboards/{dashboardId}/placements`. Supported views are `observations`, `scores-numeric`, `scores-boolean`, and `scores-categorical`. The legacy `traces` view is not supported by this unstable API. Widgets are created as v2 internally. `chartConfig` is optional and defaults to the plain config for `chartType`; when `chartConfig.type` is given it must match `chartType`. Unstable API note: - This surface may evolve while dashboard/widget APIs are being finalized. Parameters ---------- name : str view : DashboardWidgetView dimensions : typing.Sequence[DashboardWidgetDimension] metrics : typing.Sequence[DashboardWidgetMetric] filters : typing.Sequence[DashboardWidgetFilter] chart_type : DashboardWidgetChartType description : typing.Optional[str] Defaults to an empty string. chart_config : typing.Optional[DashboardWidgetChartConfigInput] Defaults to the plain config for `chartType`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DashboardWidget] """ _response = self._client_wrapper.httpx_client.request( "api/public/unstable/dashboard-widgets", method="POST", json={ "name": name, "description": description, "view": view, "dimensions": convert_and_respect_annotation_metadata( object_=dimensions, annotation=typing.Sequence[DashboardWidgetDimension], direction="write", ), "metrics": convert_and_respect_annotation_metadata( object_=metrics, annotation=typing.Sequence[DashboardWidgetMetric], direction="write", ), "filters": convert_and_respect_annotation_metadata( object_=filters, annotation=typing.Sequence[DashboardWidgetFilter], direction="write", ), "chartType": chart_type, "chartConfig": convert_and_respect_annotation_metadata( object_=chart_config, annotation=DashboardWidgetChartConfigInput, direction="write", ), }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DashboardWidget, parse_obj_as( type_=DashboardWidget, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) def get( self, widget_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DashboardWidget]: """ Get a dashboard widget by id. The response may use `view: traces` for legacy widgets. Parameters ---------- widget_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DashboardWidget] """ _response = self._client_wrapper.httpx_client.request( f"api/public/unstable/dashboard-widgets/{jsonable_encoder(widget_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DashboardWidget, parse_obj_as( type_=DashboardWidget, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) def update( self, widget_id: str, *, name: typing.Optional[str] = OMIT, description: typing.Optional[str] = OMIT, view: typing.Optional[DashboardWidgetView] = OMIT, dimensions: typing.Optional[typing.Sequence[DashboardWidgetDimension]] = OMIT, metrics: typing.Optional[typing.Sequence[DashboardWidgetMetric]] = OMIT, filters: typing.Optional[typing.Sequence[DashboardWidgetFilter]] = OMIT, chart_type: typing.Optional[DashboardWidgetChartType] = OMIT, chart_config: typing.Optional[DashboardWidgetChartConfigInput] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[DashboardWidget]: """ Update a dashboard widget. All fields are optional; at least one field is required. Changing `chartType` without sending `chartConfig` resets the config to the new chart type's defaults. When `chartConfig.type` is given it must match the widget's (possibly updated) `chartType`. `view` cannot be changed to the legacy `traces` value. Existing `traces` widgets may be updated on other fields. Parameters ---------- widget_id : str name : typing.Optional[str] description : typing.Optional[str] view : typing.Optional[DashboardWidgetView] dimensions : typing.Optional[typing.Sequence[DashboardWidgetDimension]] metrics : typing.Optional[typing.Sequence[DashboardWidgetMetric]] filters : typing.Optional[typing.Sequence[DashboardWidgetFilter]] chart_type : typing.Optional[DashboardWidgetChartType] chart_config : typing.Optional[DashboardWidgetChartConfigInput] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DashboardWidget] """ _response = self._client_wrapper.httpx_client.request( f"api/public/unstable/dashboard-widgets/{jsonable_encoder(widget_id)}", method="PATCH", json={ "name": name, "description": description, "view": view, "dimensions": convert_and_respect_annotation_metadata( object_=dimensions, annotation=typing.Sequence[DashboardWidgetDimension], direction="write", ), "metrics": convert_and_respect_annotation_metadata( object_=metrics, annotation=typing.Sequence[DashboardWidgetMetric], direction="write", ), "filters": convert_and_respect_annotation_metadata( object_=filters, annotation=typing.Sequence[DashboardWidgetFilter], direction="write", ), "chartType": chart_type, "chartConfig": convert_and_respect_annotation_metadata( object_=chart_config, annotation=DashboardWidgetChartConfigInput, direction="write", ), }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DashboardWidget, parse_obj_as( type_=DashboardWidget, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) def delete( self, widget_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DeleteDashboardWidgetResponse]: """ Delete a dashboard widget. The API returns `409` while the widget is still placed on a dashboard. Remove those placements first. Parameters ---------- widget_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DeleteDashboardWidgetResponse] """ _response = self._client_wrapper.httpx_client.request( f"api/public/unstable/dashboard-widgets/{jsonable_encoder(widget_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteDashboardWidgetResponse, parse_obj_as( type_=DeleteDashboardWidgetResponse, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 409: raise ConflictError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) class AsyncRawDashboardWidgetsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[DashboardWidgetList]: """ List dashboard widgets in the project, ordered by most recently updated first. Responses may include legacy `traces` widgets created before this API existed. New widgets cannot be created with `view: traces`. Parameters ---------- page : typing.Optional[int] 1-based page number. Defaults to `1`. limit : typing.Optional[int] Maximum number of items per page. Defaults to `50`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DashboardWidgetList] """ _response = await self._client_wrapper.httpx_client.request( "api/public/unstable/dashboard-widgets", method="GET", params={ "page": page, "limit": limit, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DashboardWidgetList, parse_obj_as( type_=DashboardWidgetList, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) async def create( self, *, name: str, view: DashboardWidgetView, dimensions: typing.Sequence[DashboardWidgetDimension], metrics: typing.Sequence[DashboardWidgetMetric], filters: typing.Sequence[DashboardWidgetFilter], chart_type: DashboardWidgetChartType, description: typing.Optional[str] = OMIT, chart_config: typing.Optional[DashboardWidgetChartConfigInput] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[DashboardWidget]: """ Create a dashboard widget (a standalone chart definition you place on any dashboard). This endpoint creates the widget only; place it on a dashboard via `POST /dashboards/{dashboardId}/placements`. Supported views are `observations`, `scores-numeric`, `scores-boolean`, and `scores-categorical`. The legacy `traces` view is not supported by this unstable API. Widgets are created as v2 internally. `chartConfig` is optional and defaults to the plain config for `chartType`; when `chartConfig.type` is given it must match `chartType`. Unstable API note: - This surface may evolve while dashboard/widget APIs are being finalized. Parameters ---------- name : str view : DashboardWidgetView dimensions : typing.Sequence[DashboardWidgetDimension] metrics : typing.Sequence[DashboardWidgetMetric] filters : typing.Sequence[DashboardWidgetFilter] chart_type : DashboardWidgetChartType description : typing.Optional[str] Defaults to an empty string. chart_config : typing.Optional[DashboardWidgetChartConfigInput] Defaults to the plain config for `chartType`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DashboardWidget] """ _response = await self._client_wrapper.httpx_client.request( "api/public/unstable/dashboard-widgets", method="POST", json={ "name": name, "description": description, "view": view, "dimensions": convert_and_respect_annotation_metadata( object_=dimensions, annotation=typing.Sequence[DashboardWidgetDimension], direction="write", ), "metrics": convert_and_respect_annotation_metadata( object_=metrics, annotation=typing.Sequence[DashboardWidgetMetric], direction="write", ), "filters": convert_and_respect_annotation_metadata( object_=filters, annotation=typing.Sequence[DashboardWidgetFilter], direction="write", ), "chartType": chart_type, "chartConfig": convert_and_respect_annotation_metadata( object_=chart_config, annotation=DashboardWidgetChartConfigInput, direction="write", ), }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DashboardWidget, parse_obj_as( type_=DashboardWidget, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) async def get( self, widget_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DashboardWidget]: """ Get a dashboard widget by id. The response may use `view: traces` for legacy widgets. Parameters ---------- widget_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DashboardWidget] """ _response = await self._client_wrapper.httpx_client.request( f"api/public/unstable/dashboard-widgets/{jsonable_encoder(widget_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DashboardWidget, parse_obj_as( type_=DashboardWidget, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) async def update( self, widget_id: str, *, name: typing.Optional[str] = OMIT, description: typing.Optional[str] = OMIT, view: typing.Optional[DashboardWidgetView] = OMIT, dimensions: typing.Optional[typing.Sequence[DashboardWidgetDimension]] = OMIT, metrics: typing.Optional[typing.Sequence[DashboardWidgetMetric]] = OMIT, filters: typing.Optional[typing.Sequence[DashboardWidgetFilter]] = OMIT, chart_type: typing.Optional[DashboardWidgetChartType] = OMIT, chart_config: typing.Optional[DashboardWidgetChartConfigInput] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[DashboardWidget]: """ Update a dashboard widget. All fields are optional; at least one field is required. Changing `chartType` without sending `chartConfig` resets the config to the new chart type's defaults. When `chartConfig.type` is given it must match the widget's (possibly updated) `chartType`. `view` cannot be changed to the legacy `traces` value. Existing `traces` widgets may be updated on other fields. Parameters ---------- widget_id : str name : typing.Optional[str] description : typing.Optional[str] view : typing.Optional[DashboardWidgetView] dimensions : typing.Optional[typing.Sequence[DashboardWidgetDimension]] metrics : typing.Optional[typing.Sequence[DashboardWidgetMetric]] filters : typing.Optional[typing.Sequence[DashboardWidgetFilter]] chart_type : typing.Optional[DashboardWidgetChartType] chart_config : typing.Optional[DashboardWidgetChartConfigInput] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DashboardWidget] """ _response = await self._client_wrapper.httpx_client.request( f"api/public/unstable/dashboard-widgets/{jsonable_encoder(widget_id)}", method="PATCH", json={ "name": name, "description": description, "view": view, "dimensions": convert_and_respect_annotation_metadata( object_=dimensions, annotation=typing.Sequence[DashboardWidgetDimension], direction="write", ), "metrics": convert_and_respect_annotation_metadata( object_=metrics, annotation=typing.Sequence[DashboardWidgetMetric], direction="write", ), "filters": convert_and_respect_annotation_metadata( object_=filters, annotation=typing.Sequence[DashboardWidgetFilter], direction="write", ), "chartType": chart_type, "chartConfig": convert_and_respect_annotation_metadata( object_=chart_config, annotation=DashboardWidgetChartConfigInput, direction="write", ), }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DashboardWidget, parse_obj_as( type_=DashboardWidget, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) async def delete( self, widget_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DeleteDashboardWidgetResponse]: """ Delete a dashboard widget. The API returns `409` while the widget is still placed on a dashboard. Remove those placements first. Parameters ---------- widget_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DeleteDashboardWidgetResponse] """ _response = await self._client_wrapper.httpx_client.request( f"api/public/unstable/dashboard-widgets/{jsonable_encoder(widget_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteDashboardWidgetResponse, parse_obj_as( type_=DeleteDashboardWidgetResponse, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 409: raise ConflictError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( 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, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, )