File Upload API Quickstart

import requests
import base64
import os
from typing import List, Optional

class FileUploadAPI:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = 'https://kieai.redpandaai.co'
        self.headers = {
            'Authorization': f'Bearer {api_key}'
        }
    
    def upload_file(self, file_path: str, upload_path: str = '', 
                   file_name: Optional[str] = None) -> dict:
        """File stream upload"""
        files = {
            'file': (os.path.basename(file_path), open(file_path, 'rb'))
        }
        
        data = {}
        if upload_path:
            data['uploadPath'] = upload_path
        if file_name:
            data['fileName'] = file_name
        
        response = requests.post(
            f'{self.base_url}/api/file-stream-upload',
            headers=self.headers,
            files=files,
            data=data
        )
        
        if not response.ok:
            raise Exception(f'Upload failed: {response.text}')
        
        return response.json()
    
    def upload_from_url(self, file_url: str, upload_path: str = '', 
                       file_name: Optional[str] = None) -> dict:
        """URL file upload"""
        payload = {
            'fileUrl': file_url,
            'uploadPath': upload_path,
            'fileName': file_name
        }
        
        response = requests.post(
            f'{self.base_url}/api/file-url-upload',
            headers={**self.headers, 'Content-Type': 'application/json'},
            json=payload
        )
        
        if not response.ok:
            raise Exception(f'Upload failed: {response.text}')
        
        return response.json()
    
    def upload_base64(self, base64_data: str, upload_path: str = '', 
                     file_name: Optional[str] = None) -> dict:
        """Base64 file upload"""
        payload = {
            'base64Data': base64_data,
            'uploadPath': upload_path,
            'fileName': file_name
        }
        
        response = requests.post(
            f'{self.base_url}/api/file-base64-upload',
            headers={**self.headers, 'Content-Type': 'application/json'},
            json=payload
        )
        
        if not response.ok:
            raise Exception(f'Upload failed: {response.text}')
        
        return response.json()

# Usage example
def main():
    uploader = FileUploadAPI('YOUR_API_KEY')
    
    # Batch upload local files
    file_paths = [
        '/path/to/file1.jpg',
        '/path/to/file2.png',
        '/path/to/document.pdf'
    ]
    
    print("Starting batch file upload...")
    for i, file_path in enumerate(file_paths):
        try:
            result = uploader.upload_file(
                file_path, 
                'user-uploads', 
                f'file-{i + 1}-{os.path.basename(file_path)}'
            )
            print(f"File {i + 1} uploaded successfully: {result['data']['fileUrl']}")
        except Exception as e:
            print(f"File {i + 1} upload failed: {e}")
    
    # Batch upload from URLs
    urls = [
        'https://example.com/image1.jpg',
        'https://example.com/image2.png'
    ]
    
    print("\nStarting batch URL upload...")
    for i, url in enumerate(urls):
        try:
            result = uploader.upload_from_url(
                url, 
                'downloads', 
                f'download-{i + 1}.jpg'
            )
            print(f"URL {i + 1} uploaded successfully: {result['data']['fileUrl']}")
        except Exception as e:
            print(f"URL {i + 1} upload failed: {e}")

if __name__ == '__main__':
    main()