UploadThing upload wrapper for python3 using requests

uploadthing.py

from requests import Session, Response
from urllib.parse import quote
from typing import Tuple
from functools import lru_cache


class UploadThing:
    def __init__(self, api_key: str, base_url: str = 'https://uploadthing.com/api') -> None:
        self.session = Session()
        self.session.headers = {
            'User-Agent': ':hmmm:',
            'x-uploadthing-api-key': api_key,
            'x-uploadthing-version': '6.9.0',
        }
        self.base_url = base_url.rstrip('/')

    def _post(self, endpoint: str, **kwargs) -> Response:
        response = self.session.post(
            f'{self.base_url}/{endpoint}',
            json=kwargs
        )
        response.raise_for_status()
        return response

    def _content_disposition(self, file_name: str, content_disposition: str = 'attachment') -> str:
        return ';'.join([content_disposition, f'filename="{quote(file_name)}"', f"filename*=UTF-8''{quote(file_name)}"])

    def _upload_part(
            self,
            url: str,
            chunk: bytes,
            file_name: str,
            content_disposition: str = 'attachment'
    ) -> str:
        s3_response = self.session.put(
            url,
            data=chunk,
            headers={
                'Content-Type': 'application/binary',
                'Content-Disposition': self._content_disposition(file_name, content_disposition),
            }
        )
        s3_response.raise_for_status()  # todo: retries

        return s3_response.headers.get('ETag').replace('"', '')

    def upload_file(self, file_name: str, file_data: bytes, custom_id: str | None = None) -> Tuple[str, str]:
        file_json = {
            'name': file_name,
            'size': len(file_data),
            'type': 'bin'
        }
        if custom_id:
            file_json['customId'] = custom_id

        start = self._post(
            'uploadFiles',
            files=[file_json],
            acl='public-read',
            metadata=None,
            contentDisposition='attachment'
        ).json()
        file_info = start['data'][0]

        if 'url' in file_info:
            self.session.post(
                file_info['url'],
                data=file_info['fields'],
                files={
                    'file': file_data
                }
            ).raise_for_status()
        elif 'urls' in file_info:
            chunk_size = file_info['chunkSize']
            etags = []

            for i, url in enumerate(file_info['urls']):
                start = chunk_size * i
                end = min(start + chunk_size, len(file_data))

                etags.append({
                    'tag': self._upload_part(url, file_data[start:end], file_name),
                    'partNumber': i + 1
                })

            self._post(
                'completeMultipart',
                fileKey=file_info['key'],
                uploadId=file_info['uploadId'],
                etags=etags
            ).raise_for_status()
        else:
            raise ValueError('Got unsupported file response')

        return file_info['fileUrl'], file_info['key']

    def delete_by_custom_id(self, custom_id: str) -> None:
        self._post(
            'deleteFile',
            customIds=[custom_id]
        ).raise_for_status()

    def delete_by_key(self, key: str) -> None:
        self._post(
            'deleteFile',
            fileKeys=[key]
        ).raise_for_status()

    def get_url_for_custom_id(self, custom_id: str) -> str:
        @lru_cache()
        def get_for(custom_id_value: str) -> str:
            response = self._post(
                'getFileUrl',
                customIds=[custom_id_value]
            ).json()
            assert len(response['data']) > 0
            return response['data'][0]['url']

        return get_for(custom_id)