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()
Handles file validation logic.
validate_file_exists
@classmethod
def validate_file_exists(cls, file_path: str) -> None
Validate that the file exists.
validate_file_size
@classmethod
def validate_file_size(cls, file_path: str, file_type: str) -> None
Validate file size against type-specific limits.
get_file_size_mb
@classmethod
def get_file_size_mb(cls, file_path: str) -> float
Get file size in MB.
MimeTypeDetector Objects
class MimeTypeDetector()
Handles MIME type detection with fallback support.
detect_mime_type
@classmethod
def detect_mime_type(cls, file_path: str) -> str
Detect MIME type with fallback support.
classify_file_type
@classmethod
def classify_file_type(cls, file_path: str, mime_type: str) -> str
Classify file type for size limit enforcement.
RequestManager Objects
class RequestManager()
Handles HTTP requests with retry logic.
create_session
@classmethod
def create_session(cls) -> requests.Session
Return the shared retry session, creating it on first use.
request_with_retry
@classmethod
def request_with_retry(cls, method: str, url: str,
**kwargs) -> requests.Response
Make HTTP request with retry logic.
A default (connect, read) timeout is applied when the caller doesn't
pass one: without it requests waits forever, so a peer that accepts
the connection and then goes silent pins the calling thread with no upper
bound. The read timeout bounds the gap between bytes, not the total
transfer, so large uploads are unaffected. Override per call with
timeout= or globally via AIXPLAIN_HTTP_CONNECT_TIMEOUT /
AIXPLAIN_HTTP_READ_TIMEOUT.
PresignedUrlManager Objects
class PresignedUrlManager()
Handles pre-signed URL requests to aiXplain backend.
get_temp_upload_url
@classmethod
def get_temp_upload_url(cls, backend_url: str) -> str
Get temporary upload URL endpoint.
get_perm_upload_url
@classmethod
def get_perm_upload_url(cls, backend_url: str) -> str
Get permanent upload URL endpoint.
build_temp_payload
@classmethod
def build_temp_payload(cls, content_type: str,
file_name: str) -> Dict[str, str]
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]
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]
Request pre-signed URL from backend.
S3Uploader Objects
class S3Uploader()
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
Upload file to S3 using pre-signed URL.
Raises:
UnsafeURLError- Ifpresigned_urldoes not point at an allowed upload host. The check sits outside thetrybelow so the genericFileUploadErrorwrapper cannot mask why the upload was refused (BUG-939).
construct_s3_url
@classmethod
def construct_s3_url(cls, presigned_url: str, path: str) -> str
Construct S3 URL from pre-signed URL and path.
ConfigManager Objects
class ConfigManager()
Handles configuration and environment variables.
get_backend_url
@classmethod
def get_backend_url(cls, custom_url: Optional[str] = None) -> str
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
Get API key from custom value or environment.
FileUploader Objects
class FileUploader()
Main file upload orchestrator.
__init__
def __init__(backend_url: Optional[str] = None,
api_key: Optional[str] = None,
require_api_key: bool = True)
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
Upload a file to S3 using the same logic as legacy FileFactory.
Arguments:
file_path- Path to the file to uploadtags- Tags to associate with the filelicense- License type for the fileis_temp- Whether this is a temporary uploadreturn_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
Convenience function to upload a file.
Arguments:
file_path- Path to the file to uploadtags- Tags to associate with the filelicense- License type for the fileis_temp- Whether this is a temporary uploadreturn_download_link- Whether to return download link instead of S3 pathbackend_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]
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