Images

View as Markdown
# Imports (run once)
from pathlib import Path
from air_sdk import AirApi
from air_sdk.endpoints.images import Image
from air_sdk.utils import sha256_file, wait_for_state
# Authentication (run once)
api = AirApi.with_ngc_config()
# OR api = AirApi.with_api_key(api_key="...")
# OR api = AirApi.with_device_login(email="...", org_num="...")
# ^ use in terminal only — not supported in Jupyter notebooks

CREATE

Creating a new image.

image: Image = api.images.create(
name='cumulus-vx-1.2.3',
default_username='user',
default_password='password',
version='1.0.0',
mountpoint='/mnt/my-image',
cpu_arch='x86',
includes_air_agent=True,
)
image.dict()
{'id': '935b994e-9e1e-4d85-9275-10f7b3111cd9',
'name': 'cumulus-vx-1.2.3',
'created': datetime.datetime(2025, 4, 22, 20, 30, 22, 744128, tzinfo=datetime.timezone.utc),
'modified': datetime.datetime(2025, 4, 22, 20, 30, 24, 744142, tzinfo=datetime.timezone.utc),
'published': False,
'includes_air_agent': True,
'cpu_arch': 'x86',
'default_username': 'user',
'default_password': 'password',
'version': '1.0.0',
'mountpoint': '/mnt/my-image',
'emulation_type': [],
'emulation_version': '',
'provider': 'VM',
'minimum_resources': {'cpu': 1, 'memory': 1024, 'storage': 10},
'is_owned_by_client': True,
'notes': '',
'release_notes': '',
'user_manual': '',
'upload_status': 'READY',
'last_uploaded_at': None,
'size': 0,
'hash': ''}
image_id = str(image.id)
image_id
'935b994e-9e1e-4d85-9275-10f7b3111cd9'

Create and Upload in One Step

You can also create an image and upload the file content in a single operation by providing the filepath parameter to create():

image: Image = api.images.create(
name='cumulus-vx-1.2.3',
default_username='user',
default_password='password',
version='2.0.0',
mountpoint='/mnt/my-image',
cpu_arch='x86',
includes_air_agent=True,
filepath='/home/user/images/cumulus-vx-5.0.0.qcow2',
)

GET

Retrieve a specific Image

image: Image = api.images.get(image_id)
image
Image(name='cumulus-vx-1.2.3', version='1.0.0', upload_status='READY')

List/Filter/Order/Search an iterable of Images

We can query for images with specific characteristics.

name_substring = 'cumulus-vx'
for image in api.images.list(search=name_substring, ordering='-name', cpu_arch='x86'):
print(image.name.ljust(25), image.created, image.upload_status)
cumulus-vx-5.6.0 2025-04-14 20:39:00.640224+00:00 READY

UPDATE

Update specific fields on an individual image.

image: Image = api.images.get(image_id)
# Perform the update
image.update(version='1.0.1')
image
Image(name='cumulus-vx-1.2.3', version='1.0.1', upload_status='READY')

UPLOAD FILE CONTENT

Upload the file content of the image (e.g. cumulus-vx-1.2.3.iso) to Air.

local_file_path = Path.home() / 'cumulus-vx-1.2.3.iso'
image: Image = api.images.get(image_id)
image.upload(filepath=local_file_path)
wait_for_state(image, 'COMPLETE', state_field='upload_status', error_states='READY')
image
Image(name='cumulus-vx-1.2.3', version='1.0.1', upload_status='COMPLETE')

Reset/clear the file content associated with an Image

If you want to upload different content to the image you must first call clear_upload to clear the uploaded content currently associated with the image. This step is in place to protect currently uploaded images.

Parallel uploads for large files

For large files, you can speed up uploads by using multiple parallel workers. The SDK automatically chunks files into ~100MB parts and uploads them to S3.

