air_sdk.utils#

Attributes#

`F`
`BUILTIN_TYPES`
`API_URI_PATTERN`
`COMPILED_API_URI_PATTERN`

Classes#

FilePartReader

File-like object that reads only a specific portion of a file.

Functions#

`filter_missing`(→ dict[str, Any])	Filter out MISSING values from kwargs.
`join_urls`(→ str)
`iso_string_to_datetime`(→ Optional[datetime.datetime])
`datetime_to_iso_string`(→ str)	Convert datetime to ISO string in UTC.
`to_uuid`(→ Optional[uuid.UUID])
`to_url`(→ Optional[urllib.parse.ParseResult])
`is_dunder`(→ bool)
`as_field`(→ Optional[dataclasses.Field])
`_resolve_type_hints_fallback`(→ dict[str, Any])	Fallback type hint resolution when get_type_hints() fails.
`validate_payload_types`(→ F)	A wrapper for validating the type of payload during create.
`sha256_file`(→ str)	Get the SHA256 hash of the local file.
`calculate_multipart_info`(→ list[dict[str, int]])	Calculate part information for multipart upload.
`create_short_uuid`(→ str)
`normalize_api_url`(→ str)	Ensures the API URL ends with the correct path.
`raise_if_invalid_response`(→ None)	Validates that a given API response has the expected status code and JSON payload
`wait_for_state`(→ None)	Wait for a model to reach one of the target states.

Module Contents#

air_sdk.utils.F#

air_sdk.utils.filter_missing(**kwargs: Any) → dict[str, Any][source]#

Filter out MISSING values from kwargs.

This is a helper function to remove dataclasses.MISSING sentinel values before passing kwargs to API methods.

Parameters:: **kwargs – Keyword arguments that may contain MISSING values
Returns:: Dictionary with MISSING values filtered out

air_sdk.utils.BUILTIN_TYPES#

air_sdk.utils.API_URI_PATTERN = '^/api/v\\d+.*$'#

air_sdk.utils.COMPILED_API_URI_PATTERN#

air_sdk.utils.join_urls(*args: str) → str[source]#

air_sdk.utils.iso_string_to_datetime(iso: str) → datetime.datetime | None[source]#

air_sdk.utils.datetime_to_iso_string(date: datetime.datetime) → str[source]#

Convert datetime to ISO string in UTC.

Accepts any timezone-aware datetime. For naive datetimes (no timezone), assumes local timezone and emits a warning.

Parameters:: date – The datetime to convert
Returns:: ISO 8601 formatted string in UTC (with ‘Z’ suffix)
Warns:: UserWarning – If datetime is naive (no timezone specified)

air_sdk.utils.to_uuid(uuid: str) → uuid.UUID | None[source]#

air_sdk.utils.to_url(url: str) → urllib.parse.ParseResult | None[source]#

air_sdk.utils.is_dunder(name: str) → bool[source]#

air_sdk.utils.as_field( class_or_instance: object, name: str, ) → dataclasses.Field | None[source]#

air_sdk.utils._resolve_type_hints_fallback( func: Callable[Ellipsis, Any], ) → dict[str, Any][source]#

Fallback type hint resolution when get_type_hints() fails.

This handles cases where TYPE_CHECKING imports aren’t available at runtime. Only validates types that can be resolved at runtime.

Parameters:: func – Function to extract type hints from
Returns:: Dictionary of resolvable type hints (name -> type)

air_sdk.utils.validate_payload_types(func: F) → F[source]#: A wrapper for validating the type of payload during create.

air_sdk.utils.sha256_file(path: str | pathlib.Path) → str[source]#: Get the SHA256 hash of the local file.

air_sdk.utils.calculate_multipart_info( file_size: int, chunk_size: int, ) → list[dict[str, int]][source]#

Calculate part information for multipart upload.

Parameters:

file_size – Total size of the file in bytes
chunk_size – Size of each chunk in bytes

Returns:

part_number: 1-based part number
start: Start byte offset
size: Size of this part in bytes

Return type:

List of dictionaries containing part info with keys

class air_sdk.utils.FilePartReader(file_path: str | pathlib.Path, start: int, size: int)[source]#

File-like object that reads only a specific portion of a file.

Used for streaming multipart uploads without loading entire file into memory. This class implements the context manager protocol and provides a read() method compatible with requests streaming uploads.

Parameters:

file_path – Path to the file to read from
start – Starting byte offset in the file
size – Number of bytes to read from the start offset

Example

>>> with FilePartReader('large_file.bin', start=0, size=5242880) as reader:
...     requests.put(presigned_url, data=reader)

file_path#

start#

size#

remaining#

f: BinaryIO | None = None#

__enter__() → FilePartReader[source]#

__exit__(*args: Any) → None[source]#

__len__() → int[source]#

Return the total size of this part.

This is required for requests to set Content-Length header instead of using Transfer-Encoding: chunked (which S3 doesn’t support).

read(chunk_size: int = -1) → bytes[source]#

Read up to chunk_size bytes, but never exceed the part size.

Parameters:: chunk_size – Number of bytes to read. If -1, reads remaining bytes.
Returns:: Bytes read from the file, up to the specified chunk size

air_sdk.utils.create_short_uuid() → str[source]#

air_sdk.utils.normalize_api_url(url: str) → str[source]#: Ensures the API URL ends with the correct path.

air_sdk.utils.raise_if_invalid_response( res: requests.Response, status_code: http.HTTPStatus = HTTPStatus.OK, data_type: type | None = dict, ) → None[source]#

Validates that a given API response has the expected status code and JSON payload

Arguments: res (requests.HTTPResponse) - API response object status_code [int] - Expected status code (default: 200)

Raises: AirUnexpectedResponse - Raised if an unexpected response is received from the API

air_sdk.utils.wait_for_state( model: air_sdk.air_model.AirModel, target_states: str | list[str], *, state_field: str = 'state', timeout: datetime.timedelta | None = None, poll_interval: datetime.timedelta | None = None, error_states: str | list[str] | None = None, ) → None[source]#

Wait for a model to reach one of the target states.

This is a generic utility that works with any AirModel that has a state field (e.g., Simulation.state, Node.state, Image.upload_status, etc.).

Parameters:

model – The AirModel instance to monitor (Simulation, Node, Image, etc.)
target_states – Single state or list of states to wait for
state_field – Name of the field containing the state (default: ‘state’). Use ‘upload_status’ for Images, ‘state’ for most other models.
timeout – Maximum time to wait (default: 120 seconds)
poll_interval – Time between status checks (default: 2 seconds)
error_states – Single state or list of states that should raise an error. If None, no error states are checked.

Raises:

ValueError – If the model enters one of the error states
TimeoutError – If timeout is reached before target state
AttributeError – If the model doesn’t have the specified state_field

Example

>>> # Wait for simulation to become active
>>> wait_for_state(simulation, 'ACTIVE', error_states=['INVALID', 'DELETING'])
>>>
>>> # Wait for image upload to complete
>>> wait_for_state(image, 'COMPLETE', state_field='upload_status')
>>>
>>> # Wait for node to boot or become active
>>> wait_for_state(node, ['BOOTING', 'ACTIVE'], error_states='ERROR')