Skip to main content
Version: 2.0

aixplain.v2.upload_utils

File upload utilities for v2 Resource system.

This module provides comprehensive file upload functionality that ports the exact logic from the legacy FileFactory while maintaining a clean, modular architecture.

FileValidator Objects​

class FileValidator()

[view_source]

Handles file validation logic.

validate_file_exists​

@classmethod
def validate_file_exists(cls, file_path: str) -> None

[view_source]

Validate that the file exists.

validate_file_size​

@classmethod
def validate_file_size(cls, file_path: str, file_type: str) -> None

[view_source]

Validate file size against type-specific limits.

get_file_size_mb​

@classmethod
def get_file_size_mb(cls, file_path: str) -> float

[view_source]

Get file size in MB.

MimeTypeDetector Objects​

class MimeTypeDetector()

[view_source]

Handles MIME type detection with fallback support.

detect_mime_type​

@classmethod
def detect_mime_type(cls, file_path: str) -> str

[view_source]

Detect MIME type with fallback support.

classify_file_type​

@classmethod
def classify_file_type(cls, file_path: str, mime_type: str) -> str

[view_source]

Classify file type for size limit enforcement.

RequestManager Objects​

class RequestManager()

[view_source]

Handles HTTP requests with retry logic.

create_session​

@classmethod
def create_session(cls) -> requests.Session

[view_source]

Create a requests session with retry configuration.

request_with_retry​

@classmethod
def request_with_retry(cls, method: str, url: str,
                       **kwargs) -> requests.Response

[view_source]

Make HTTP request with retry logic.

PresignedUrlManager Objects​

class PresignedUrlManager()

[view_source]

Handles pre-signed URL requests to aiXplain backend.

get_temp_upload_url​

@classmethod
def get_temp_upload_url(cls, backend_url: str) -> str

[view_source]

Get temporary upload URL endpoint.

get_perm_upload_url​

@classmethod
def get_perm_upload_url(cls, backend_url: str) -> str

[view_source]

Get permanent upload URL endpoint.

build_temp_payload​

@classmethod
def build_temp_payload(cls, content_type: str,
                       file_name: str) -> Dict[str, str]

[view_source]

Build payload for temporary upload request.

build_perm_payload​

@classmethod
def build_perm_payload(cls, content_type: str, file_path: str, tags: List[str],
                       license: str) -> Dict[str, str]

[view_source]

Build payload for permanent upload request.

request_presigned_url​

@classmethod
def request_presigned_url(cls, url: str, payload: Dict[str, str],
                          api_key: str) -> Dict[str, Any]

[view_source]

Request pre-signed URL from backend.

S3Uploader Objects​

class S3Uploader()

[view_source]

Handles S3 file uploads using pre-signed URLs.

upload_file​

@classmethod
def upload_file(cls, file_path: str, presigned_url: str,
                content_type: str) -> None

[view_source]

Upload file to S3 using pre-signed URL.

construct_s3_url​

@classmethod
def construct_s3_url(cls, presigned_url: str, path: str) -> str

[view_source]

Construct S3 URL from pre-signed URL and path.

ConfigManager Objects​

class ConfigManager()

[view_source]

Handles configuration and environment variables.

get_backend_url​

@classmethod
def get_backend_url(cls, custom_url: Optional[str] = None) -> str

[view_source]

Get backend URL from custom value or environment.

get_api_key​

@classmethod
def get_api_key(cls,
                custom_key: Optional[str] = None,
                required: bool = True) -> str

[view_source]

Get API key from custom value or environment.

FileUploader Objects​

class FileUploader()

[view_source]

Main file upload orchestrator.

__init__​

def __init__(backend_url: Optional[str] = None,
             api_key: Optional[str] = None,
             require_api_key: bool = True)

[view_source]

Initialize file uploader with configuration.

upload​

def upload(file_path: str,
           tags: Optional[List[str]] = None,
           license: str = "MIT",
           is_temp: bool = True,
           return_download_link: bool = False) -> str

[view_source]

Upload a file to S3 using the same logic as legacy FileFactory.

Arguments:

  • file_path - Path to the file to upload
  • tags - Tags to associate with the file
  • license - License type for the file
  • is_temp - Whether this is a temporary upload
  • return_download_link - Whether to return download link instead of S3 path

Returns:

S3 path (s3://bucket/key) or download URL

Raises:

  • FileUploadError - If upload fails

upload_file​

def upload_file(file_path: str,
                tags: Optional[List[str]] = None,
                license: str = "MIT",
                is_temp: bool = True,
                return_download_link: bool = False,
                backend_url: Optional[str] = None,
                api_key: Optional[str] = None) -> str

[view_source]

Convenience function to upload a file.

Arguments:

  • file_path - Path to the file to upload
  • tags - Tags to associate with the file
  • license - License type for the file
  • is_temp - Whether this is a temporary upload
  • return_download_link - Whether to return download link instead of S3 path
  • backend_url - Custom backend URL (optional)
  • api_key - Custom API key (optional)

Returns:

S3 path (s3://bucket/key) or download URL

validate_file_for_upload​

def validate_file_for_upload(file_path: str) -> Dict[str, Any]

[view_source]

Validate a file for upload without actually uploading.

Arguments:

  • file_path - Path to the file to validate

Returns:

Dictionary with validation results

Raises:

  • FileUploadError - If validation fails