large_file_path = Path.home() / 'large-cumulus-image.qcow2'
image: Image = api.images.get(image_id)
# Upload with 4 parallel workers (recommended for large files on fast connections)
# Each worker uploads a ~100MB part concurrently
image.upload(filepath=large_file_path, max_workers=4)
# Or with custom timeout per part (default is 5 minutes per part)
# image.upload(filepath=large_file_path, max_workers=4, timeout=timedelta(minutes=10))
wait_for_state(image, 'COMPLETE', state_field='upload_status', error_states='READY')
print(f'Upload complete! Status: {image.upload_status}')
new_file = Path.home() / 'cumulus-vx-1.2.3.iso'
print('1. Status:', image.upload_status, 'Hash:', image.hash)
image.clear_upload()
print('2. Status:', image.upload_status, 'Hash:', image.hash)
image.upload(filepath=new_file)
wait_for_state(image, 'COMPLETE', state_field='upload_status', error_states='READY')
print('3. Status:', image.upload_status, 'Hash:', image.hash)
1. Status: COMPLETE Hash: 1894a19c85ba153acbf743ac4e43fc004c891604b26f8c69e1e83ea2afc7c48f
2. Status: READY Hash:
3. Status: COMPLETE Hash: ec7e5b4a32e4c00a786ded0a1632990716c2447f6f800fe96d253f91850e0ab3

Verifying / Checking Image Content

You can verify the content of an uploaded image by comparing the hash of an image with the hash of a local file.

1. Comparing hashes

An Image will have a populated hash when the Image has an associated file upload. This hash is the SHA256 hash of the file calculated using the air_sdk.utils.sha256_file method.

If you have a file locally, you can use this sha256_file method to determine your local hash and can compare this to the hash associated with the Image instance to see if the contents are identical.

local_file_path = Path.home() / 'cumulus-vx-1.2.3.iso'
local_file_hash = sha256_file(local_file_path)
print('Local hash:', local_file_hash)
image: Image = api.images.get(image_id)
print('Image hash:', image.hash)
if image.hash == local_file_hash:
print('The image content is identical to the local file content')
else:
print('Content is different')
Local hash: ec7e5b4a32e4c00a786ded0a1632990716c2447f6f800fe96d253f91850e0ab3
Image hash: ec7e5b4a32e4c00a786ded0a1632990716c2447f6f800fe96d253f91850e0ab3
The image content is identical to the local file content

DELETE

Delete an image

image: Image = api.images.get(image_id)
image.delete()

Image sharing

Share an image you own with another organization. The recipient must claim the share to copy the image into their org.

Pass target_org as the recipient’s NGC org name.

The returned share’s id is what the recipient passes to api.images.claim_image_share(image_share=...).

image: Image = api.images.get(image_id) # or any image you own
share = image.share(target_org='ngc-org-name')
print(f'Share id: {share.id}')
print(f'Image: {share.image_name}')
print(f'Target org: {share.target_org_display_name}')
print(f'State: {share.state}')
Share id: 3f9a7b21-5c48-4e3d-a2f1-8c0b7d9e1234
Image: image-share-test
Target org: Org-1
State: ACTIVE

List shares

api.image_shares lists shares visible to your account.

for s in api.image_shares.list():
print(f'{s.image_name} (id={s.id}) — state: {s.state}')
image-share-test (id=3f9a7b21-5c48-4e3d-a2f1-8c0b7d9e1234) — state: ACTIVE
image-1.0 (id=8c1f2a34-b7d9-4f02-9e8a-1c3d5b7a9f21) — state: EXPIRED
image-2.0 (id=52e7bd90-1a3c-4b8f-bc42-7f91d0e6ac34) — state: EXPIRED
image-3.0 (id=fa047c29-6d11-4e65-83b9-2e705d9c4b8e) — state: CLAIMED

Get a specific image share

share = api.images.shares.get('image-share-id') # replace with actual share id
print(f'Image: {share.image_name}')
print(f'Share id: {share.id}')
print(f'State: {share.state}')
Image: image-share-test
Share id: 3f9a7b21-5c48-4e3d-a2f1-8c0b7d9e1234
State: ACTIVE

Delete a specific image share (not the image itself)

Deleting an image share only removes that pending share so the target organization can no longer claim a copy, it does not delete the source image or its stored content.

share = api.images.shares.get('share-id') # replace with actual share id
share.delete()

Claim a shared image

If another organization shared an image to yours, claim it to create a copy in your org.

claimed = api.images.claim_image_share(image_share='share-id')
print(f'Claimed image: {claimed.name} (id={claimed.id})')
Claimed image: image-share-test (id=c47e9b12-3fa6-4d8b-92e1-5ab0c3d49f68)