# This file was auto-generated by Fern from our API Definition. import contextlib 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 ..core.serialization import convert_and_respect_annotation_metadata from ..errors.forbidden_error import ForbiddenError from ..errors.unauthorized_error import UnauthorizedError from ..types.attachment_page import AttachmentPage from ..types.complete_multipart_upload_request_entity_type import CompleteMultipartUploadRequestEntityType from ..types.error_message import ErrorMessage from ..types.multipart_upload_part import MultipartUploadPart from ..types.start_multipart_upload_response import StartMultipartUploadResponse from .types.attachment_list_request_entity_type import AttachmentListRequestEntityType from .types.download_attachment_request_entity_type import DownloadAttachmentRequestEntityType from .types.start_multipart_upload_request_entity_type import StartMultipartUploadRequestEntityType from .types.upload_attachment_request_entity_type import UploadAttachmentRequestEntityType # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawAttachmentsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def attachment_list( self, *, project_id: str, entity_type: AttachmentListRequestEntityType, entity_id: str, path: str, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AttachmentPage]: """ Attachments list for entity Parameters ---------- project_id : str entity_type : AttachmentListRequestEntityType entity_id : str path : str page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AttachmentPage] Attachment Resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/attachment/list", method="GET", params={ "page": page, "size": size, "project_id": project_id, "entity_type": entity_type, "entity_id": entity_id, "path": path, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AttachmentPage, parse_obj_as( type_=AttachmentPage, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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 complete_multi_part_upload( self, *, file_name: str, entity_type: CompleteMultipartUploadRequestEntityType, entity_id: str, file_size: int, upload_id: str, uploaded_file_parts: typing.Sequence[MultipartUploadPart], project_name: typing.Optional[str] = OMIT, container_id: typing.Optional[str] = OMIT, mime_type: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Complete multipart attachment upload Parameters ---------- file_name : str entity_type : CompleteMultipartUploadRequestEntityType entity_id : str file_size : int upload_id : str uploaded_file_parts : typing.Sequence[MultipartUploadPart] project_name : typing.Optional[str] If null, the default project is used container_id : typing.Optional[str] mime_type : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/attachment/upload-complete", method="POST", json={ "file_name": file_name, "project_name": project_name, "entity_type": entity_type, "entity_id": entity_id, "container_id": container_id, "file_size": file_size, "mime_type": mime_type, "upload_id": upload_id, "uploaded_file_parts": convert_and_respect_annotation_metadata( object_=uploaded_file_parts, annotation=typing.Sequence[MultipartUploadPart], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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_attachments( self, *, file_name: str, entity_type: CompleteMultipartUploadRequestEntityType, entity_id: str, file_size: int, upload_id: str, uploaded_file_parts: typing.Sequence[MultipartUploadPart], project_name: typing.Optional[str] = OMIT, container_id: typing.Optional[str] = OMIT, mime_type: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Delete attachments Parameters ---------- file_name : str entity_type : CompleteMultipartUploadRequestEntityType entity_id : str file_size : int upload_id : str uploaded_file_parts : typing.Sequence[MultipartUploadPart] project_name : typing.Optional[str] If null, the default project is used container_id : typing.Optional[str] mime_type : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/attachment/delete", method="POST", json={ "file_name": file_name, "project_name": project_name, "entity_type": entity_type, "entity_id": entity_id, "container_id": container_id, "file_size": file_size, "mime_type": mime_type, "upload_id": upload_id, "uploaded_file_parts": convert_and_respect_annotation_metadata( object_=uploaded_file_parts, annotation=typing.Sequence[MultipartUploadPart], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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) @contextlib.contextmanager def download_attachment( self, *, container_id: str, entity_type: DownloadAttachmentRequestEntityType, entity_id: str, file_name: str, mime_type: str, workspace_name: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[HttpResponse[typing.Iterator[bytes]]]: """ Download attachment from MinIO Parameters ---------- container_id : str entity_type : DownloadAttachmentRequestEntityType entity_id : str file_name : str mime_type : str workspace_name : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.Iterator[HttpResponse[typing.Iterator[bytes]]] Attachment Resource """ with self._client_wrapper.httpx_client.stream( "v1/private/attachment/download", method="GET", params={ "workspace_name": workspace_name, "container_id": container_id, "entity_type": entity_type, "entity_id": entity_id, "file_name": file_name, "mime_type": mime_type, }, request_options=request_options, ) as _response: def stream() -> HttpResponse[typing.Iterator[bytes]]: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", None) if request_options is not None else None return HttpResponse( response=_response, data=(_chunk for _chunk in _response.iter_bytes(chunk_size=_chunk_size)) ) _response.read() if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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) yield stream() def start_multi_part_upload( self, *, file_name: str, num_of_file_parts: int, entity_type: StartMultipartUploadRequestEntityType, entity_id: str, path: str, mime_type: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[StartMultipartUploadResponse]: """ Start multipart attachment upload Parameters ---------- file_name : str num_of_file_parts : int entity_type : StartMultipartUploadRequestEntityType entity_id : str path : str mime_type : typing.Optional[str] project_name : typing.Optional[str] If null, the default project is used request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[StartMultipartUploadResponse] MultipartUploadResponse """ _response = self._client_wrapper.httpx_client.request( "v1/private/attachment/upload-start", method="POST", json={ "file_name": file_name, "num_of_file_parts": num_of_file_parts, "mime_type": mime_type, "project_name": project_name, "entity_type": entity_type, "entity_id": entity_id, "path": path, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( StartMultipartUploadResponse, parse_obj_as( type_=StartMultipartUploadResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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 upload_attachment( self, *, file_name: str, entity_type: UploadAttachmentRequestEntityType, entity_id: str, request: typing.Dict[str, typing.Optional[typing.Any]], project_name: typing.Optional[str] = None, mime_type: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Upload attachment to MinIO Parameters ---------- file_name : str entity_type : UploadAttachmentRequestEntityType entity_id : str request : typing.Dict[str, typing.Optional[typing.Any]] project_name : typing.Optional[str] mime_type : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/attachment/upload", method="PUT", params={ "file_name": file_name, "project_name": project_name, "mime_type": mime_type, "entity_type": entity_type, "entity_id": entity_id, }, json=request, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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 AsyncRawAttachmentsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def attachment_list( self, *, project_id: str, entity_type: AttachmentListRequestEntityType, entity_id: str, path: str, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AttachmentPage]: """ Attachments list for entity Parameters ---------- project_id : str entity_type : AttachmentListRequestEntityType entity_id : str path : str page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AttachmentPage] Attachment Resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/attachment/list", method="GET", params={ "page": page, "size": size, "project_id": project_id, "entity_type": entity_type, "entity_id": entity_id, "path": path, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AttachmentPage, parse_obj_as( type_=AttachmentPage, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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 complete_multi_part_upload( self, *, file_name: str, entity_type: CompleteMultipartUploadRequestEntityType, entity_id: str, file_size: int, upload_id: str, uploaded_file_parts: typing.Sequence[MultipartUploadPart], project_name: typing.Optional[str] = OMIT, container_id: typing.Optional[str] = OMIT, mime_type: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Complete multipart attachment upload Parameters ---------- file_name : str entity_type : CompleteMultipartUploadRequestEntityType entity_id : str file_size : int upload_id : str uploaded_file_parts : typing.Sequence[MultipartUploadPart] project_name : typing.Optional[str] If null, the default project is used container_id : typing.Optional[str] mime_type : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/attachment/upload-complete", method="POST", json={ "file_name": file_name, "project_name": project_name, "entity_type": entity_type, "entity_id": entity_id, "container_id": container_id, "file_size": file_size, "mime_type": mime_type, "upload_id": upload_id, "uploaded_file_parts": convert_and_respect_annotation_metadata( object_=uploaded_file_parts, annotation=typing.Sequence[MultipartUploadPart], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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_attachments( self, *, file_name: str, entity_type: CompleteMultipartUploadRequestEntityType, entity_id: str, file_size: int, upload_id: str, uploaded_file_parts: typing.Sequence[MultipartUploadPart], project_name: typing.Optional[str] = OMIT, container_id: typing.Optional[str] = OMIT, mime_type: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Delete attachments Parameters ---------- file_name : str entity_type : CompleteMultipartUploadRequestEntityType entity_id : str file_size : int upload_id : str uploaded_file_parts : typing.Sequence[MultipartUploadPart] project_name : typing.Optional[str] If null, the default project is used container_id : typing.Optional[str] mime_type : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/attachment/delete", method="POST", json={ "file_name": file_name, "project_name": project_name, "entity_type": entity_type, "entity_id": entity_id, "container_id": container_id, "file_size": file_size, "mime_type": mime_type, "upload_id": upload_id, "uploaded_file_parts": convert_and_respect_annotation_metadata( object_=uploaded_file_parts, annotation=typing.Sequence[MultipartUploadPart], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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) @contextlib.asynccontextmanager async def download_attachment( self, *, container_id: str, entity_type: DownloadAttachmentRequestEntityType, entity_id: str, file_name: str, mime_type: str, workspace_name: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]]: """ Download attachment from MinIO Parameters ---------- container_id : str entity_type : DownloadAttachmentRequestEntityType entity_id : str file_name : str mime_type : str workspace_name : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]] Attachment Resource """ async with self._client_wrapper.httpx_client.stream( "v1/private/attachment/download", method="GET", params={ "workspace_name": workspace_name, "container_id": container_id, "entity_type": entity_type, "entity_id": entity_id, "file_name": file_name, "mime_type": mime_type, }, request_options=request_options, ) as _response: async def stream() -> AsyncHttpResponse[typing.AsyncIterator[bytes]]: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", None) if request_options is not None else None return AsyncHttpResponse( response=_response, data=(_chunk async for _chunk in _response.aiter_bytes(chunk_size=_chunk_size)), ) await _response.aread() if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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) yield await stream() async def start_multi_part_upload( self, *, file_name: str, num_of_file_parts: int, entity_type: StartMultipartUploadRequestEntityType, entity_id: str, path: str, mime_type: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[StartMultipartUploadResponse]: """ Start multipart attachment upload Parameters ---------- file_name : str num_of_file_parts : int entity_type : StartMultipartUploadRequestEntityType entity_id : str path : str mime_type : typing.Optional[str] project_name : typing.Optional[str] If null, the default project is used request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[StartMultipartUploadResponse] MultipartUploadResponse """ _response = await self._client_wrapper.httpx_client.request( "v1/private/attachment/upload-start", method="POST", json={ "file_name": file_name, "num_of_file_parts": num_of_file_parts, "mime_type": mime_type, "project_name": project_name, "entity_type": entity_type, "entity_id": entity_id, "path": path, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( StartMultipartUploadResponse, parse_obj_as( type_=StartMultipartUploadResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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 upload_attachment( self, *, file_name: str, entity_type: UploadAttachmentRequestEntityType, entity_id: str, request: typing.Dict[str, typing.Optional[typing.Any]], project_name: typing.Optional[str] = None, mime_type: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Upload attachment to MinIO Parameters ---------- file_name : str entity_type : UploadAttachmentRequestEntityType entity_id : str request : typing.Dict[str, typing.Optional[typing.Any]] project_name : typing.Optional[str] mime_type : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/attachment/upload", method="PUT", params={ "file_name": file_name, "project_name": project_name, "mime_type": mime_type, "entity_type": entity_type, "entity_id": entity_id, }, json=request, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # 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)