# 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 jsonable_encoder from ..core.request_options import RequestOptions from ..core.serialization import convert_and_respect_annotation_metadata from ..core.unchecked_base_model import construct_type from ..types.view import View from .types.create_views_request_data import CreateViewsRequestData from .types.update_views_request_data import UpdateViewsRequestData # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawViewsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, *, project: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[typing.List[View]]: """ List all views for a specific project. Parameters ---------- project : typing.Optional[int] Project ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.List[View]] """ _response = self._client_wrapper.httpx_client.request( "api/dm/views/", method="GET", params={ "project": project, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[View], construct_type( type_=typing.List[View], # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _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, *, data: typing.Optional[CreateViewsRequestData] = OMIT, project: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[View]: """ Create a view for a specific project. Parameters ---------- data : typing.Optional[CreateViewsRequestData] Custom view data project : typing.Optional[int] Project ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[View] """ _response = self._client_wrapper.httpx_client.request( "api/dm/views/", method="POST", json={ "data": convert_and_respect_annotation_metadata( object_=data, annotation=CreateViewsRequestData, direction="write" ), "project": project, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( View, construct_type( type_=View, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _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_order( self, *, ids: typing.Sequence[int], project: int, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Update the order field of views based on the provided list of view IDs Parameters ---------- ids : typing.Sequence[int] A list of view IDs in the desired order. project : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "api/dm/views/order/", method="POST", json={ "ids": ids, "project": project, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _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_all( self, *, project: int, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete all views for a specific project. Parameters ---------- project : int Project ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "api/dm/views/reset/", method="DELETE", params={ "project": project, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _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, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[View]: """ Get the details about a specific view in the data manager Parameters ---------- id : str View ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[View] """ _response = self._client_wrapper.httpx_client.request( f"api/dm/views/{jsonable_encoder(id)}/", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( View, construct_type( type_=View, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _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, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]: """ Delete a specific view by ID. Parameters ---------- id : str View ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"api/dm/views/{jsonable_encoder(id)}/", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _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, id: str, *, data: typing.Optional[UpdateViewsRequestData] = OMIT, project: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[View]: """ Update view data with additional filters and other information for a specific project. Parameters ---------- id : str View ID data : typing.Optional[UpdateViewsRequestData] Custom view data project : typing.Optional[int] Project ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[View] """ _response = self._client_wrapper.httpx_client.request( f"api/dm/views/{jsonable_encoder(id)}/", method="PATCH", json={ "data": convert_and_respect_annotation_metadata( object_=data, annotation=UpdateViewsRequestData, direction="write" ), "project": project, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( View, construct_type( type_=View, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _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 AsyncRawViewsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, project: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[typing.List[View]]: """ List all views for a specific project. Parameters ---------- project : typing.Optional[int] Project ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.List[View]] """ _response = await self._client_wrapper.httpx_client.request( "api/dm/views/", method="GET", params={ "project": project, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[View], construct_type( type_=typing.List[View], # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _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, *, data: typing.Optional[CreateViewsRequestData] = OMIT, project: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[View]: """ Create a view for a specific project. Parameters ---------- data : typing.Optional[CreateViewsRequestData] Custom view data project : typing.Optional[int] Project ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[View] """ _response = await self._client_wrapper.httpx_client.request( "api/dm/views/", method="POST", json={ "data": convert_and_respect_annotation_metadata( object_=data, annotation=CreateViewsRequestData, direction="write" ), "project": project, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( View, construct_type( type_=View, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _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_order( self, *, ids: typing.Sequence[int], project: int, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Update the order field of views based on the provided list of view IDs Parameters ---------- ids : typing.Sequence[int] A list of view IDs in the desired order. project : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "api/dm/views/order/", method="POST", json={ "ids": ids, "project": project, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _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_all( self, *, project: int, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete all views for a specific project. Parameters ---------- project : int Project ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "api/dm/views/reset/", method="DELETE", params={ "project": project, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _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, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> AsyncHttpResponse[View]: """ Get the details about a specific view in the data manager Parameters ---------- id : str View ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[View] """ _response = await self._client_wrapper.httpx_client.request( f"api/dm/views/{jsonable_encoder(id)}/", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( View, construct_type( type_=View, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _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, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete a specific view by ID. Parameters ---------- id : str View ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"api/dm/views/{jsonable_encoder(id)}/", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _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, id: str, *, data: typing.Optional[UpdateViewsRequestData] = OMIT, project: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[View]: """ Update view data with additional filters and other information for a specific project. Parameters ---------- id : str View ID data : typing.Optional[UpdateViewsRequestData] Custom view data project : typing.Optional[int] Project ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[View] """ _response = await self._client_wrapper.httpx_client.request( f"api/dm/views/{jsonable_encoder(id)}/", method="PATCH", json={ "data": convert_and_respect_annotation_metadata( object_=data, annotation=UpdateViewsRequestData, direction="write" ), "project": project, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( View, construct_type( type_=View, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _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)