# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from .. import core 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.unchecked_base_model import construct_type from ..types.file_upload import FileUpload # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawFilesClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def get(self, id: int, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[FileUpload]: """ Retrieve details about a specific uploaded file. Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[FileUpload] """ _response = self._client_wrapper.httpx_client.request( f"api/import/file-upload/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( FileUpload, construct_type( type_=FileUpload, # 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: int, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]: """ Delete a specific uploaded file. Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"api/import/file-upload/{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: int, *, file: typing.Optional[core.File] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[FileUpload]: """ Update a specific uploaded file. Parameters ---------- id : int file : typing.Optional[core.File] See core.File for more documentation request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[FileUpload] """ _response = self._client_wrapper.httpx_client.request( f"api/import/file-upload/{jsonable_encoder(id)}", method="PATCH", data={}, files={ **({"file": file} if file is not None else {}), }, request_options=request_options, omit=OMIT, force_multipart=True, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( FileUpload, construct_type( type_=FileUpload, # 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 list( self, id: int, *, all_: typing.Optional[bool] = None, ids: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[typing.List[FileUpload]]: """ Retrieve the list of uploaded files used to create labeling tasks for a specific project. Parameters ---------- id : int all_ : typing.Optional[bool] Set to "true" if you want to retrieve all file uploads ids : typing.Optional[typing.Union[str, typing.Sequence[str]]] Specify the list of file upload IDs to retrieve, e.g. ids=[1,2,3] ordering : typing.Optional[str] Which field to use when ordering the results. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.List[FileUpload]] """ _response = self._client_wrapper.httpx_client.request( f"api/projects/{jsonable_encoder(id)}/file-uploads", method="GET", params={ "all": all_, "ids": ids, "ordering": ordering, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[FileUpload], construct_type( type_=typing.List[FileUpload], # 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_many(self, id: int, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]: """ Delete uploaded files for a specific project. Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"api/projects/{jsonable_encoder(id)}/file-uploads", 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 download(self, filename: str, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]: """ Download a specific uploaded file. Parameters ---------- filename : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"data/upload/{jsonable_encoder(filename)}", method="GET", 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) class AsyncRawFilesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def get( self, id: int, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[FileUpload]: """ Retrieve details about a specific uploaded file. Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[FileUpload] """ _response = await self._client_wrapper.httpx_client.request( f"api/import/file-upload/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( FileUpload, construct_type( type_=FileUpload, # 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: int, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete a specific uploaded file. Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"api/import/file-upload/{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: int, *, file: typing.Optional[core.File] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[FileUpload]: """ Update a specific uploaded file. Parameters ---------- id : int file : typing.Optional[core.File] See core.File for more documentation request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[FileUpload] """ _response = await self._client_wrapper.httpx_client.request( f"api/import/file-upload/{jsonable_encoder(id)}", method="PATCH", data={}, files={ **({"file": file} if file is not None else {}), }, request_options=request_options, omit=OMIT, force_multipart=True, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( FileUpload, construct_type( type_=FileUpload, # 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 list( self, id: int, *, all_: typing.Optional[bool] = None, ids: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[typing.List[FileUpload]]: """ Retrieve the list of uploaded files used to create labeling tasks for a specific project. Parameters ---------- id : int all_ : typing.Optional[bool] Set to "true" if you want to retrieve all file uploads ids : typing.Optional[typing.Union[str, typing.Sequence[str]]] Specify the list of file upload IDs to retrieve, e.g. ids=[1,2,3] ordering : typing.Optional[str] Which field to use when ordering the results. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.List[FileUpload]] """ _response = await self._client_wrapper.httpx_client.request( f"api/projects/{jsonable_encoder(id)}/file-uploads", method="GET", params={ "all": all_, "ids": ids, "ordering": ordering, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[FileUpload], construct_type( type_=typing.List[FileUpload], # 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_many( self, id: int, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete uploaded files for a specific project. Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"api/projects/{jsonable_encoder(id)}/file-uploads", 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 download( self, filename: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Download a specific uploaded file. Parameters ---------- filename : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"data/upload/{jsonable_encoder(filename)}", method="GET", 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)