air_sdk.helpers#

Helper modules for the AIR SDK.

This package contains helper functions and utilities that support specific SDK operations but are not general-purpose utilities.

Submodules#

Functions#

abort_multipart_upload(→ None)

Abort a multipart upload to clean up S3 resources.

calculate_parts_info(→ list[dict[str, int]])

Calculate byte ranges for each part based on file size and number of parts.

complete_multipart_upload(→ None)

Complete a multipart upload by sending ETags to backend.

upload_image(→ air_sdk.endpoints.images.Image)

Upload an image file using multipart upload.

upload_parts_to_s3(→ list[dict[str, Any]])

Upload file parts directly to S3 using presigned URLs.

upload_single_part(→ dict[str, Any])

Upload a single part to S3 with retry logic for transient failures.

Package Contents#

air_sdk.helpers.abort_multipart_upload(
*,
api_client: air_sdk.AirAPI,
base_url: str,
image: air_sdk.endpoints.images.Image,
) → None[source]#

Abort a multipart upload to clean up S3 resources.

This should be called when a multipart upload fails to prevent orphaned uploads in S3 storage.

Parameters:

  • api_client – The AirApi client instance for making HTTP requests

  • base_url – Base URL for the images endpoint

  • image – Image instance (must be in UPLOADING status)

Note

This method does not raise exceptions on failure, as it’s meant for cleanup during error handling. Warnings are issued instead.

air_sdk.helpers.calculate_parts_info(
file_size: int,
num_parts: int,
chunk_size: int,
) → list[dict[str, int]][source]#

Calculate byte ranges for each part based on file size and number of parts.

The API determines the number of parts and chunk size automatically. This function calculates the actual byte ranges for each part.

Parameters:

  • file_size – Total file size in bytes

  • num_parts – Number of parts (from API response)

  • chunk_size – Size of each chunk in bytes (from API response)

Returns:

List of dicts with part_number, start, and size for each part

air_sdk.helpers.complete_multipart_upload(
*,
api_client: air_sdk.AirAPI,
base_url: str,
image: air_sdk.endpoints.images.Image,
uploaded_parts: list[dict[str, Any]],
) → None[source]#

Complete a multipart upload by sending ETags to backend.

Backend will use boto3.complete_multipart_upload() with the ETags.

Parameters:

  • api_client – The AirApi client instance for making HTTP requests

  • base_url – Base URL for the images endpoint

  • image – Image instance

  • uploaded_parts – List of uploaded parts with part_number and etag

Raises:

requests.RequestException – If completion fails

air_sdk.helpers.upload_image(
*,
api_client: air_sdk.AirAPI,
base_url: str,
image: air_sdk.endpoints.images.Image,
filepath: str | pathlib.Path,
timeout: datetime.timedelta | None = None,
max_workers: int = 1,
**kwargs: Any,
) → air_sdk.endpoints.images.Image[source]#

Upload an image file using multipart upload.

All uploads use multipart upload to S3. The API calculates parts (~100MB each) automatically based on file size.

The upload flow is: 1. Start upload with hash and size → get upload_id and presigned URLs 2. Upload each part directly to S3 using presigned URLs 3. Complete upload with part ETags

If any step fails, the multipart upload is automatically aborted to prevent orphaned data in S3 storage.

Parameters:

  • api_client – The AirApi client instance for making HTTP requests

  • base_url – Base URL for the images endpoint

  • image – Image instance

  • filepath – Path to the file to upload

  • timeout – Timeout per part upload (default: DEFAULT_UPLOAD_TIMEOUT = 5 minutes). This timeout applies to EACH part, not the entire multipart operation.

  • max_workers – Number of concurrent upload workers. Default: 1 (sequential). Set > 1 for parallel uploads (e.g., 4 for 4 concurrent uploads).

  • **kwargs – Additional arguments (currently unused, kept for API compatibility)

Note

Presigned URL expiration varies by file size. Check the backend documentation for the exact expiration time.

Returns:

Updated Image instance

Raises:
air_sdk.helpers.upload_parts_to_s3(
*,
api_client: air_sdk.AirAPI,
filepath: str | pathlib.Path,
parts_info: list[dict[str, int]],
part_urls: list[dict[str, Any]],
timeout_per_part: float,
max_workers: int = 1,
) → list[dict[str, Any]][source]#

Upload file parts directly to S3 using presigned URLs.

Supports both sequential (max_workers=1) and parallel (max_workers>1) uploads.

Parameters:

  • api_client – The AirApi client instance

  • filepath – Path to the file to upload

  • parts_info – List of part information (part_number, start, size)

  • part_urls – List of presigned URL data from backend

  • timeout_per_part – Timeout in seconds for each part upload

  • max_workers – Number of concurrent upload workers. Default: 1 (sequential)

Returns:

List of uploaded parts with part_number and etag, sorted by part_number

Raises:

AirUnexpectedResponse – If any part upload fails

air_sdk.helpers.upload_single_part(
*,
api_client: air_sdk.AirAPI,
filepath: str | pathlib.Path,
part_number: int,
start_offset: int,
part_size: int,
presigned_url: str,
timeout: float,
max_retries: int = DEFAULT_RETRY_ATTEMPTS,
) → dict[str, Any][source]#

Upload a single part to S3 with retry logic for transient failures.

Automatically retries on transient network errors (connection errors, timeouts, 503/429 responses) with exponential backoff. Non-transient errors are raised immediately.

Parameters:

  • api_client – The AirApi client instance (used for verify setting)

  • filepath – Path to the file to upload

  • part_number – Part number (1-indexed)

  • start_offset – Starting byte offset in file

  • part_size – Size of this part in bytes

  • presigned_url – S3 presigned URL for this part

  • timeout – Timeout in seconds for the upload

  • max_retries – Maximum number of retry attempts (default: DEFAULT_RETRY_ATTEMPTS)

Returns:

Dict with ‘part_number’ and ‘etag’ keys

Raises: