air_sdk.endpoints#

Submodules#

Attributes#

`ServiceAPI`
`ServiceEndpointApi`
`SimulationApi`
`SimulationEndpointApi`
`SimulationNodeApi`
`ImageApi`
`ImageEndpointApi`
`MarketplaceDemoApi`
`MarketplaceDemoEndpointApi`
`InterfaceApi`
`InterfaceEndpointApi`
`SimulationInterfaceApi`
`NodeInstructionsEndpointApi`
`UserConfigAPI`
`UserConfigEndpointApi`
`ResourceBudget`
`ResourceBudgetEndpointAPI`

Classes#

`Fleet`	Helper class that provides a standard way to create an ABC using
`FleetEndpointAPI`	Returns an iterable of model objects.
`History`	Represents a history entry in the Air API.
`HistoryEndpointAPI`	API for querying history entries.
`Image`	Image model representing a network image.
`ImageEndpointAPI`	API client for image endpoints.
`ImageShare`	Helper class that provides a standard way to create an ABC using
`ImageShareEndpointAPI`	API client for shared image endpoints.
`Interface`	Interface model representing a network interface.
`InterfaceEndpointAPI`	API client for interface endpoints.
`Manifest`	Manifest model representing a simulator/platform configuration.
`ManifestEndpointAPI`	API client for manifest endpoints.
`MarketplaceDemoTag`	Marketplace demo tag model representing a tag for marketplace demos.
`MarketplaceDemoTagEndpointAPI`	API client for marketplace demo tag endpoints.
`MarketplaceDemo`	Marketplace demo model representing a marketplace demo.
`MarketplaceDemoEndpointAPI`	API client for marketplace demo endpoints.
`NodeInstruction`	Node instruction model representing an automation instruction for a node.
`NodeInstructionEndpointAPI`	API client for node instructions endpoints.
`Node`	Node model representing a network node.
`NodeEndpointAPI`	API client for simulation endpoints.
`Organization`	Represents an organization and its resource budget in the Air platform.
`OrganizationEndpointAPI`	API client for Organization / ResourceBudget endpoints.
`Service`	Represents a service in the Air API.
`ServiceEndpointAPI`	API for managing services.
`Simulation`	Simulation model representing a network simulation.
`SimulationEndpointAPI`	API client for simulation endpoints.
`SSHKey`	SSH Key model representing a user's SSH public key.
`SSHKeyEndpointAPI`	API client for SSH key endpoints.
`System`	System model representing a system in the AIR platform.
`SystemEndpointAPI`	Endpoint API for System operations.
`UserConfig`	UserConfig model representing cloud-init configurations.
`UserConfigEndpointAPI`	API interface for managing UserConfig resources.
`Worker`	Helper class that provides a standard way to create an ABC using
`WorkerClientCertificate`	Helper class that provides a standard way to create an ABC using
`WorkerClientCertificateEndpointAPI`	Returns an iterable of model objects.
`WorkerEndpointAPI`	Returns an iterable of model objects.
`ZTPScript`	A ZTP (Zero Touch Provisioning) script for a simulation.
`ZTPScriptEndpointAPI`	Retrieve, update, and delete ZTP scripts for simulations.

Package Contents#

class air_sdk.endpoints.Fleet[source]#

Bases: air_sdk.air_model.AirModel

Helper class that provides a standard way to create an ABC using inheritance.

id: str#

created: datetime.datetime#

modified: datetime.datetime#

name: str#

classmethod get_model_api() → type[FleetEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

update(*, name: str | dataclasses._MISSING_TYPE = MISSING) → None[source]#

Update specific fields of the fleet.

Example

>>> fleet = api.fleets.get('123e4567-e89b-12d3-a456-426614174000')
>>> fleet.update(name='new-name')

create_worker( *, fqdn: str, ip_address: str, cpu_arch: str | dataclasses._MISSING_TYPE = MISSING, ) → air_sdk.endpoints.workers.Worker[source]#

Create a new worker in the fleet.

Example

>>> fleet = api.fleets.get('123e4567-e89b-12d3-a456-426614174000')
>>> worker = fleet.create_worker(fqdn='w1.example.com', ip_address='1.1.1.1')
>>> print(worker)

property workers: air_sdk.endpoints.workers.WorkerEndpointAPI#: Query for the related workers of the fleet.

class air_sdk.endpoints.FleetEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.endpoints.mixins.ListApiMixin[Fleet], air_sdk.endpoints.mixins.CreateApiMixin[Fleet], air_sdk.endpoints.mixins.GetApiMixin[Fleet], air_sdk.endpoints.mixins.PatchApiMixin[Fleet], air_sdk.endpoints.mixins.DeleteApiMixin, air_sdk.air_model.BaseEndpointAPI[Fleet]

Returns an iterable of model objects.

Handles pagination in the background.

API_PATH = 'infra/fleets/'#

model#

list(

name: str | dataclasses._MISSING_TYPE = MISSING,

search: str | dataclasses._MISSING_TYPE = MISSING,

ordering: str | dataclasses._MISSING_TYPE = MISSING,

**params: Any,

) → Iterator[Fleet][source]#

List all fleets.

Example

>>> for fleet in api.fleets.list(ordering='name'):
...     print(fleet.name)

create(*, name: str) → Fleet[source]#

Create a new fleet.

Example

>>> fleet = api.fleets.create(name='My Fleet')
>>> print(fleet)

class air_sdk.endpoints.History[source]#

Bases: air_sdk.air_model.AirModel

Represents a history entry in the Air API.

History entries track actions and events for Air resources (simulations, nodes, etc.). They are immutable and read-only - history is created automatically by the Air API.

object_id#: ID of the entity this history entry is about (e.g., a simulation ID)

model#: Type of entity being tracked (e.g., ‘simulation’)

created#: Timestamp when the history entry was created

actor#: Email or identifier of the user who performed the action

description#: Human-readable description of what happened

category#: Category of the event. Values: ‘INFO’, ‘ERROR’

Note

History entries cannot be created, updated, or deleted via the SDK. They are automatically generated by the Air API.

object_id: str#

model: str#

created: datetime.datetime#

actor: str#

description: str#

category: str#

classmethod get_model_api() → type[HistoryEndpointAPI][source]#: Returns the respective AirModelAPI type for this model

refresh() → None[source]#

Refresh the history entry.

Raises:: NotImplementedError – History entries are immutable and cannot be refreshed

Note

History entries are read-only and created automatically by the Air API. They cannot be modified or refreshed.

class air_sdk.endpoints.HistoryEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.endpoints.mixins.ListApiMixin[History], air_sdk.air_model.BaseEndpointAPI[History]

API for querying history entries.

History entries are read-only records of actions and events for Air resources. Use this endpoint to track changes and audit activity.

Note

This endpoint only supports list() operations. History entries cannot be created, updated, or deleted via the API.

API_PATH: str#

model: type[History]#

List history entries with optional filtering and pagination.

Parameters:

model – Entity type to get history for (required). Values: ‘simulation’
object_id – Filter by the ID of the entity being tracked (e.g., a specific simulation’s ID)
actor – Filter by actor email or identifier
category – Filter by event category. Values: ‘INFO’, ‘ERROR’
search – Search for substrings in actor or description fields
ordering – Order by field (prefix with ‘-’ for descending). Available fields: actor, category, created, model, object_id
limit – Maximum number of results to return per page
offset – Number of results to skip (for pagination)

Yields:

History instances

Example

>>> # List all simulation history
>>> for entry in api.histories.list(model='simulation'):
...     print(entry.description)
>>>
>>> # Get history for a specific simulation
>>> for entry in api.histories.list(
...     model='simulation',
...     object_id='3dadd54d-583c-432e-9383-a2b0b1d7f551'
... ):
...     print(f'{entry.created}: {entry.description}')
>>>
>>> # Filter by category
>>> errors = list(api.histories.list(model='simulation', category='ERROR'))
>>> print(f'Found {len(errors)} errors')
>>>
>>> # Filter by actor
>>> user_actions = list(api.histories.list(
...     model='simulation',
...     actor='user@nvidia.com'
... ))
>>> print(f'User performed {len(user_actions)} actions')
>>>
>>> # Search descriptions
>>> for entry in api.histories.list(model='simulation', search='started'):
...     print(entry.description)
>>>
>>> # Order by creation time (newest first)
>>> for entry in api.histories.list(
...     model='simulation',
...     ordering='-created',
...     limit=5
... ):
...     print(f'{entry.created}: {entry.description}')
>>>
>>> # Pagination
>>> page_1 = list(api.histories.list(model='simulation', limit=10, offset=0))
>>> page_2 = list(api.histories.list(model='simulation', limit=10, offset=10))

class air_sdk.endpoints.Image[source]#

Bases: air_sdk.air_model.AirModel

Image model representing a network image.

id#: Unique identifier for the image

name#: Human-readable name of the image

version#: Version of the image

created#: Timestamp when the image was created

creator#: User who created the image

modified#: Timestamp when the image was last modified

mountpoint#: Mountpoint of the image

minimum_resources#: Minimum resources required to run the image

includes_air_agent#: Whether the image includes the Air agent

cpu_arch#: CPU architecture of the image

default_username#: Default username for the image

default_password#: Default password for the image

emulation_type#: The types of emulation the image supports

emulation_version#: The version of the emulation the image supports

provider#: Provider of the image

published#: Whether the image is published

upload_status#: Status of the image upload

last_uploaded_at#: Timestamp when the image was last uploaded

size#: Size of the image

hash#: Hash of the image

is_owned_by_client#: Whether the image is owned by the client

id: str#

name: str#

version: str#

created: datetime.datetime#

creator: str#

modified: datetime.datetime#

mountpoint: str | None#

minimum_resources: MinimumResources#

includes_air_agent: bool#

cpu_arch: str#

default_username: str#

default_password: str#

emulation_type: list[str]#

emulation_version: str#

provider: str#

published: bool#

upload_status: str#

last_uploaded_at: datetime.datetime | None#

size: int#

hash: str#

is_owned_by_client: bool#

classmethod get_model_api() → type[ImageEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: ImageEndpointAPI#

update( *, name: str | dataclasses._MISSING_TYPE = ..., version: str | dataclasses._MISSING_TYPE = ..., default_username: str | dataclasses._MISSING_TYPE = ..., default_password: str | dataclasses._MISSING_TYPE = ..., mountpoint: str | None | dataclasses._MISSING_TYPE = ..., cpu_arch: str | dataclasses._MISSING_TYPE = ..., includes_air_agent: bool | dataclasses._MISSING_TYPE = ..., emulation_type: list[str] | dataclasses._MISSING_TYPE = ..., emulation_version: str | dataclasses._MISSING_TYPE = ..., provider: str | dataclasses._MISSING_TYPE = ..., ) → None#

Update the image’s properties.

Parameters:

name – Name of the image
version – Version of the image
default_username – Default username for the image
default_password – Default password for the image
mountpoint – Mountpoint of the image
cpu_arch – CPU architecture of the image
includes_air_agent – Whether the image includes the Air agent
emulation_type – The types of emulation the image supports
emulation_version – The version of the emulation the image supports
provider – Provider of the image

Example

>>> image.update(name='new-name', version='1.0.0')
>>> image.update(default_username='user', default_password='pass')

upload( *, filepath: str | pathlib.Path, timeout: datetime.timedelta | None | dataclasses._MISSING_TYPE = ..., max_workers: int | dataclasses._MISSING_TYPE = ..., ) → Image#

Upload the image to the Air platform.

All uploads use multipart upload to S3. Parts are ~100MB each, calculated automatically by the API.

Parameters:

filepath – local file path to the image
timeout – Timeout per part upload (default: DEFAULT_UPLOAD_TIMEOUT).
max_workers – number of concurrent workers for parallel uploads (default: 1 for sequential).

Returns:

the uploaded image instance

Return type:

Image

Example

>>> image.upload(filepath='local_file_path')

clear_upload() → Image[source]#

Clear the upload status of the image.

Returns:: the image instance
Return type:: Image

Example

>>> image.clear_upload()

publish( *, name: str | dataclasses._MISSING_TYPE = ..., version: str | dataclasses._MISSING_TYPE = ..., ) → Image#

Publish the image.

Parameters:

name – new name of the image
version – new version of the image

Returns:

the published image instance

Return type:

Image

Example

>>> image.publish()
>>> image.publish(name='new-name', version='1.0.0')

unpublish( *, name: str | dataclasses._MISSING_TYPE = ..., version: str | dataclasses._MISSING_TYPE = ..., ) → Image[source]#

Unpublish the image.

Parameters:

name – new name of the image
version – new version of the image

Returns:

the image instance

Return type:

Image

Example

>>> image.unpublish()
>>> image.unpublish(name='new-name', version='1.0.0')

share( *, target_org: str, expires_at: datetime.datetime | dataclasses._MISSING_TYPE = ..., ) → ImageShare[source]#

Share the image with another organization.

Parameters:

target_org – The NGC org name of the organization receiving the image
expires_at – The date and time the image share expires

Returns:

The created share instance

Return type:

ImageShare

Example

>>> share = image.share(target_org='target-org-name')

class air_sdk.endpoints.ImageEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[Image]

API client for image endpoints.

API_PATH: str#

model: type[Image]#

create( *, name: str, version: str, default_username: str, default_password: str, mountpoint: str | None | dataclasses._MISSING_TYPE = ..., cpu_arch: str | dataclasses._MISSING_TYPE = ..., includes_air_agent: bool | dataclasses._MISSING_TYPE = ..., emulation_type: list[str] | dataclasses._MISSING_TYPE = ..., emulation_version: str | dataclasses._MISSING_TYPE = ..., provider: str | dataclasses._MISSING_TYPE = ..., filepath: str | pathlib.Path | dataclasses._MISSING_TYPE = ..., timeout: datetime.timedelta | None | dataclasses._MISSING_TYPE = ..., max_workers: int | dataclasses._MISSING_TYPE = ..., ) → Image#

Create a new image.

Parameters:

name – Name of the image
version – Version of the image
default_username – Default username for the image
default_password – Default password for the image
mountpoint – Mountpoint of the image
cpu_arch – CPU architecture of the image
includes_air_agent – Whether the image includes the Air agent
emulation_type – The types of emulation the image supports
emulation_version – The version of the emulation the image supports
provider – Provider of the image
filepath – Optional path to image file. If provided, uploads the image after creation using upload.
timeout – Timeout per part upload (default: DEFAULT_UPLOAD_TIMEOUT). Only used if filepath is provided.
max_workers – Number of concurrent workers for parallel uploads (default: 1). Only used if filepath is provided.

Returns:

The created Image instance

Example

>>> # Create image without upload
>>> api.images.create(
...     name='cumulus-vx-1.2.3',
...     version='1.0.0',
...     default_username='user',
...     default_password='password',
... )

>>> # Create and upload image in single step
>>> api.images.create(
...     name='cumulus-vx-1.2.3',
...     version='1.0.0',
...     default_username='user',
...     default_password='password',
...     filepath='./cumulus-vx.qcow2',
... )

>>> # Create and upload with parallel workers
>>> api.images.create(
...     name='cumulus-vx-1.2.3',
...     version='1.0.0',
...     default_username='user',
...     default_password='password',
...     filepath='./large-image.qcow2',
...     max_workers=4,
... )

list( *, name: str = ..., version: str = ..., cpu_arch: Literal['x86', 'ARM'] = ..., creator: str = ..., includes_air_agent: bool = ..., provider: Literal['VM', 'CONTAINER'] = ..., published: bool = ..., upload_status: Literal['READY', 'UPLOADING', 'VALIDATING', 'COMPLETED', 'PUBLISHED', 'UNPUBLISHED', 'UNPUBLISHING', 'COPYING_FROM_IMAGE_SHARE'] = ..., hash: str = ..., is_owned_by_client: bool = ..., limit: int = ..., offset: int = ..., ordering: str = ..., search: str = ..., ) → Iterator[Image]#

List all images with optional filtering.

Parameters:

name – Name of the image
version – Version of the image
cpu_arch – CPU architecture of the image
creator – Creator of the image
includes_air_agent – Whether the image includes the Air agent
emulation_type – The types of emulation the image supports
emulation_version – The version of the emulation the image supports
provider – Provider of the image
published – Whether the image is published
upload_status – Status of the image upload
last_uploaded_at – Timestamp when the image was last uploaded
hash – Hash of the image
is_owned_by_client – Whether the image is owned by the client
limit – Maximum number of results to return
offset – Offset for pagination
ordering – Ordering of the results
search – Search query

Returns:

Iterator of Image instances

Example

>>> for image in api.images.list():
...     print(image.name)

>>> # Filter by name
>>> for image in api.images.list(search='image-name'):
...     print(image.name)

>>> # Order by name descending
>>> for image in api.images.list(ordering='-name'):
...     print(image.name)

get(pk: air_sdk.air_model.PrimaryKey) → Image#

Get a specific image by ID.

Parameters:: pk – The image ID (string or UUID)
Returns:: The Image instance

Example

>>> image = api.images.get('image-id')

Upload the image to the Air platform.

All uploads use multipart upload to S3. Parts are ~100MB each, calculated automatically by the API.

Parameters:

image – Image instance or image ID
filepath – Path to the file to upload
timeout – Timeout per part upload (default: DEFAULT_UPLOAD_TIMEOUT). This timeout applies to EACH part upload (not total operation).
max_workers – Number of concurrent workers for uploads. Default: 1 (sequential uploads). Set > 1 for parallel uploads.

Returns:

Updated Image instance

Raises:

FileNotFoundError – If the file does not exist
ValueError – If filepath is not a regular file or max_workers < 1
PermissionError – If the file is not readable
AirUnexpectedResponse – If upload fails or backend returns invalid data
requests.RequestException – For network/HTTP errors

Example

>>> # File upload
>>> image.upload(filepath='image.qcow2')

>>> # Large file with parallel upload
>>> image.upload(filepath='large.qcow2', max_workers=4)

clear_upload(*, image: Image | air_sdk.air_model.PrimaryKey) → None[source]#

Clear the upload status of the image.

Parameters:: image – image to clear upload
Returns:: None

Example

>>> api.images.clear_upload(image)

publish( *, image: Image | air_sdk.air_model.PrimaryKey, name: str | dataclasses._MISSING_TYPE = ..., version: str | dataclasses._MISSING_TYPE = ..., ) → Image#

Publish the image.

Args: Required parameters:

image: image to publish

Optional Parameters:: name: The name of the image version: The version of the image

Returns:: None

Example

>>> api.images.publish(image)

unpublish( *, image: Image | air_sdk.air_model.PrimaryKey, name: str | dataclasses._MISSING_TYPE = ..., version: str | dataclasses._MISSING_TYPE = ..., ) → Image[source]#

Unpublish the image.

Parameters:

image – image to unpublish (Image instance or image ID)
name – new name of the image
version – new version of the image

Returns:

the unpublished image instance

Return type:

Image

Example

>>> api.images.unpublish(image)
>>> api.images.unpublish(image, name='new-name', version='new-version')

share( *, image: Image | air_sdk.air_model.PrimaryKey, target_org: str, expires_at: datetime.datetime | dataclasses._MISSING_TYPE = ..., ) → ImageShare[source]#

Share the image with another organization.

Args: Required parameters:

image: The image to share (Image instance or image ID) target_org: The NGC org name of the organization receiving the image

Optional parameters:: expires_at: The date and time the image share expires

Returns:: The created share object
Return type:: ImageShare

Example

>>> share = api.images.share(image='image-id', target_org='target-org-name')

claim_image_share( *, image_share: air_sdk.air_model.PrimaryKey, name: str | dataclasses._MISSING_TYPE = ..., version: str | dataclasses._MISSING_TYPE = ..., ) → Image[source]#

Claim a shared image into your organization.

Parameters:

image_share – The share ID to claim (string or UUID)
name – new name of the image
version – new version of the image

Returns:

The claimed image

Return type:

Image

Example

>>> image = api.images.claim_image_share(image_share='share-id')
>>> image = api.images.claim_image_share(
...     image_share='share-id', name='new-name', version='1.0.0'
... )

property shares: ImageShareEndpointAPI#

Access the image shares API.

Returns:: The API for managing image shares
Return type:: ImageShareEndpointAPI

Example

>>> # List all shared images
>>> for share in api.images.shares.list():
...     print(share.image_name)

>>> # Create a new share
>>> share = api.images.shares.create(
...     image='image-id',
...     target_org='target-org-name',
... )

>>> # Get a specific share
>>> share = api.images.shares.get('share-id')

>>> # Delete a share
>>> api.images.shares.delete('share-id')

class air_sdk.endpoints.ImageShare[source]#

Bases: air_sdk.air_model.AirModel

Helper class that provides a standard way to create an ABC using inheritance.

id: str#

created: datetime.datetime#

modified: datetime.datetime#

image: Image#

image_name: str#

image_version: str#

source_org_display_name: str#

target_org_display_name: str#

target_org: str#

expires_at: datetime.datetime#

claimed_by: str#

state: str#

classmethod get_model_api() → type[ImageShareEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: ImageShareEndpointAPI#

class air_sdk.endpoints.ImageShareEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[ImageShare]

API client for shared image endpoints.

API_PATH: str#

model: type[ImageShare]#

list( *, limit: int | dataclasses._MISSING_TYPE = ..., offset: int | dataclasses._MISSING_TYPE = ..., ordering: str | dataclasses._MISSING_TYPE = ..., search: str | dataclasses._MISSING_TYPE = ..., ) → Iterator[ImageShare]#

List all shared images.

Parameters:

limit – Maximum number of results to return
offset – Offset for pagination
ordering – Ordering of the results
search – Search query

Returns:

Iterator of ImageShare instances

Example

>>> # List all shared images
>>> for share in api.images.shares.list():
...     print(share.image_name)

>>> # Filter by image name
>>> for share in api.images.shares.list(search='image-name'):
...     print(share.image_name)

>>> # Order by image name descending
>>> for share in api.images.shares.list(ordering='-image_name'):
...     print(share.image_name)

create( *, image: Image | air_sdk.air_model.PrimaryKey, target_org: str, expires_at: datetime.datetime | dataclasses._MISSING_TYPE = ..., ) → ImageShare#

Create a new image share.

Parameters:

image – The image or image ID to share
target_org – The NGC org name of the organization receiving the image
expires_at – The date and time the image share expires

Returns:

The created image share instance

Return type:

ImageShare

Example

>>> share = api.images.shares.create(
...     image='image-id', target_org='target-org-name'
... )

get(pk: air_sdk.air_model.PrimaryKey) → ImageShare#

Get a specific shared image by ID.

Parameters:: pk – The image share ID (string or UUID)
Returns:: The ImageShare instance

Example

>>> share = api.images.shares.get('share-id')

delete(pk: air_sdk.air_model.PrimaryKey) → None#

Delete (unshare) an image share.

Parameters:: pk – The share ID to delete (string or UUID)
Returns:: None

Example

>>> api.images.shares.delete('share-id')
>>> api.images.shares.delete(share.id)

class air_sdk.endpoints.Interface[source]#

Bases: air_sdk.air_model.AirModel

Interface model representing a network interface.

id#: Unique identifier for the interface

name#: Human-readable name of the interface

created#: Timestamp when the interface was created

modified#: Timestamp when the interface was last modified

node#: Node that the interface is related to

interface_type#: Type of the interface

mac_address#: MAC address of the interface

connection#: Interface that the interface is connected to

outbound#: Whether the interface is outbound

attributes#: Attributes of the interface

id: str#

name: str#

created: datetime.datetime#

modified: datetime.datetime#

node: air_sdk.endpoints.nodes.Node#

interface_type: str#

mac_address: str#

connection: Interface | None#

outbound: bool#

attributes: InterfaceAttributes | None#

classmethod get_model_api() → type[InterfaceEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: InterfaceEndpointAPI#

update( *, name: str | dataclasses._MISSING_TYPE = ..., interface_type: Literal['DATA_PLANE_INTF', 'PCIE_INTF', 'OOB_INTF'] | dataclasses._MISSING_TYPE = ..., outbound: bool | dataclasses._MISSING_TYPE = ..., attributes: InterfaceAttributes | None | dataclasses._MISSING_TYPE = ..., ) → None#

Update the interface’s properties.

Parameters:

name – New name for the interface
interface_type – New type for the interface
outbound – New outbound status for the interface
attributes – New attributes for the interface

Example

>>> interface.update(name='New Name')

delete() → None#

Delete the interface.

Example

>>> interface.delete()

connect( *, target: Interface | air_sdk.air_model.PrimaryKey, ) → Interface[source]#

Connect the interface to another interface.

Parameters:: target – Interface or primary key of the interface to connect to

Example

>>> interface.connect(interface)
>>> interface.connect('interface-id')

disconnect() → Interface[source]#

Disconnect the interface from its connection.

Example

>>> interface.disconnect()

property services: air_sdk.endpoints.services.ServiceEndpointAPI#

Query for the related services of the interface.

Returns:: ServiceEndpointAPI instance filtered for this interface’s services

Example

>>> for service in interface.services.list():
...     print(f'{service.name}: {service.worker_fqdn}:{service.worker_port}')
>>>
>>> # Create service on this interface
>>> service = interface.services.create(
...     name='SSH', node_port=22, service_type='ssh'
... )

class air_sdk.endpoints.InterfaceEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[Interface]

API client for interface endpoints.

API_PATH: str#

model: type[Interface]#

create( *, name: str, node: air_sdk.endpoints.nodes.Node | air_sdk.air_model.PrimaryKey, interface_type: Literal['DATA_PLANE_INTF', 'PCIE_INTF', 'OOB_INTF'] | dataclasses._MISSING_TYPE = ..., mac_address: str | dataclasses._MISSING_TYPE = ..., outbound: bool | dataclasses._MISSING_TYPE = ..., attributes: InterfaceAttributes | None | dataclasses._MISSING_TYPE = ..., ) → Interface#

Create a new interface.

Parameters:

name – Name of the interface
node – Node to create the interface on

Returns:

The created Interface instance

Example

>>> interface = api.interfaces.create(name='eth0', node=node)
>>> interface = node.interfaces.create(name='eth0', node=node.id)

list( *, interface_type: Literal['DATA_PLANE_INTF', 'PCIE_INTF', 'OOB_INTF'] | dataclasses._MISSING_TYPE = ..., mac_address: str | dataclasses._MISSING_TYPE = ..., name: str | dataclasses._MISSING_TYPE = ..., node: air_sdk.endpoints.nodes.Node | air_sdk.air_model.PrimaryKey | dataclasses._MISSING_TYPE = ..., outbound: bool | dataclasses._MISSING_TYPE = ..., simulation: str | dataclasses._MISSING_TYPE = ..., limit: int | dataclasses._MISSING_TYPE = ..., offset: int | dataclasses._MISSING_TYPE = ..., search: str | dataclasses._MISSING_TYPE = ..., ordering: str | dataclasses._MISSING_TYPE = ..., ) → Iterator[Interface]#

List all interfaces with optional filtering.

Parameters:

interface_type – Filter by interface type
mac_address – Filter by MAC address
name – Filter by name
node – Filter by node
outbound – Filter by outbound status
simulation – Filter by simulation
limit – Number of results to return per page
offset – The initial index from which to return the results
search – Search by name
ordering – Order by field

Returns:

Iterator of Interface instances

Example

>>> # List all interfaces
>>> for interface in api.interfaces.list():
...     print(interface.name)

>>> # Filter by interface type
>>> for interface in api.interfaces.list(interface_type='DATA_PLANE_INTF'):
...     print(interface.name)

>>> # Search by name
>>> for interface in api.interfaces.list(search='eth0'):
...     print(interface.name)

>>> # Order by name descending
>>> for interface in api.interfaces.list(ordering='-name'):
...     print(interface.name)

update( *, interface: Interface | air_sdk.air_model.PrimaryKey, name: str | dataclasses._MISSING_TYPE = ..., interface_type: Literal['DATA_PLANE_INTF', 'PCIE_INTF', 'OOB_INTF'] | dataclasses._MISSING_TYPE = ..., outbound: bool | dataclasses._MISSING_TYPE = ..., attributes: InterfaceAttributes | None | dataclasses._MISSING_TYPE = ..., ) → Interface#

Update the interface’s properties.

Parameters:

interface – Interface or primary key of the interface to update
name – New name for the interface
interface_type – New type for the interface
outbound – New outbound status for the interface
attributes – New attributes for the interface

Returns:

The updated Interface instance

Example

>>> # Using Interface object
>>> updated_interface = api.interfaces.update(
...     interface=interface, name='New Name'
... )
>>> # Using interface ID
>>> updated_interface = api.interfaces.update(
...     interface='interface-id',
...     name='New Name',
...     interface_type='DATA_PLANE_INTF',
... )

get(pk: air_sdk.air_model.PrimaryKey) → Interface#

Get a specific interface by ID.

Parameters:: pk – The interface ID (string or UUID)
Returns:: The Interface instance

Example

>>> interface = api.interfaces.get('interface-id')

delete(pk: air_sdk.air_model.PrimaryKey) → None#

Delete a specific interface by ID.

Parameters:: pk – The interface ID (string or UUID)

Example

>>> api.interfaces.delete('interface-id')

connect( *, interface: Interface | air_sdk.air_model.PrimaryKey, target: Interface | air_sdk.air_model.PrimaryKey, ) → Interface[source]#

Connect the interface to another interface.

Parameters:

interface – Interface or primary key of the interface to connect
target – Interface or primary key of the interface to connect to

Returns:

The source interface instance

Example

>>> api.interfaces.connect(interface=interface, target=target)
>>> api.interfaces.connect(interface='interface-id', target='target-id')

disconnect( *, interface: Interface | air_sdk.air_model.PrimaryKey, ) → Interface[source]#

Disconnect the interface from its connection.

Parameters:: interface – Interface or primary key of the interface to disconnect
Returns:: The source interface instance

Example

>>> api.interfaces.disconnect(interface=interface)
>>> api.interfaces.disconnect(interface='interface-id')

class air_sdk.endpoints.Manifest[source]#

Bases: air_sdk.bc.BaseCompatMixin, air_sdk.bc.manifests.ManifestCompatMixin, air_sdk.air_model.AirModel

Manifest model representing a simulator/platform configuration.

Manifests define how a specific network device or platform should be emulated, including resource requirements, docker parameters, and platform capabilities.

id#: Unique identifier for the manifest (required)

org_name#: Organization name that owns this manifest (read-only, set by API)

docker_run_parameters#: Docker container runtime parameters (required)

emulation_type#: Type of emulation (required)

platform_information#: Dictionary of platform configurations (required)

simulator_image#: Docker image used for the simulator (required)

simulator_resources#: Resource requirements for the simulator (required)

description#: Manifest description (optional)

artifacts_directory#: Directory path for simulator artifacts (optional)

artifacts_directory_max_size_gb#: Maximum size for artifacts dir in GB (optional)

boot_group#: Boot group number for startup ordering (optional)

configure_node_properties#: Node configuration properties (optional)

configure_simulator#: Simulator configuration parameters (optional)

simulation_engine_versions#: List of supported sim engine versions (optional)

emulation_params#: Emulation-specific parameters (optional)

port_mapping_required#: Whether port mapping is required (optional, read-only)

id: str#

org_name: str#

docker_run_parameters: dict[str, Any]#

emulation_type: str#

platform_information: dict[str, Any]#

simulator_image: air_sdk.endpoints.images.Image#

simulator_resources: dict[str, Any]#

description: str | None#

artifacts_directory: str | None#

artifacts_directory_max_size_gb: int | None#

boot_group: int | None#

configure_node_properties: dict[str, Any] | None#

configure_simulator: dict[str, Any] | None#

simulation_engine_versions: list[str] | None#

emulation_params: dict[str, Any] | None#

port_mapping_required: bool | None#

classmethod get_model_api() → type[ManifestEndpointAPI][source]#

Returns the respective AirModelAPI type for this model.

Returns:: ManifestEndpointAPI class

property model_api: ManifestEndpointAPI#

The current model API instance.

Returns:: ManifestEndpointAPI instance

update( *, docker_run_parameters: air_sdk.types.DockerRunParameters | dict[str, Any] = ..., emulation_type: str = ..., platform_information: dict[str, air_sdk.types.Platform | dict[str, Any]] = ..., simulator_image: air_sdk.endpoints.images.Image | str = ..., simulator_resources: air_sdk.types.Resources | dict[str, Any] = ..., description: str = ..., artifacts_directory: str = ..., artifacts_directory_max_size_gb: int = ..., boot_group: int = ..., configure_node_properties: dict[str, Any] = ..., configure_simulator: dict[str, Any] = ..., simulation_engine_versions: list[str] = ..., emulation_params: air_sdk.types.EmulationParams | dict[str, Any] = ..., ) → None#

Update the manifest’s properties.

Parameters:

docker_run_parameters – Docker runtime parameters
emulation_type – Type of emulation
platform_information – Platform configurations
simulator_image – Docker image for the simulator
simulator_resources – Resource requirements
description – Manifest description
artifacts_directory – Directory path for simulator artifacts
artifacts_directory_max_size_gb – Max size for artifacts dir
boot_group – Boot group number
configure_node_properties – Node configuration properties
configure_simulator – Simulator configuration
simulation_engine_versions – Supported engine versions
emulation_params – Emulation parameters

Example

>>> manifest.update(boot_group=2)
>>> manifest.update(description='Updated manifest')

delete() → None#

Delete this manifest.

Example

>>> manifest = api.manifests.get('manifest-id-123')
>>> manifest.delete()

class air_sdk.endpoints.ManifestEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[Manifest]

API client for manifest endpoints.

API_PATH: str#

model: type[Manifest]#

list(

*,

id: str = ...,

org_name: str = ...,

emulation_type: str = ...,

port_mapping_required: bool = ...,

limit: int = ...,

offset: int = ...,

ordering: str = ...,

search: str = ...,

**params: Any,

) → Iterator[Manifest]#

List all manifests with optional filtering.

Parameters:

id – Filter by manifest ID
org_name – Filter by organization name
emulation_type – Filter by emulation type
port_mapping_required – Filter by port mapping requirement (V2 BC field)
limit – Number of results to return per page
offset – The initial index from which to return the results
ordering – Order objects by field. Prefix with “-” for desc order
search – Search term to filter manifests
**params – Additional query parameters

Returns:

Iterator of Manifest instances

Example

>>> # List all manifests
>>> for manifest in api.manifests.list():
...     print(manifest.emulation_type)

>>> # Filter by emulation type
>>> for manifest in api.manifests.list(emulation_type='NIC_INFINIBAND'):
    ...     print(manifest.org_name)

>>> # Search and order
>>> for manifest in api.manifests.list(search='cumulus', ordering='-created'):
...     print(manifest.id)

create(

*,

docker_run_parameters: air_sdk.types.DockerRunParameters | dict[str, Any],

emulation_type: str,

platform_information: dict[str, air_sdk.types.Platform | dict[str, Any]],

simulator_image: air_sdk.endpoints.images.Image | str,

simulator_resources: air_sdk.types.Resources | dict[str, Any],

description: str = ...,

artifacts_directory: str = ...,

artifacts_directory_max_size_gb: int = ...,

boot_group: int = ...,

configure_node_properties: dict[str, Any] = ...,

configure_simulator: dict[str, Any] = ...,

simulation_engine_versions: List[str] = ...,

emulation_params: air_sdk.types.EmulationParams | dict[str, Any] = ...,

**kwargs: Any,

) → Manifest#

Create a new manifest.

Note: org_name is automatically determined by API and cannot be specified.

Parameters:

docker_run_parameters – Docker runtime parameters (required)
emulation_type – Type of emulation (required)
platform_information – Platform configurations (required)
simulator_image – Docker image for the simulator - Image obj or ID (required)
simulator_resources – Resource requirements (required)
description – Manifest description (optional)
artifacts_directory – Directory path for simulator artifacts (optional)
artifacts_directory_max_size_gb – Max size for artifacts dir (optional)
boot_group – Boot group number (optional)
configure_node_properties – Node configuration properties (optional)
configure_simulator – Simulator configuration (optional)
simulation_engine_versions – Supported engine versions (optional)
emulation_params – Emulation parameters (optional)
**kwargs – Additional fields

Returns:

The created Manifest instance

Example

>>> manifest = api.manifests.create(
...     artifacts_directory='/artifacts',
...     artifacts_directory_max_size_gb=10,
...     boot_group=1,
...     configure_node_properties={},
...     configure_simulator={},
...     docker_run_parameters={
...         'tmpfs': [],
...         'cap_add': ['NET_ADMIN'],
...         'devices': ['/dev/kvm'],
...         'volumes': [],
...         'environment': {},
...     },
...     emulation_type='NIC_INFINIBAND',
...     platform_information={},
...     simulation_engine_versions=['1.0'],
...     simulator_image='image-id',
...     simulator_resources={'cpu': 2.0, 'memory': 4096},
...     emulation_params={
...         'direct_link_emulation': True,
...         'max_network_pci': 8,
...     },
... )

get(pk: air_sdk.air_model.PrimaryKey) → Manifest#

Get a specific manifest by ID.

Parameters:: pk – The manifest ID (string or UUID)
Returns:: The Manifest instance

Example

>>> manifest = api.manifests.get('manifest-id-123')
>>> print(manifest.emulation_type)

patch(

pk: air_sdk.air_model.PrimaryKey,

*,

docker_run_parameters: air_sdk.types.DockerRunParameters | dict[str, Any] = ...,

emulation_type: str = ...,

platform_information: dict[str, air_sdk.types.Platform | dict[str, Any]] = ...,

simulator_image: air_sdk.endpoints.images.Image | str = ...,

simulator_resources: air_sdk.types.Resources | dict[str, Any] = ...,

description: str = ...,

artifacts_directory: str = ...,

artifacts_directory_max_size_gb: int = ...,

boot_group: int = ...,

configure_node_properties: dict[str, Any] = ...,

configure_simulator: dict[str, Any] = ...,

simulation_engine_versions: List[str] = ...,

emulation_params: air_sdk.types.EmulationParams | dict[str, Any] = ...,

**kwargs: Any,

) → Manifest#

Update a manifest’s properties.

Parameters:

pk – The manifest ID (string or UUID)
docker_run_parameters – Docker runtime parameters
emulation_type – Type of emulation
platform_information – Platform configurations
simulator_image – Docker image for the simulator
simulator_resources – Resource requirements
description – Manifest description
artifacts_directory – Directory path for simulator artifacts
artifacts_directory_max_size_gb – Max size for artifacts dir
boot_group – Boot group number
configure_node_properties – Node configuration properties
configure_simulator – Simulator configuration
simulation_engine_versions – Supported engine versions
emulation_params – Emulation parameters
**kwargs – Additional fields to update

Returns:

The updated Manifest instance

Example

>>> updated_manifest = api.manifests.patch(
...     pk='manifest-id-123',
...     boot_group=2,
... )

delete(pk: air_sdk.air_model.PrimaryKey, **kwargs: Any) → None#

Delete a manifest.

Parameters:

pk – The manifest ID (string or UUID) to delete
**kwargs – Additional parameters

Example

>>> api.manifests.delete('manifest-id-123')

class air_sdk.endpoints.MarketplaceDemoTag[source]#

Bases: air_sdk.air_model.AirModel

Marketplace demo tag model representing a tag for marketplace demos.

id#: Unique identifier for the tag

name#: Name of the tag

created#: Timestamp when the tag was created

modified#: Timestamp when the tag was last modified

id: str#

name: str#

created: datetime.datetime#

modified: datetime.datetime#

classmethod get_model_api() → type[MarketplaceDemoTagEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: MarketplaceDemoTagEndpointAPI#

class air_sdk.endpoints.MarketplaceDemoTagEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[MarketplaceDemoTag]

API client for marketplace demo tag endpoints.

API_PATH: str#

model: type[MarketplaceDemoTag]#

create(*, name: str) → MarketplaceDemoTag#

Create a new marketplace demo tag.

Parameters:: name – Name for the new tag
Returns:: The created MarketplaceDemoTag instance

Example

>>> tag = api.marketplace_demo_tags.create(name='networking')

list( *, limit: int | dataclasses._MISSING_TYPE = ..., offset: int | dataclasses._MISSING_TYPE = ..., search: str | dataclasses._MISSING_TYPE = ..., ordering: str | dataclasses._MISSING_TYPE = ..., ) → Iterator[MarketplaceDemoTag]#

List all marketplace demo tags.

Optional parameters:: limit: Number of results to return per page offset: The initial index from which to return the results search: Search term to filter tags ordering: Order the response by the specified field

Returns:: Iterator of MarketplaceDemoTag instances

Example

>>> # List all tags
>>> for tag in api.marketplace_demo_tags.list():
...     print(tag.name)

get(pk: air_sdk.air_model.PrimaryKey) → MarketplaceDemoTag#

Get a specific marketplace demo tag by ID.

Parameters:: pk – The tag ID (string or UUID)
Returns:: The MarketplaceDemoTag instance

Example

>>> tag = api.marketplace_demo_tags.get('tag-id')
>>> print(tag.name)

patch( pk: air_sdk.air_model.PrimaryKey, *, name: str | dataclasses._MISSING_TYPE = ..., ) → MarketplaceDemoTag#

Update a marketplace demo tag.

Parameters:

pk – The tag ID (string or UUID)
name – New name for the tag

Returns:

The updated MarketplaceDemoTag instance

Example

>>> tag = api.marketplace_demo_tags.patch('tag-id', name='new-name')

delete(pk: air_sdk.air_model.PrimaryKey) → None#

Delete a marketplace demo tag.

Parameters:: pk – The tag ID (string or UUID)

Example

>>> api.marketplace_demo_tags.delete('tag-id')

class air_sdk.endpoints.MarketplaceDemo[source]#

Bases: air_sdk.air_model.AirModel

Marketplace demo model representing a marketplace demo.

id#: Unique identifier for the marketplace demo

name#: Human-readable name of the marketplace demo

created#: Timestamp when the marketplace demo was created

modified#: Timestamp when the marketplace demo was last modified

creator#: The creator of the marketplace demo

description#: The description of the demo

documentation#: The documentation of the marketplace demo

repo#: The repository of the marketplace demo

tags#: The tags of the marketplace demo

like_count#: How many unique users have liked the marketplace demo

liked_by_client#: Whether the current user has liked the marketplace demo

published#: Whether the marketplace demo is published

icon#: The icon of the marketplace demo

demo#: Demo simulation to be used as a base for cloned simulations.

id: str#

name: str#

demo: air_sdk.endpoints.simulations.Simulation#

created: datetime.datetime#

modified: datetime.datetime#

creator: str#

documentation: str | None#

tags: list[str]#

like_count: int#

liked_by_client: bool#

published: bool#

description: str | None#

repo: str | None#

icon: str | None#

classmethod get_model_api() → type[MarketplaceDemoEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: MarketplaceDemoEndpointAPI#

Update the marketplace demo’s properties.

Parameters:

name – New name for the marketplace demo
description – Description of the marketplace demo
documentation – Documentation of the marketplace demo
repo – Repository of the marketplace demo
tags – Tags of the marketplace demo
icon – Icon of the marketplace demo

Returns:

The updated MarketplaceDemo instance

Example

>>> marketplace_demo.update(name='New Name', description='New Desc')
>>> print(marketplace_demo.name)

publish(**kwargs: Any) → None[source]#

Publish the marketplace demo.

Example

>>> marketplace_demo.publish()

unpublish(**kwargs: Any) → None[source]#

Unpublish the marketplace demo.

Example

>>> marketplace_demo.unpublish()

provision(**kwargs: Any) → air_sdk.endpoints.simulations.Simulation[source]#

Provision a simulation from this marketplace demo.

Returns:: The newly created simulation instance.
Return type:: Simulation

Example

>>> simulation = marketplace_demo.provision()
>>> print(simulation.name)

class air_sdk.endpoints.MarketplaceDemoEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[MarketplaceDemo]

API client for marketplace demo endpoints.

API_PATH: str#

model: type[MarketplaceDemo]#

Create a new marketplace demo.

Parameters:

name – Name for the new marketplace demo
simulation – Simulation to be used to provision the marketplace demo
description – Description of the marketplace demo
documentation – Documentation of the marketplace demo
repo – Repository of the marketplace demo
tags – Tags of the marketplace demo
icon – Icon of the marketplace demo
checkpoint – A COMPLETE checkpoint to clone from. Provided checkpoint must belong to the simulation. If not specified, latest COMPLETE checkpoint will be used.

Returns:

The created MarketplaceDemo instance

Example

>>> marketplace_demo = api.marketplace_demos.create(
...     name='My Marketplace Demo',
...     simulation='sim-id',
...     description='My Demo Description',
...     documentation='My Demo Documentation',
...     repo='My Demo Repo',
...     tags=['networking', 'sonic'],
... )

delete(pk: air_sdk.air_model.PrimaryKey) → None#

Delete a marketplace demo.

Parameters:: pk – The marketplace demo ID (string or UUID)

Example

>>> api.marketplace_demos.delete('marketplace-demo-id')

list( *, creator: str | dataclasses._MISSING_TYPE = ..., published: bool | dataclasses._MISSING_TYPE = ..., tags: list[str] | dataclasses._MISSING_TYPE = ..., search: str | dataclasses._MISSING_TYPE = ..., ordering: str | dataclasses._MISSING_TYPE = ..., liked_by_client: bool | dataclasses._MISSING_TYPE = ..., limit: int | dataclasses._MISSING_TYPE = ..., offset: int | dataclasses._MISSING_TYPE = ..., ) → Iterator[MarketplaceDemo]#

List all marketplace demos.

Optional parameters:: creator: Filter by creator email published: Filter by published status tags: Filter by tags search: Search term to filter demos ordering: Order the response by the specified field liked_by_client: Filter by liked by client status limit: Number of results to return per page offset: The initial index from which to return the results

Returns:: Iterator of MarketplaceDemo instances

Example

>>> # List all demos
>>> for demo in api.marketplace_demos.list():
...     print(demo.name)
>>> # List with filters
>>> results = list(
...     api.marketplace_demos.list(
...         creator='test@example.com',
...         published=True,
...         tags=['networking'],
...     )
... )

get(pk: air_sdk.air_model.PrimaryKey) → MarketplaceDemo#

Get a specific marketplace demo by ID.

Parameters:: pk – The marketplace demo ID (string or UUID)
Returns:: The MarketplaceDemo instance

Example

>>> marketplace_demo = api.marketplace_demos.get('marketplace-demo-id')
>>> print(marketplace_demo.name)

publish(

*,

marketplace_demo: MarketplaceDemo | air_sdk.air_model.PrimaryKey,

**kwargs: Any,

) → None[source]#

Publish a marketplace demo.

Parameters:: marketplace_demo – The marketplace demo to publish (object or ID)
Returns:: None

Example

>>> # Using demo object
>>> api.marketplace_demos.publish(marketplace_demo=marketplace_demo)
>>> # Or using ID
>>> api.marketplace_demos.publish(marketplace_demo='marketplace-demo-id')

unpublish(

*,

marketplace_demo: MarketplaceDemo | air_sdk.air_model.PrimaryKey,

**kwargs: Any,

) → None[source]#

Unpublish a marketplace demo.

Parameters:: marketplace_demo – The marketplace demo to unpublish (object or ID)
Returns:: None

Example

>>> api.marketplace_demos.unpublish(marketplace_demo=marketplace_demo)

provision(

*,

marketplace_demo: MarketplaceDemo | air_sdk.air_model.PrimaryKey,

**kwargs: Any,

) → air_sdk.endpoints.simulations.Simulation[source]#

Provision a simulation from a marketplace demo.

Creates a new simulation by cloning the demo simulation.

Parameters:: marketplace_demo – The marketplace demo to provision (object or ID)
Returns:: The newly created simulation instance.
Return type:: Simulation

Example

>>> # Using demo object
>>> marketplace_demo = api.marketplace_demos.get('marketplace-demo-id')
>>> simulation = marketplace_demo.provision()
>>> print(simulation.id)
>>> # Or using API directly with ID
>>> demo_id = 'marketplace-demo-id'
>>> simulation = api.marketplace_demos.provision(marketplace_demo=demo_id)
>>> print(simulation.name)

class air_sdk.endpoints.NodeInstruction[source]#

Bases: air_sdk.air_model.AirModel

Node instruction model representing an automation instruction for a node.

Node instructions are commands or automation scripts that can be executed on simulation nodes. They support various executors and can be configured to run again on rebuild.

id#: Unique identifier for the node instruction

name#: Human-readable name of the node instruction

node#: Node this instruction belongs to

data#: Instruction data (ShellData, FileData, or InitData)

created#: Timestamp when the node instruction was created

modified#: Timestamp when the node instruction was last modified

run_again_on_rebuild#: Whether to re-run this instruction on simulation rebuild

state#: Execution state (e.g., ‘pending’, ‘running’, ‘completed’, ‘failed’)

id: str#

name: str#

node: air_sdk.endpoints.nodes.Node#

data: ShellDataResponse | FileDataResponse | InitDataResponse#

created: datetime.datetime#

modified: datetime.datetime#

state: str#

run_again_on_rebuild: bool#

classmethod get_model_api() → type[NodeInstructionEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: NodeInstructionEndpointAPI#

update( *, name: str | None = ..., run_again_on_rebuild: bool | None = ..., ) → None#

Update the node instruction’s properties.

Parameters:

name – New name for the node instruction
run_again_on_rebuild – Whether to re-run the instruction on simulation rebuild

Example

>>> instruction = api.node_instructions.get('instruction-id')
>>> instruction.update(name='New Name', run_again_on_rebuild=True)

delete() → None#

Delete this instruction.

Example

>>> instruction = api.node_instructions.get('instruction-id')
>>> instruction.delete()

class air_sdk.endpoints.NodeInstructionEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[NodeInstruction]

API client for node instructions endpoints.

API_PATH: str#

model: type[NodeInstruction]#

Create a new node instruction.

Parameters:

node – Node object or ID to execute the instruction on
name – Name for the instruction
executor – Type of executor (‘shell’, ‘init’, ‘file’)
data –
Instruction data/payload (executor-specific):

ShellData - For ‘shell’ executor
Contains: ‘commands’ (list of shell commands)

FileData - For ‘file’ executor
Contains: ‘files’ (list of dicts with ‘path’ and ‘content’) Optional: ‘post_commands’ (shell commands to run after creating files)

InitData - For ‘init’ executor
Contains: ‘hostname’ (hostname to set for the node)
run_again_on_rebuild – Optional whether to run instruction again on rebuild

Returns:

The created NodeInstruction instance

Example

>>> # Create shell instruction
>>> instruction = api.node_instructions.create(
...     name='Setup Script',
...     node='node-id',
...     executor='shell',
...     data=ShellData(commands=['#!/bin/bash\necho "Hello"']),
...     run_again_on_rebuild=True,
... )

>>> # Create file instruction
>>> instruction = api.node_instructions.create(
...     name='Install Package',
...     node='node-id',
...     executor='file',
...     data=FileData(
...         files=[{'path': 'package.txt', 'content': 'package=1.0.0'}],
...         post_commands=['#!/bin/bash\necho "Hello"'],
...     ),
...     run_again_on_rebuild=True,
... )

>>> # Create init instruction
>>> instruction = api.node_instructions.create(
...     name='Initialize Environment',
...     node='node-id',
...     executor='init',
...     data=InitData(hostname='my-node'),
...     run_again_on_rebuild=True,
... )

list( *, created_by_client: bool = ..., executor: Literal['shell', 'init', 'file'] = ..., name: str = ..., node: str = ..., run_again_on_rebuild: bool = ..., simulation: str = ..., state: str = ..., search: str = ..., ordering: str = ..., limit: int = ..., offset: int = ..., ) → Iterator[NodeInstruction]#

List all node instructions.

Parameters:

created_by_client – Filter by created by client
executor – Filter by executor (‘shell’, ‘init’, ‘file’)
name – Filter by name
node – Filter by node
run_again_on_rebuild – Filter by run_again_on_rebuild
simulation – Filter by simulation
state – Filter by state
search – Search by name
ordering – Order by field
limit – Limit the number of results
offset – Offset the results

Returns:

Iterator of NodeInstruction instances

Example

>>> # List all instructions for a node
>>> for instruction in api.node_instructions.list(node='node-id'):
...     print(instruction.name)

>>> # List with multiple filters
>>> for instruction in api.node_instructions.list(
...     simulation='sim-id',
...     executor='shell',
...     state='complete',
...     ordering='-created',
... ):
...     print(instruction.name, instruction.state)

get(pk: air_sdk.air_model.PrimaryKey) → NodeInstruction#

Get a specific node instruction by ID.

Parameters:: pk – The node instruction ID (string or UUID)
Returns:: The NodeInstruction instance

Example

>>> instruction = api.node_instructions.get('instruction-id')
>>> print(instruction.name)

patch( *, pk: air_sdk.air_model.PrimaryKey, name: str | None | dataclasses._MISSING_TYPE = ..., run_again_on_rebuild: bool | None | dataclasses._MISSING_TYPE = ..., ) → NodeInstruction#

Partially update a node instruction.

Parameters:

pk – The node instruction ID
name – New name for the instruction
run_again_on_rebuild – Whether to run again on rebuild

Returns:

The updated NodeInstruction instance

Example

>>> instruction = api.node_instructions.patch(
...     'instruction-id',
...     name='Updated Name',
...     run_again_on_rebuild=True
... )

delete(pk: air_sdk.air_model.PrimaryKey) → None#

Delete a node instruction.

Parameters:: pk – The node instruction ID

Example

>>> api.node_instructions.delete('instruction-id')

class air_sdk.endpoints.Node[source]#

Bases: air_sdk.air_model.AirModel

Node model representing a network node.

id#: Unique identifier for the node

name#: Human-readable name of the node

created#: Timestamp when the node was created

modified#: Timestamp when the node was last modified

state#: Current state of the node

category#: Category of the node

status_from_worker#: Status of the node from the worker

split_options#: Split options for the node

cpu#: CPU of the node

memory#: Memory of the node

storage#: Storage of the node

pos_x#: X position of the node

pos_y#: Y position of the node

attributes#: Attributes of the node

advanced#: Advanced attributes of the node

cdrom#: CDROM attributes of the node

storage_pci#: Storage PCI attributes of the node

cloud_init#: Cloud-Init assignments of the node

id: str#

created: datetime.datetime#

modified: datetime.datetime#

name: str#

simulation: air_sdk.endpoints.simulations.Simulation#

image: air_sdk.endpoints.images.Image#

state: str#

category: str#

status_from_worker: str#

split_options: list[int] | None#

cpu: int#

memory: int#

storage: int#

pos_x: int#

pos_y: int#

attributes: NodeAttributes | None#

advanced: NodeAdvanced#

cdrom: NodeCDROM | None#

storage_pci: dict[str, StoragePCIField] | None#

classmethod get_model_api() → type[NodeEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: NodeEndpointAPI#

update( *, name: str | dataclasses._MISSING_TYPE = ..., image: air_sdk.endpoints.images.Image | dataclasses._MISSING_TYPE = ..., cpu: int | dataclasses._MISSING_TYPE = ..., memory: int | dataclasses._MISSING_TYPE = ..., storage: int | dataclasses._MISSING_TYPE = ..., pos_x: int | dataclasses._MISSING_TYPE = ..., pos_y: int | dataclasses._MISSING_TYPE = ..., attributes: dict[str, Any] | dataclasses._MISSING_TYPE = ..., advanced: dict[str, Any] | dataclasses._MISSING_TYPE = ..., storage_pci: dict[str, Any] | dataclasses._MISSING_TYPE = ..., ) → None#

Update the node’s properties.

Parameters:

name – Name of the node
image – Image of the node
cpu – CPU of the node
memory – Memory of the node
storage – Storage of the node
pos_x – X position of the node
pos_y – Y position of the node
attributes – Attributes of the node
advanced – Advanced attributes of the node
storage_pci – Storage PCI of the node

Example

>>> node = api.nodes.update(
...     name='my-node',
...     image=image,
...     cpu=1,
...     memory=1024,
...     storage=10,
...     pos_x=0,
...     pos_y=0,
...     attributes={'key': 'value'},
...     advanced={'key': 'value'},
...     storage_pci={'key': 'value'},
... )

clear_cloud_init_assignment(**kwargs: Any) → None#

Clear cloud-init assignment for this node (V3).

Removes any user_data and meta_data cloud-init configurations from this node.

Example

>>> node.clear_cloud_init_assignment()

property instructions: Any#

Access the instructions endpoint for this node.

Returns:: Any instance scoped to this node

Example

>>> node = api.nodes.instructions.list()

property interfaces: air_sdk.endpoints.interfaces.InterfaceEndpointAPI#

Access the interfaces endpoint for this node.

Returns:: Any instance scoped to this node

Example

>>> for interface in node.interfaces.list():
...     print(interface.name)

property services: air_sdk.endpoints.services.ServiceEndpointAPI#

Query for the related services of the node.

Returns:: ServiceEndpointAPI instance filtered for this node’s services

Example

>>> for service in node.services.list():
...     print(f'{service.name}: {service.worker_fqdn}:{service.worker_port}')
>>>
>>> # Create service by interface ID
>>> service = node.services.create(
...     name='ssh-service',
...     interface='interface-id',
...     node_port=22,
...     service_type='SSH'
... )

create_service( *, interface_name: str, node_port: int, name: str = ..., service_type: Literal['SSH', 'HTTPS', 'HTTP', 'OTHER'] = ..., ) → air_sdk.endpoints.services.Service[source]#

Create service by resolving interface name (v3 convenience).

Parameters:

interface_name – Interface name on this node (e.g., ‘eth0’, ‘swp1’)
node_port – Port number on the node/interface
name – Service name
service_type – Service type - ‘SSH’, ‘HTTPS’, ‘HTTP’, or ‘OTHER’

Returns:

Service object

Raises:

ValueError – If interface not found on this node

Example

>>> service = node.create_service(
...     interface_name='eth0',
...     name='ssh-service',
...     node_port=22,
...     service_type='SSH',
... )

delete_all_node_instructions() → None[source]#

Delete all node instructions for this node.

Example

>>> node = api.nodes.delete_all_node_instructions()

property cloud_init: air_sdk.bc.cloud_init.CloudInit#

Get the cloud-init assignment for this node.

Returns:: CloudInit object containing user_data and meta_data assignments

Example

>>> cloud_init = node.cloud_init
>>> print(cloud_init.user_data)
>>> print(cloud_init.meta_data)

reset() → None[source]#

Reset this node.

Resetting the node emulates the hardware reset button on physical machines where the machine is immediately restarted without a clean shutdown of the operating system. For nodes that are not currently running, this means simply booting them back up.

Example

>>> node.reset()

rebuild() → None[source]#

Rebuild this node.

Rebuilding a node means returning the node to the state of its simulation’s current checkpoint. When rebuilding from the initial state, all repeatable instructions for this node will be applied. All existing instructions created for this node which have not yet been completed will be failed. All existing instructions created for this node which have not yet been delivered will be cancelled.

Example

>>> node.rebuild()

class air_sdk.endpoints.NodeEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[Node]

API client for simulation endpoints.

API_PATH: str#

model: type[Node]#

create( *, name: str, simulation: air_sdk.endpoints.simulations.Simulation, image: air_sdk.endpoints.images.Image, cpu: int | None = ..., memory: int | None = ..., storage: int | None = ..., pos_x: int | None = ..., pos_y: int | None = ..., attributes: NodeAttributes | None = ..., advanced: NodeAdvanced | None = ..., storage_pci: dict[str, StoragePCIField] | None = ..., ) → Node#

Create a new node.

Parameters:

name – Name of the node
simulation – Simulation of the node
image – Image of the node
cpu – (optional) CPU of the node
memory – (optional) Memory of the node
storage – (optional) Storage of the node
pos_x – (optional) X position of the node
pos_y – (optional) Y position of the node
attributes – (optional) Attributes of the node
advanced – (optional) Advanced attributes of the node
storage_pci – (optional) Storage PCI of the node

Returns:

The created node instance

Example

>>> node = api.nodes.create(simulation=sim, image=image, name='my-node')

list(

*,

limit: int | None = ...,

offset: int | None = ...,

ordering: str | None = ...,

search: str | None = ...,

**kwargs: Any,

) → Iterator[Node]#

List all nodes.

Parameters:

limit – (optional) Limit the number of nodes returned
offset – (optional) Offset the number of nodes returned
ordering – (optional) Order the nodes by a field
search – (optional) Search for nodes by a keyword

Returns:

Iterator of Node instances

Example

>>> for node in api.nodes.list(ordering='name'):
...     print(node.name)

>>> for node in api.nodes.list(search='my-node'):
...     print(node.name)

>>> for node in api.nodes.list(ordering='-name'):
...     print(node.name)

get(pk: air_sdk.air_model.PrimaryKey) → Node#

Get a specific node by ID.

Parameters:: pk – The node ID (string or UUID)
Returns:: The Node instance

Example

>>> node = api.nodes.get('node-id')

delete(pk: air_sdk.air_model.PrimaryKey) → None#

Delete a specific node by ID.

Parameters:: pk – The node ID (string or UUID)

Example

>>> api.nodes.delete('node-id')

update( *, node: Node | air_sdk.air_model.PrimaryKey, name: str | dataclasses._MISSING_TYPE = ..., image: air_sdk.endpoints.images.Image | dataclasses._MISSING_TYPE = ..., cpu: int | dataclasses._MISSING_TYPE = ..., memory: int | dataclasses._MISSING_TYPE = ..., storage: int | dataclasses._MISSING_TYPE = ..., pos_x: int | dataclasses._MISSING_TYPE = ..., pos_y: int | dataclasses._MISSING_TYPE = ..., attributes: NodeAttributes | dataclasses._MISSING_TYPE = ..., advanced: NodeAdvanced | dataclasses._MISSING_TYPE = ..., storage_pci: dict[str, StoragePCIField] | None = ..., ) → Node[source]#

Update a specific node by ID.

Parameters:

node – The node to update (Node object or ID)
name – Name of the node
image – Image of the node (image object or ID)
cpu – CPU of the node
memory – Memory of the node
storage – Storage of the node
pos_x – X position of the node
pos_y – Y position of the node
attributes – Attributes of the node
advanced – Advanced attributes of the node
storage_pci – Storage PCI of the node

Example

>>> node = api.nodes.update(node=node, name='my-node')

create_from_system_node(

system_node: air_sdk.endpoints.systems.System | air_sdk.air_model.PrimaryKey,

name: str,

simulation: air_sdk.endpoints.simulations.Simulation | air_sdk.air_model.PrimaryKey,

image: air_sdk.endpoints.images.Image | air_sdk.air_model.PrimaryKey | dataclasses._MISSING_TYPE = ...,

cpu: int | dataclasses._MISSING_TYPE = ...,

memory: int | dataclasses._MISSING_TYPE = ...,

storage: int | dataclasses._MISSING_TYPE = ...,

pos_x: int | dataclasses._MISSING_TYPE = ...,

pos_y: int | dataclasses._MISSING_TYPE = ...,

attributes: dict[str, Any] | dataclasses._MISSING_TYPE = ...,

advanced: dict[str, Any] | dataclasses._MISSING_TYPE = ...,

storage_pci: dict[str, StoragePCIField] | dataclasses._MISSING_TYPE = ...,

**kwargs: Any,

) → Node[source]#

Create a node from a system node template.

Parameters:

system_node – System node template ID to create from
name – Name of the new node
simulation – Simulation object or ID where the node will be created
image – Optional image to use (overrides template)
cpu – Optional CPU count (overrides template)
memory – Optional memory in MB (overrides template)
storage – Optional storage in GB (overrides template)
pos_x – Optional X position on canvas
pos_y – Optional Y position on canvas
attributes – Optional node attributes
advanced – Optional advanced configuration
storage_pci – Optional storage PCI configuration
**kwargs – Additional parameters

Returns:

The created Node object

Example

>>> node = api.nodes.from_system_node(
...     system_node='system-node-template-id',
...     name='my-node',
...     simulation='simulation-id'
... )

list_system_nodes(

**kwargs: Any,

) → list[air_sdk.endpoints.systems.System][source]#

List all available system nodes.

Parameters:

limit – (optional) Limit the number of system nodes returned
offset – (optional) Offset the number of system nodes returned
ordering – (optional) Order the system nodes by a field
search – (optional) Search for system nodes by a keyword
simulation – (optional) Filter system nodes by simulation

Returns:

List of System objects that can be used to create nodes.

reset(*, node: Node | air_sdk.air_model.PrimaryKey) → None[source]#

Reset a node.

Parameters:: node – The node object or node ID to reset

Example

>>> # Reset using node object
>>> api.nodes.reset(node=node)

>>> # Reset using node ID
>>> api.nodes.reset(node='node-uuid-123')

rebuild(*, node: Node | air_sdk.air_model.PrimaryKey) → None[source]#

Rebuild a node.

Rebuilding a node means returning the node to the state of its simulation’s current checkpoint. When rebuilding from the initial state, all repeatable instructions for the node will be applied. All existing instructions created for the node which have not yet been completed will be failed. All existing instructions created for the node which have not yet been delivered will be cancelled.

Parameters:: node – The node object or node ID to rebuild

Example

>>> # Rebuild using node object
>>> api.nodes.rebuild(node=node)

>>> # Rebuild using node ID
>>> api.nodes.rebuild(node='node-uuid-123')

class air_sdk.endpoints.Organization[source]#

Bases: air_sdk.air_model.AirModel

Represents an organization and its resource budget in the Air platform.

This model combines organization metadata with resource budget information.

id#: Unique identifier for the resource budget

created#: Timestamp when the resource budget was created

modified#: Timestamp when the resource budget was last modified

org_display_name#: Display name of the organization

org_nca_id#: NCA ID of the organization

cpu#: Number of CPU cores allocated

memory#: Memory allocated, in MiB

disk_storage_total#: Total disk storage allocated, in GB

disk_storage_per_node#: Disk storage per node, in GB

image_storage#: Image storage allocated, in GB

userconfigs#: Total UserConfig content allocated, in bytes

usage#: Current resource usage

id: str#

created: datetime.datetime#

modified: datetime.datetime#

org_display_name: str#

org_nca_id: str#

cpu: int | float#

memory: int | float#

disk_storage_total: int | float#

disk_storage_per_node: int#

image_storage: int#

userconfigs: int#

usage: air_sdk.types.ResourceBudgetUsage#

property name: str#: Organization name (alias for org_display_name).

classmethod get_model_api() → type[OrganizationEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: OrganizationEndpointAPI#

class air_sdk.endpoints.OrganizationEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[Organization]

API client for Organization / ResourceBudget endpoints.

This endpoint is read-only and provides access to organization resource budget information.

Also aliased as ResourceBudgetEndpointAPI.

API_PATH: str#

model: type[Organization]#

list( *, limit: int = ..., offset: int = ..., ordering: str = ..., org_display_name: str = ..., org_nca_id: str = ..., search: str = ..., ) → Iterator[Organization]#

List all organizations / resource budgets with optional filtering.

Parameters:

limit – Maximum number of results to return per page
offset – The initial index from which to return the results
ordering – Order by field. Prefix with “-” for descending order
org_display_name – Filter by the display name of the organization
org_nca_id – Filter by the NCA ID of the organization
search – Search resource budgets by org_display_name

Returns:

Iterator of Organization instances

Example

>>> # List all organizations
>>> for org in api.organizations.list():
...     print(org.name, org.org_nca_id)

>>> # Filter by org_nca_id
>>> for org in api.organizations.list(org_nca_id='nca-123'):
...     print(org.name)

>>> # Search by name
>>> for org in api.organizations.list(search='NVIDIA'):
...     print(org.name)

get(pk: air_sdk.air_model.PrimaryKey) → Organization#

Get a specific organization / resource budget by ID.

Parameters:: pk – The resource budget ID (string or UUID)
Returns:: The Organization instance

Example

>>> org = api.organizations.get('b4d0480c-6f0b-4c40-b143-c141531fc14e')
>>> print(org.name, org.cpu, org.memory)

class air_sdk.endpoints.Service[source]#

Bases: air_sdk.bc.ServiceCompatMixin, air_sdk.bc.BaseCompatMixin, air_sdk.air_model.AirModel

Represents a service in the Air API.

A service exposes a port on a simulation interface to external networks, enabling connectivity between simulations and the outside world.

id#: Unique identifier

name#: Service name

node_port#: Port number on the node/interface

interface#: Interface object (foreign key relationship)

service_type#: Type of service (e.g., ‘ssh’, ‘http’, ‘https’)

worker_port#: External port on the worker (assigned by Air)

worker_fqdn#: Fully qualified domain name of the worker

created#: Timestamp when service was created

modified#: Timestamp when service was last modified

Example

>>> # Access service details
>>> print(f'Service: {service.name}')
>>> print(f'Connect via: {service.worker_fqdn}:{service.worker_port}')
>>> print(f'Interface: {service.interface.name}')
>>>
>>> # Delete service
>>> service.delete()

id: str#

name: str#

node_port: int#

interface: air_sdk.endpoints.interfaces.Interface#

service_type: str#

worker_port: int | None#

worker_fqdn: str | None#

created: datetime.datetime#

modified: datetime.datetime#

classmethod get_model_api() → type[ServiceEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

delete() → None#

Delete the service.

After deletion, the service object should not be used.

Example

>>> service.delete()

refresh() → None#

Refresh service data from the API.

Example

>>> service.refresh()

class air_sdk.endpoints.ServiceEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.bc.ServiceEndpointAPICompatMixin, air_sdk.endpoints.mixins.ListApiMixin[Service], air_sdk.endpoints.mixins.CreateApiMixin[Service], air_sdk.endpoints.mixins.GetApiMixin[Service], air_sdk.endpoints.mixins.DeleteApiMixin, air_sdk.air_model.BaseEndpointAPI[Service]

API for managing services.

Services expose ports on simulation interfaces to enable external connectivity.

Example

>>> # List all services
>>> for service in api.services.list():
...     print(f'{service.name}: {service.worker_fqdn}:{service.worker_port}')
>>>
>>> # Create a service
>>> interface = api.interfaces.get('interface-id')
>>> service = api.services.create(
...     name='SSH Service',
...     node_port=22,
...     interface=interface,
...     service_type='ssh',
... )
>>>
>>> # Get a service
>>> service = api.services.get('service-id')
>>>
>>> # Delete a service
>>> api.services.delete(service)

API_PATH: str#

model: type[Service]#

List services with optional filtering and pagination.

Parameters:

interface – Filter by interface ID or instance
name – Filter by service name
node_port – Filter by node port
worker_port – Filter by worker port
worker_fqdn – Filter by worker FQDN
service_type – Filter by service type (ssh, http, https, etc.)
search – Search term to filter results
ordering – Field to order results by (prefix with ‘-’ for descending)
limit – Maximum number of results to return
offset – Number of results to skip

Yields:

Service instances

Example

>>> # List all services
>>> for service in api.services.list():
...     print(service.name)
>>>
>>> # Filter by interface
>>> for service in api.services.list(interface='interface-id'):
...     print(f"{service.name}: {service.worker_port}")
>>>
>>> # Filter by service type
>>> for service in api.services.list(service_type='ssh'):
...     print(service.name)

get(pk: air_sdk.air_model.PrimaryKey) → Service#

Get a service by ID.

Parameters:: pk – Service ID
Returns:: Service instance
Raises:: AirUnexpectedResponse – Service not found or API error

Example

>>> service = api.services.get('3dadd54d-583c-432e-9383-a2b0b1d7f551')
>>> print(f'{service.name}: {service.worker_fqdn}:{service.worker_port}')

create( *, name: str, node_port: int, interface: air_sdk.air_model.PrimaryKey, service_type: Literal['SSH', 'HTTPS', 'HTTP', 'OTHER'] = ..., ) → Service#

Create a new service.

Parameters:

name – Service name
node_port – Port number on the node/interface
interface – Interface instance or ID
service_type – Service type - ‘SSH’, ‘HTTPS’, ‘HTTP’, or ‘OTHER’

Returns:

Created Service instance

Raises:

AirUnexpectedResponse – Creation failed

Example

>>> service = api.services.create(
...     name='SSH Access',
...     node_port=22,
...     interface='interface-id',
...     service_type='ssh',
... )

delete(*, service: Service | air_sdk.air_model.PrimaryKey) → None#

Delete a service.

Parameters:: service – Service instance or ID

Example

>>> api.services.delete(service='service-id')
>>> # Or using instance
>>> api.services.delete(service=service)

class air_sdk.endpoints.Simulation[source]#

Bases: air_sdk.air_model.AirModel

Simulation model representing a network simulation.

The string representation shows: id, name, state, creator

id#: Unique identifier for the simulation

name#: Human-readable name of the simulation

created#: Timestamp when the simulation was created

modified#: Timestamp when the simulation was last modified

state#: Current state of the simulation (see Literal values for all states)

creator#: Email of the user who created the simulation

auto_oob_enabled#: Whether automatic out-of-band management is enabled

disable_auto_oob_dhcp#: Whether DHCP should be disabled on the OOB server

auto_netq_enabled#: Whether automatic NetQ is enabled

sleep_at#: When the simulation should be automatically put to sleep (stored)

expires_at#: When the simulation should be automatically deleted

documentation#: Documentation markdown or URL to documentation markdown

complete_checkpoint_count#: Number of complete checkpoints in the simulation

id: str#

name: str#

created: datetime.datetime#

modified: datetime.datetime#

state: Literal['CLONING', 'CREATING', 'IMPORTING', 'INVALID', 'INACTIVE', 'REQUESTING', 'PROVISIONING', 'PREPARE_BOOT', 'BOOTING', 'ACTIVE', 'PREPARE_SHUTDOWN', 'SHUTTING_DOWN', 'SAVING', 'PREPARE_REBUILD', 'REBUILDING', 'DELETING', 'PREPARE_PURGE', 'PURGING', 'DEMO', 'TRAINING']#

creator: str#

auto_oob_enabled: bool | None#

disable_auto_oob_dhcp: bool | None#

auto_netq_enabled: bool | None#

sleep_at: datetime.datetime | None#

expires_at: datetime.datetime | None#

documentation: str | None#

complete_checkpoint_count: int#

classmethod get_model_api() → type[SimulationEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: SimulationEndpointAPI#

Update the simulation’s properties.

Note: For OOB, DHCP and NetQ configuration, use dedicated methods like enable_auto_oob(), disable_auto_oob(), enable_auto_netq(), disable_auto_netq(), etc.

Parameters:

name – New name for the simulation
sleep_at – When the simulation should be automatically put to sleep
expires_at – When the simulation should be automatically deleted
documentation – Documentation markdown or URL to documentation markdown

Example

>>> simulation.update(name='New Name', documentation='https://docs.example.com')

enable_auto_oob(disable_auto_oob_dhcp: bool = ...) → None[source]#

Enable automatic Out-of-band management for this simulation.

Parameters:: disable_auto_oob_dhcp – If True, disable DHCP on the OOB management network

Example

# Enable OOB with DHCP: >>> simulation.enable_auto_oob()

# Enable OOB without DHCP: >>> simulation.enable_auto_oob(disable_auto_oob_dhcp=True)

disable_auto_oob() → None[source]#

Disable automatic Out-of-band management for this simulation.

Example

>>> simulation.disable_auto_oob()

enable_auto_netq() → None[source]#

Enable automatic NetQ for this simulation.

Example

>>> simulation.enable_auto_netq()

disable_auto_netq() → None[source]#

Disable automatic NetQ for this simulation.

Example

>>> simulation.disable_auto_netq()

start(*, checkpoint: str | None = ...) → None[source]#

Start the simulation.

Parameters:: checkpoint – Optional checkpoint ID to start from. If not specified, the API will use its default behavior (typically uses the most recent checkpoint if available). If explicitly set to None, starts from clean state (rebuild). If a string, starts from the specified checkpoint.

Example

# Normal start (API determines behavior): >>> simulation.start()

# Start from specific checkpoint: >>> simulation.start(checkpoint=’checkpoint-id’)

# Rebuild (start from clean state, no checkpoint): >>> simulation.start(checkpoint=None)

shutdown(*, create_checkpoint: bool = ...) → None[source]#

Shut down the simulation.

Parameters:: create_checkpoint – Whether to create a checkpoint before shutting down. If not specified, the API will use its default behavior.

Example

# Normal shutdown (API determines behavior): >>> simulation.shutdown()

# Explicitly create checkpoint before shutdown: >>> simulation.shutdown(create_checkpoint=True)

# Explicitly don’t create checkpoint: >>> simulation.shutdown(create_checkpoint=False)

rebuild(*, checkpoint: str | None = ...) → None[source]#

Rebuild the simulation from a given checkpoint.

Tears down the simulation and starts it from the given checkpoint. If no checkpoint is provided, the simulation will be rebuilt from the clean state.

Parameters:: checkpoint – Optional checkpoint ID to rebuild from. If not specified, the API will use its default behavior (the current checkpoint the simulation is running off of). If explicitly set to None, starts from clean state (rebuild). If a string, starts from the specified checkpoint.

Example

# Rebuild simulation from specific checkpoint: >>> simulation.rebuild(checkpoint=’checkpoint-id’)

Wait for simulation to reach one of the target states.

Parameters:

target_states – Single state or list of states to wait for
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

Raises:

ValueError – If the simulation enters one of the error states
TimeoutError – If timeout is reached before target state

Example

>>> simulation.wait_for_state('ACTIVE', error_states=['INVALID'])

>>> # Wait for multiple possible states
>>> simulation.wait_for_state(['INACTIVE', 'ACTIVE'])

>>> # Custom timeout
>>> simulation.wait_for_state(
...     'ACTIVE',
...     timeout=timedelta(minutes=5),
...     error_states=['INVALID', 'DELETING']
... )

set_sleep_time(sleep_at: datetime.datetime | None) → None[source]#

Set when the simulation should be automatically put to sleep (stored).

Accepts any timezone-aware datetime, which will be automatically converted to UTC. Naive datetimes (without timezone) will trigger a warning and assume local timezone.

Parameters:: sleep_at – Timezone-aware datetime when simulation should sleep, or None to clear. Naive datetimes will assume local timezone and emit a warning.

Example

>>> from datetime import datetime, timedelta, timezone

# Timezone-aware datetime (recommended): >>> sleep_time = datetime.now(timezone.utc) + timedelta(hours=1) >>> simulation.set_sleep_time(sleep_time)

# Naive datetime (triggers warning): >>> now = datetime.now() >>> simulation.set_sleep_time(now + timedelta(hours=1))

# Clear sleep time: >>> simulation.set_sleep_time(None)

set_expire_time(expires_at: datetime.datetime | None) → None[source]#

Set when the simulation should be automatically deleted.

Accepts any timezone-aware datetime, which will be automatically converted to UTC. Naive datetimes (without timezone) will trigger a warning and assume local timezone.

Parameters:: expires_at – Timezone-aware datetime when simulation should expire, or None to clear. Naive datetimes will assume local timezone and emit a warning.

Example

>>> from datetime import datetime, timedelta, timezone

# Timezone-aware datetime (recommended): >>> expire_time = datetime.now(timezone.utc) + timedelta(hours=2) >>> simulation.set_expire_time(expire_time)

# Naive datetime (triggers warning): >>> now = datetime.now() >>> simulation.set_expire_time(now + timedelta(hours=2))

# Clear expiration time: >>> simulation.set_expire_time(None)

create_ztp_script( *, content: str, ) → air_sdk.endpoints.ztp_scripts.ZTPScript[source]#

Create a ZTP (Zero Touch Provisioning) script for the simulation.

Parameters:: content – The content of the ZTP script
Returns:: The created ZTPScript instance

Example

>>> script = simulation.create_ztp_script(
...     content='#!/bin/bash\n#CUMULUS-AUTOPROVISIONING\necho "Hello World"'
... )

update_ztp_script( *, content: str, ) → air_sdk.endpoints.ztp_scripts.ZTPScript[source]#

Update the ZTP script for this simulation.

Parameters:: content – The new script content
Returns:: The updated ZTPScript instance

Example

>>> updated_script = simulation.update_ztp_script(
...     content='#!/bin/bash\n#CUMULUS-AUTOPROVISIONING\necho "Updated"'
... )

delete_ztp_script() → None[source]#

Delete the ZTP script for this simulation.

After deletion, simulation.ztp_script will return None.

Example

>>> simulation.delete_ztp_script()
>>> print(simulation.ztp_script)
None

export( *, image_ids: bool = ..., topology_format: Literal['JSON'] = 'JSON', ) → dict[str, Any][source]#

Export the simulation.

Parameters:

image_ids – Whether to include image IDs in the export. If not specified, the API will use its default behavior.
topology_format – Format for the topology in the export. If not specified, the API will use its default behavior.

Returns:

Dictionary containing the exported simulation data

Example

# Export with API defaults: >>> export_data = simulation.export()

# Export with specific options: >>> export_data = simulation.export(image_ids=True, topology_format=’JSON’)

clone( *, checkpoint: str | None = ..., attempt_start: bool = ..., ) → Simulation[source]#

Clone/duplicate the simulation.

Parameters:

checkpoint – Optional checkpoint ID to clone from. If not specified, Air will find the most recent COMPLETE checkpoint if it exists.
attempt_start – If the simulation should start immediately after cloning

Returns:

The cloned Simulation instance

Example

# Simple clone: >>> cloned_sim = simulation.clone()

# Clone with specific checkpoint: >>> cloned_sim = simulation.clone(checkpoint=’checkpoint-id’)

# Clone and start immediately: >>> cloned_sim = simulation.clone(attempt_start=True)

property ztp_script: air_sdk.endpoints.ztp_scripts.ZTPScript | None#

Get the simulation’s ZTP script if it exists.

Returns:: The ZTPScript instance or None if no script exists

Example

>>> if script := simulation.ztp_script:
...     print(script.content)

get_history( *, category: str = ..., actor: str = ..., search: str = ..., ordering: Literal['actor', 'category', 'created', 'model', 'object_id', 'description'] = ..., ) → Iterator[air_sdk.endpoints.history.History][source]#

Get the historical entries for the simulation.

Parameters:

category – Filter by category of the history entries
actor – Filter by actor who performed the actions
search – Search term to filter the history entries
ordering – Order the response by the specified field

Returns:

Iterator of History objects for the simulation

Example

# Basic usage with ordering: >>> for history in simulation.get_history(ordering=’created’): … print(history.description)

# Search and filter: >>> for history in simulation.get_history(search=’OOB’, ordering=’category’): … print(history.description)

property nodes: air_sdk.endpoints.nodes.NodeEndpointAPI#

Query for the related nodes of the simulation.

Returns:: NodeEndpointAPI instance filtered for this simulation’s nodes

Example

>>> for node in simulation.nodes.list():
...     print(node.name)

property interfaces: air_sdk.endpoints.interfaces.InterfaceEndpointAPI#

Query for the related interfaces of the simulation.

Returns:: InterfaceEndpointAPI instance filtered for this simulation’s interfaces

Example

>>> for interface in simulation.interfaces.list():
...     print(interface.name)

property node_instructions: air_sdk.endpoints.node_instructions.NodeInstructionEndpointAPI#

Query for the related node instructions of the simulation.

Returns:: NodeInstructionEndpointAPI filtered for this simulation’s instructions

Example

>>> for instruction in simulation.node_instructions.list():
...     print(instruction.name, instruction.state)

property services: air_sdk.endpoints.services.ServiceEndpointAPI#

Query for the related services of the simulation.

Returns:: ServiceEndpointAPI instance filtered for this simulation’s services

Example

>>> for service in simulation.services.list():
...     print(f'{service.name}: {service.worker_fqdn}:{service.worker_port}')
>>>
>>> # Create service using 'node:interface' string (BC)
>>> service = simulation.services.create(
...     name='SSH', interface='server-1:eth0', dest_port=22
... )

create_service( *, node_name: str, interface_name: str, node_port: int, name: str = ..., service_type: Literal['SSH', 'HTTPS', 'HTTP', 'OTHER'] = ..., ) → air_sdk.endpoints.services.Service[source]#

Create service using node and interface names.

Parameters:

node_name – Node name in this simulation
interface_name – Interface name on the node (e.g., ‘eth0’, ‘swp1’)
node_port – Port number on the node
name – Service name (optional)
service_type – Service type - ‘SSH’, ‘HTTPS’, ‘HTTP’, or ‘OTHER’

Returns:

Service object

Raises:

ValueError – If node or interface not found in simulation

Example

>>> service = sim.create_service(
...     node_name='server-1',
...     interface_name='eth0',
...     name='SSH Access',
...     node_port=22,
...     service_type='SSH'
... )

node_bulk_assign( *, nodes: List[air_sdk.types.NodeAssignmentDataV3], ) → None[source]#

Bulk assign configurations to nodes in this simulation.

Parameters:: nodes – List of node assignment data containing node, user_data, and meta_data

Example

>>> # Bulk assign cloud-init configs to multiple nodes
>>> simulation.node_bulk_assign(
...     nodes=[
...         {'node': 'node1', 'user_data': 'config1'},
...         {'node': 'node2', 'meta_data': 'config2'},
...     ],
... )

node_bulk_reset( *, nodes: List[air_sdk.types.NodeResetPayload], ) → None[source]#

Reset specific nodes within this simulation.

Parameters:: nodes – List of node reset payloads, each containing a node object or ID

Example

>>> # Reset a single node
>>> simulation.node_bulk_reset(nodes=[{'id': node.id}])

>>> # Reset multiple nodes using node IDs
>>> simulation.node_bulk_reset(
...     nodes=[
...         {'id': 'node-uuid-1'},
...         {'id': 'node-uuid-2'},
...     ],
... )

>>> # Reset all nodes in simulation
>>> nodes_to_reset = [{'id': n} for n in simulation.nodes.list()]
>>> simulation.node_bulk_reset(nodes=nodes_to_reset)

node_bulk_rebuild( *, nodes: List[air_sdk.types.NodeRebuildPayload], checkpoint: str | None = ..., ) → None[source]#

Rebuild specific nodes within this simulation.

Rebuilding a node means returning the node to either the state of the current checkpoint of its simulation or its initial, first boot state. When rebuilding from the initial state, all repeatable instructions for selected nodes will be applied. All existing instructions created for the selected nodes which have not yet been completed will be failed. All existing instructions created for the selected nodes which have not yet been delivered will be cancelled.

Parameters:

nodes – List of node rebuild payloads, each containing a node object or ID
checkpoint – Optional checkpoint ID to rebuild from

Example

>>> # Rebuild a single node
>>> simulation.node_bulk_rebuild(nodes=[{'id': node.id}])

>>> # Rebuild multiple nodes using node IDs
>>> simulation.node_bulk_rebuild(
...     nodes=[
...         {'id': 'node-uuid-1'},
...         {'id': 'node-uuid-2'},
...     ],
... )

>>> # Rebuild all nodes in simulation
>>> nodes_to_rebuild = [{'id': n} for n in simulation.nodes.list()]
>>> simulation.node_bulk_rebuild(nodes=nodes_to_rebuild)

class air_sdk.endpoints.SimulationEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[Simulation]

API client for simulation endpoints.

API_PATH: str#

model: type[Simulation]#

create( *, name: str, sleep_at: datetime.datetime | None = ..., expires_at: datetime.datetime | None = ..., documentation: str | None = ..., ) → Simulation#

Create a blank simulation.

Parameters:

name – Name for the new simulation
sleep_at – When the simulation should be automatically put to sleep
expires_at – When the simulation should be automatically deleted
documentation – Documentation/description for the simulation

Returns:

The created Simulation instance

Example

# Simple creation: >>> simulation = api.simulations.create(name=’My Simulation’)

# With expiration and documentation: >>> from datetime import datetime, timedelta, timezone >>> expires = datetime.now(timezone.utc) + timedelta(days=7) >>> simulation = api.simulations.create( … name=’My Simulation’, … expires_at=expires, … documentation=’Test simulation’ … )

import_from_data( *, format: str, content: dict[str, Any] | str, name: str, ztp: str | None = ..., attempt_start: bool = ..., start_timeout: datetime.timedelta | None = ..., ) → Simulation[source]#

Import a simulation from raw data.

Parameters:

format – Format of the content (‘JSON’ or ‘DOT’)
content – The topology content (dict for JSON, str for DOT)
name – Name for the new simulation
ztp – Optional ZTP script content
attempt_start – When enabled, waits for the simulation creation to complete and then starts it automatically
start_timeout – Maximum time to wait for simulation creation (default: 120 seconds)

Returns:

The imported Simulation instance

Example

# Import from JSON: >>> simulation = api.simulations.import_from_data( … format=’JSON’, content={‘nodes’: […]}, name=’My Sim’ … )

# Import and start: >>> simulation = api.simulations.import_from_data( … format=’JSON’, … content={‘nodes’: […]}, … name=’My Sim’, … attempt_start=True … )

# Import and start with custom timeout: >>> simulation = api.simulations.import_from_data( … format=’JSON’, … content={‘nodes’: […]}, … name=’My Sim’, … attempt_start=True, … start_timeout=300 … )

import_from_simulation_manifest( *, simulation_manifest: dict[str, Any] | str | pathlib.Path | io.TextIOBase, attempt_start: bool = ..., start_timeout: datetime.timedelta | None = ..., ) → Simulation[source]#

Import simulation from a full JSON manifest file.

The manifest should contain all import parameters including: - format: ‘JSON’ - name: Simulation name - content: Topology data (for JSON format: dict with ‘nodes’, ‘links’,

‘oob’, ‘netq’)

ztp: Optional ZTP script content

Parameters:

simulation_manifest – Full simulation manifest (dict, JSON string, file path, or file handle)
attempt_start – When enabled, waits for the simulation creation to complete and then starts it automatically
start_timeout – Maximum time to wait for simulation creation (default: 120 seconds)

Returns:

The created Simulation instance

Raises:

ValueError – If manifest is missing required fields
FileNotFoundError – If file path doesn’t exist
JSONDecodeError – If JSON content is malformed

Example

# From dict of simulation manifest: >>> simulation_manifest = { … ‘format’: ‘JSON’, … ‘name’: ‘My Simulation’, … ‘ztp’: ‘#!/bin/bashnecho “ZTP”’, … ‘content’: { … ‘nodes’: {…}, … ‘links’: [], … ‘oob’: True, … ‘netq’: False … } … } >>> simulation = api.simulations.import_from_simulation_manifest( … simulation_manifest=simulation_manifest … )

# With attempt_start: >>> simulation_manifest = { … ‘format’: ‘JSON’, … ‘name’: ‘My Simulation’, … ‘content’: {…} … } >>> simulation = api.simulations.import_from_simulation_manifest( … simulation_manifest=simulation_manifest, … attempt_start=True, … start_timeout=300 … )

# From JSON file: >>> simulation = api.simulations.import_from_simulation_manifest( … ‘/path/to/manifest.json’ … )

# From Path object: >>> from pathlib import Path >>> simulation = api.simulations.import_from_simulation_manifest( … Path(‘/path/to/manifest.json’) … )

import_from_dot( *, topology_data: str | pathlib.Path | io.TextIOBase, name: str, ztp: str | None = ..., attempt_start: bool = ..., start_timeout: datetime.timedelta | None = ..., ) → Simulation[source]#

Import simulation from DOT topology file/content.

Parameters:

topology_data – DOT topology content (string, file path, Path object, or file handle)
name – Simulation name. If not provided, defaults to the graph name declared in the DOT content
ztp – Optional ZTP script content
attempt_start – When enabled, waits for the simulation creation to complete and then starts it automatically
start_timeout – Maximum time to wait for simulation creation (default: 120 seconds)

Returns:

The created Simulation instance

Raises:

ValueError – If content is invalid
FileNotFoundError – If file path doesn’t exist

Example

# From DOT string: >>> dot_content = ‘’’ … graph MyNetwork { … “server1” [ os=”generic/ubuntu2204” ] … “switch1” [ os=”cumulus/vx:5.11.0” ] … “server1”:”eth1” – “switch1”:”swp1” … } … ‘’’ >>> simulation = api.simulations.import_from_dot( … topology_data=dot_content, name=’My Simulation’ … )

# Import and start: >>> simulation = api.simulations.import_from_dot( … topology_data=dot_content, … name=’My Simulation’, … attempt_start=True … )

# Import and start with custom timeout: >>> simulation = api.simulations.import_from_dot( … topology_data=dot_content, … name=’My Simulation’, … attempt_start=True, … start_timeout=300 … )

# From DOT file path: >>> simulation = api.simulations.import_from_dot( … topology_data=’/path/to/topology.dot’, name=’My Simulation’ … )

# With ZTP script: >>> simulation = api.simulations.import_from_dot( … topology_data=dot_content, … name=’My Simulation’, … ztp=’#!/bin/bashnecho “ZTP”’ … )

list( *, auto_netq_enabled: bool = ..., auto_oob_enabled: bool = ..., disable_auto_oob_dhcp: bool = ..., id: str = ..., limit: int = ..., name: str = ..., offset: int = ..., ordering: str = ..., search: str = ..., state: str = ..., ) → Iterator[Simulation]#

List all simulations with optional filtering.

Args:
auto_netq_enabled: Filter by auto NetQ enabled status auto_oob_enabled: Filter by auto OOB enabled status disable_auto_oob_dhcp: Filter by disable auto OOB DHCP status id: Filter by simulation ID limit: Number of results to return per page name: Filter by simulation name offset: The initial index from which to return the results ordering: Order objects by field. Prefix with “-” for desc order search: Search by name state: Filter by simulation state (e.g., ‘ACTIVE’, ‘INACTIVE’,

‘CREATING’, ‘CLONING’, etc.)

Returns:: Iterator of Simulation instances

Example

>>> # List all simulations

>>> for sim in api.simulations.list():
...     print(sim.name)
        >>> # Filter by state
        >>> for sim in api.simulations.list(state='ACTIVE'):
        ...     print(sim.name)

>>> # Search by name
>>> for sim in api.simulations.list(search='my-sim'):
...     print(sim.name)

>>> # Order by name descending
>>> for sim in api.simulations.list(ordering='-name'):
...     print(sim.name)

get(pk: air_sdk.air_model.PrimaryKey) → Simulation#

Get a specific simulation by ID.

Parameters:: pk – The simulation ID (string or UUID)
Returns:: The Simulation instance

Example

>>> simulation = api.simulations.get('sim-id')

Update a simulation’s properties.

Parameters:

simulation – The simulation to update (Simulation object or ID)
name – New name for the simulation
sleep_at – When the simulation should be automatically put to sleep
expires_at – When the simulation should be automatically deleted
documentation – Documentation/description for the simulation

Returns:

The updated Simulation instance

Example

# Using Simulation object: >>> updated_sim = api.simulations.update( … simulation=simulation, name=’Updated Name’ … )

# Using simulation ID: >>> updated_sim = api.simulations.update( … simulation=’sim-123-abc’, … name=’Updated Name’, … documentation=’New docs’ … )

export( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, image_ids: bool = ..., topology_format: Literal['JSON'] = 'JSON', ) → dict[str, Any][source]#

Export a simulation.

Parameters:

simulation – The simulation to export (Simulation object or simulation ID)
image_ids – Whether to include image IDs in the export
topology_format – Format for the topology in the export

Returns:

Dictionary containing the exported simulation data

Example

# Using Simulation object: >>> export_data = api.simulations.export(simulation=simulation)

# Using simulation ID: >>> export_data = api.simulations.export(simulation=’sim-123-abc’)

# With optional parameters: >>> export_data = api.simulations.export( … simulation=simulation, image_ids=True, topology_format=’JSON’ … )

clone( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, checkpoint: str | None = ..., attempt_start: bool = ..., ) → Simulation[source]#

Clone/duplicate a simulation.

Parameters:

simulation – The simulation to clone (Simulation object or simulation ID)
checkpoint – Optional checkpoint ID to clone from. If not specified, Air will find the most recent COMPLETE checkpoint if it exists.
attempt_start – If the simulation should start immediately after cloning

Returns:

The cloned Simulation instance

Example

# Using Simulation object: >>> cloned_sim = api.simulations.clone(simulation=sim)

# Using simulation ID: >>> cloned_sim = api.simulations.clone(simulation=’sim-123-abc’)

# Clone with checkpoint: >>> cloned_sim = api.simulations.clone( … simulation=sim, checkpoint=’checkpoint-id’ … )

# Clone and start immediately: >>> cloned_sim = api.simulations.clone(simulation=sim, attempt_start=True)

enable_auto_oob( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, disable_auto_oob_dhcp: bool = ..., ) → None[source]#

Enable automatic Out-of-band management for a simulation.

Parameters:

simulation – The simulation object or simulation ID
disable_auto_oob_dhcp – If True, disable DHCP on the OOB management network

Example

# Enable OOB with DHCP (using simulation object): >>> api.simulations.enable_auto_oob(simulation=simulation)

# Enable OOB using simulation ID: >>> api.simulations.enable_auto_oob(simulation=’uuid-123’)

# Enable OOB without DHCP: >>> api.simulations.enable_auto_oob( … simulation=simulation, disable_auto_oob_dhcp=True … )

disable_auto_oob( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, ) → None[source]#

Disable automatic Out-of-band management for a simulation.

Parameters:: simulation – The simulation object or simulation ID

Example

# Using simulation object: >>> api.simulations.disable_auto_oob(simulation=simulation)

# Using simulation ID: >>> api.simulations.disable_auto_oob(simulation=’uuid-123’)

enable_auto_netq( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, ) → None[source]#

Enable automatic NetQ for a simulation.

Parameters:: simulation – The simulation object or simulation ID

Example

# Using simulation object: >>> api.simulations.enable_auto_netq(simulation=simulation)

# Using simulation ID: >>> api.simulations.enable_auto_netq(simulation=’uuid-123’)

disable_auto_netq( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, ) → None[source]#

Disable automatic NetQ for a simulation.

Parameters:: simulation – The simulation object or simulation ID

Example

# Using simulation object: >>> api.simulations.disable_auto_netq(simulation=simulation)

# Using simulation ID: >>> api.simulations.disable_auto_netq(simulation=’uuid-123’)

start( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, checkpoint: str | None = ..., ) → None[source]#

Start a simulation.

Parameters:

simulation – The simulation object or simulation ID to start
checkpoint – Optional checkpoint ID to start from. If not specified, the API will use its default behavior (typically uses the most recent checkpoint if available). If explicitly set to None, starts from clean state (rebuild). If a string, starts from the specified checkpoint.

Example

# Normal start (API determines behavior): >>> api.simulations.start(simulation=simulation)

# Start using simulation ID: >>> api.simulations.start(simulation=’uuid-123’)

# Start from specific checkpoint: >>> api.simulations.start(simulation=simulation, checkpoint=’checkpoint-id’)

# Rebuild (start from clean state, no checkpoint): >>> api.simulations.start(simulation=simulation, checkpoint=None)

rebuild( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, checkpoint: str | None = ..., ) → None[source]#

Rebuild a simulation from a given checkpoint.

Tears down the simulation and starts it from the given checkpoint. If no checkpoint is provided, the simulation will be rebuilt from the clean state.

Parameters:

simulation – The simulation object or simulation ID to rebuild
checkpoint – Optional checkpoint ID to rebuild from. If not specified, the API will use its default behavior (the current checkpoint the simulation is running off of). If explicitly set to None, starts from clean state (rebuild). If a string, starts from the specified checkpoint.

Example

# Rebuild simulation from specific checkpoint: >>> api.simulations.rebuild(simulation=simulation, checkpoint=’checkpoint-id’)

# Rebuild from clean state (no checkpoint): >>> api.simulations.rebuild(simulation=simulation, checkpoint=None)

# Rebuild using simulation ID: >>> api.simulations.rebuild(simulation=’uuid-123’)

shutdown( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, create_checkpoint: bool = ..., ) → None[source]#

Shut down a simulation.

Parameters:

simulation – The simulation object or simulation ID to shut down
create_checkpoint – Whether to create a checkpoint before shutting down. If not specified, the API will use its default behavior.

Example

# Normal shutdown (API determines behavior): >>> api.simulations.shutdown(simulation=simulation)

# Shutdown using simulation ID: >>> api.simulations.shutdown(simulation=’uuid-123’)

# Explicitly create checkpoint before shutdown: >>> api.simulations.shutdown(simulation=simulation, create_checkpoint=True)

# Explicitly don’t create checkpoint: >>> api.simulations.shutdown(simulation=simulation, create_checkpoint=False)

create_service( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, node_name: str, interface_name: str, node_port: int, name: str = ..., service_type: Literal['SSH', 'HTTPS', 'HTTP', 'OTHER'] = ..., ) → air_sdk.endpoints.services.Service[source]#

Create service for a simulation by resolving node and interface names.

Parameters:

simulation – Simulation ID or object
node_name – Node name in the simulation
interface_name – Interface name on the node (e.g., ‘eth0’, ‘swp1’)
node_port – Port number on the node
name – Service name (optional)
service_type – Service type - ‘SSH’, ‘HTTPS’, ‘HTTP’, or ‘OTHER’

Returns:

Service object

Raises:

ValueError – If node or interface not found in simulation

Example

>>> service = api.simulations.create_service(
...     simulation='sim-id',
...     node_name='server-1',
...     interface_name='eth0',
...     node_port=22,
...     service_type='SSH'
... )

parse( *, topology_data: str, source_format: str, destination_format: str, ) → dict[str, Any][source]#

Parse topology content between different formats.

Convert topology data between different formats (e.g., DOT to JSON).

Parameters:

topology_data – The topology content to parse
source_format – The format to parse the topology from (e.g., ‘DOT’).
destination_format – The format to parse the topology to (e.g., ‘JSON’).

Returns:

Parsed topology data as a dictionary

Example

>>> dot_content = '''
... graph MyNetwork {
...     "node-1" [ os="generic/ubuntu2204" cpu=2 ]
...     "node-2" [ os="generic/ubuntu2204" memory=2048 ]
...     "node-1":"eth1" -- "node-2":"eth1"
... }
... '''
>>> # Convert DOT to JSON
>>> parsed = api.simulations.parse(
...     topology_data=dot_content,
...     source_format='DOT',
...     destination_format='JSON',
... )

node_bulk_assign( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, nodes: List[air_sdk.types.NodeAssignmentDataV3], ) → None[source]#

Bulk assign configurations to nodes in given simulation.

Parameters:

simulation – The simulation to bulk assign to
nodes – List of node assignment data containing node, user_data, and meta_data

Example

>>> # Bulk assign cloud-init configs to multiple nodes
>>> api.simulations.node_bulk_assign(
...     simulation=simulation,
...     nodes=[
...         {'node': 'node1', 'user_data': 'config1'},
...         {'node': 'node2', 'meta_data': 'config2'},
...     ],
... )

node_bulk_reset( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, nodes: List[air_sdk.types.NodeResetPayload], ) → None[source]#

Reset specific nodes within a simulation.

Parameters:

simulation – The simulation object or simulation ID containing the nodes
nodes – List of node reset payloads, each containing a node object or ID

Example

>>> # Reset a single node using node object
>>> api.simulations.node_bulk_reset(
...     simulation=simulation,
...     nodes=[{'id': node}],
... )

>>> # Reset multiple nodes using node IDs
>>> api.simulations.node_bulk_reset(
...     simulation='sim-uuid-123',
...     nodes=[
...         {'id': 'node-uuid-1'},
...         {'id': 'node-uuid-2'},
...     ],
... )

>>> # Reset nodes retrieved from simulation
>>> nodes = [{'id': n} for n in simulation.nodes.list()]
>>> api.simulations.node_bulk_reset(simulation=simulation, nodes=nodes)

node_bulk_rebuild( *, simulation: Simulation | air_sdk.air_model.PrimaryKey, nodes: List[air_sdk.types.NodeRebuildPayload], checkpoint: str | None = ..., ) → None[source]#

Rebuild specific nodes within a simulation.

Rebuilding a node means returning the node to either the state of the current checkpoint of its simulation or its initial, first boot state. When rebuilding from the initial state, all repeatable instructions for selected nodes will be applied. All existing instruction created for the selected nodes which have not yet been completed will be failed. All existing instructions created for the selected nodes which have not yet been delivered will be cancelled.

Parameters:

simulation – The simulation object or simulation ID containing the nodes
nodes – List of node rebuild payloads, each containing a node object or ID
checkpoint – Optional checkpoint ID to rebuild from

Example

>>> # Rebuild a single node using node object
>>> api.simulations.node_bulk_rebuild(
...     simulation=simulation,
...     nodes=[{'id': node}],
... )

>>> # Rebuild multiple nodes using node IDs
>>> api.simulations.node_bulk_rebuild(
...     simulation='sim-uuid-123',
...     nodes=[
...         {'id': 'node-uuid-1'},
...         {'id': 'node-uuid-2'},
...     ],
... )

class air_sdk.endpoints.SSHKey[source]#

Bases: air_sdk.air_model.AirModel

SSH Key model representing a user’s SSH public key.

The string representation shows: id, name

id#: Unique identifier for the SSH key

created#: Timestamp when the SSH key was created

name#: Human-readable name for the SSH key

fingerprint#: SSH key fingerprint (automatically generated)

id: str#

created: datetime.datetime#

name: str#

fingerprint: str#

classmethod get_model_api() → type[SSHKeyEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: SSHKeyEndpointAPI#

delete() → None#

Delete this SSH key.

After deletion, the instance’s id will be set to None.

Example

>>> ssh_key = api.ssh_keys.get('key-id')
>>> ssh_key.delete()

class air_sdk.endpoints.SSHKeyEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[SSHKey]

API client for SSH key endpoints.

API_PATH: str#

model: type[SSHKey]#

list( *, limit: int = ..., offset: int = ..., ordering: str = ..., search: str = ..., ) → Iterator[SSHKey]#

List all SSH keys with optional filtering.

Parameters:

limit – Maximum number of results to return per page
offset – The initial index from which to return the results
ordering – Order objects by field. Prefix with “-” for desc order
search – Search by name

Returns:

Iterator of SSHKey instances

Example

>>> # List all SSH keys
>>> for key in api.ssh_keys.list():
...     print(key.name, key.fingerprint)

>>> # Search by name
>>> for key in api.ssh_keys.list(search='my-key'):
...     print(key.name)

>>> # Order by name descending
>>> for key in api.ssh_keys.list(ordering='-name'):
...     print(key.name)

create(*, name: str, public_key: str) → SSHKey#

Create a new SSH key.

Parameters:

name – Human-readable name for the SSH key
public_key – The SSH public key content (e.g., “ssh-rsa AAAA…”)

Returns:

The created SSHKey instance

Example

>>> ssh_key = api.ssh_keys.create(
name='my-laptop-key', public_key='ssh-rsa AAAAB3NzaC1yc2EAAAADAQAB...'
)
>>> print(ssh_key.fingerprint)

get(pk: air_sdk.air_model.PrimaryKey) → SSHKey#

Get a specific SSH key by ID.

Parameters:: pk – The SSH key ID (string or UUID)
Returns:: The SSHKey instance

Example

>>> ssh_key = api.ssh_keys.get('key-id')
>>> print(ssh_key.name, ssh_key.fingerprint)

delete(pk: air_sdk.air_model.PrimaryKey) → None#

Delete an SSH key by ID.

Parameters:: pk – The SSH key ID (string or UUID)
Returns:: None

Example

>>> # Delete by ID
>>> api.ssh_keys.delete('key-id')

class air_sdk.endpoints.System[source]#

Bases: air_sdk.bc.BaseCompatMixin, air_sdk.air_model.AirModel

System model representing a system in the AIR platform.

id: str#

created: datetime.datetime#

modified: datetime.datetime#

name: str#

simulation: air_sdk.endpoints.simulations.Simulation | None#

image: air_sdk.endpoints.images.Image#

memory: int#

storage: int#

cpu: int#

category: str#

attributes: dict[str, Any] | None#

classmethod get_model_api() → type[SystemEndpointAPI][source]#: Returns the respective AirModelAPI type for this model

property model_api: SystemEndpointAPI#: The current model API instance.

class air_sdk.endpoints.SystemEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.endpoints.mixins.ListApiMixin[System], air_sdk.endpoints.mixins.GetApiMixin[System], air_sdk.air_model.BaseEndpointAPI[System]

Endpoint API for System operations.

API_PATH = 'systems/nodes'#

model#

class air_sdk.endpoints.UserConfig[source]#

Bases: air_sdk.bc.BaseCompatMixin, air_sdk.air_model.AirModel

UserConfig model representing cloud-init configurations.

The string representation shows: id, name, kind

id#: Unique identifier for the user config

name#: Human-readable name of the user config

kind#: Type of cloud-init config (user-data or meta-data)

content#: The cloud-init script content (lazy-loaded)

Constants:: KIND_CLOUD_INIT_USER_DATA: Constant for user-data kind KIND_CLOUD_INIT_META_DATA: Constant for meta-data kind

id: str#

name: str#

kind: str#

content: str | None#

KIND_CLOUD_INIT_USER_DATA: str#

KIND_CLOUD_INIT_META_DATA: str#

classmethod get_model_api() → type[UserConfigEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

property model_api: UserConfigEndpointAPI#

update(*, name: str | None = ..., content: str | None = ...) → None[source]#

Update specific fields of the user config.

Parameters:

name – New name for the user config
content – New cloud-init script content

Example

>>> user_config = api.user_configs.get('config-id')
>>> user_config.update(name='Updated Config')
>>> user_config.update(content='#cloud-config\n...')

refresh() → None#

Refresh the user config from the API, loading all fields including content.

Example

>>> user_config.refresh()
>>> print(user_config.content)

delete() → None#

Delete this user config from the system.

Example

>>> user_config.delete()

class air_sdk.endpoints.UserConfigEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.bc.UserConfigEndpointAPICompatMixin, air_sdk.air_model.BaseEndpointAPI[UserConfig]

API interface for managing UserConfig resources.

Provides methods for CRUD operations on user configs (cloud-init configurations).

API_PATH: str#

model: type[UserConfig]#

List user configs with optional filtering and pagination.

Parameters:

kind – Filter by config kind (user-data or meta-data)
limit – Maximum number of results to return
offset – Number of results to skip
ordering – Field to order results by (prefix with ‘-’ for descending)
search – Search term to filter results

Yields:

UserConfig instances

Example

>>> # List all user configs
>>> for config in api.user_configs.list():
...     print(config.name)
>>>
>>> # Filter by kind
>>> for config in api.user_configs.list(
...     kind=api.user_configs.model.KIND_CLOUD_INIT_USER_DATA
... ):
...     print(config.name)
>>>
>>> # Search with pagination
>>> for config in api.user_configs.list(search='my-config', limit=10):
...     print(config.name)

get( user_config: UserConfig | air_sdk.air_model.PrimaryKey, ) → UserConfig#

Get a specific user config by ID.

Parameters:: user_config – UserConfig instance or ID string
Returns:: UserConfig instance with all fields populated

Example

>>> config = api.user_configs.get('config-id-here')
>>> print(config.content)

create( *, name: str, kind: str, content: str | pathlib.Path | io.TextIOBase, ) → UserConfig[source]#

Create a new user config.

Parameters:

name – Name for the user config
kind – Type of config (use KIND_CLOUD_INIT_USER_DATA or KIND_CLOUD_INIT_META_DATA constants)
content – Cloud-init script content. Can be provided as: - str: Raw content or file path - Path: Path object to a file - TextIOBase: Open file handle

Returns:

Newly created UserConfig instance

Note

User configs are automatically associated with your default organization. The ‘organization’ parameter from v1/v2 has been removed in v3.

Example

>>> # String content
>>> config = api.user_configs.create(
...     name='my-user-data',
...     kind=api.user_configs.model.KIND_CLOUD_INIT_USER_DATA,
...     content='#cloud-config\npackages:\n  - vim\n',
... )
>>>
>>> # File path
>>> config = api.user_configs.create(
...     name='my-user-data',
...     kind='cloud-init-user-data',
...     content='/path/to/file.txt',
... )
>>>
>>> # Path object
>>> from pathlib import Path
>>> config = api.user_configs.create(
...     name='my-user-data',
...     kind='cloud-init-user-data',
...     content=Path('/path/to/file.txt'),
... )

Update a user config (patch operation).

Parameters:

user_config – UserConfig instance or ID string
name – New name for the user config
content – New cloud-init script content

Returns:

Updated UserConfig instance

Example

>>> config = api.user_configs.update(
...     user_config='config-id',
...     name='New Name',
...     content='#cloud-config\n...',
... )

patch( pk: air_sdk.air_model.PrimaryKey, *, name: str | None = ..., content: str | pathlib.Path | io.TextIOBase | None = ..., ) → UserConfig[source]#

Patch a user config with partial updates.

Parameters:

pk – UserConfig ID string
name – New name for the user config
content – New cloud-init script content

Returns:

Updated UserConfig instance

Example

>>> config = api.user_configs.patch(
...     pk='config-id',
...     name='Updated Name',
... )

delete(pk: air_sdk.air_model.PrimaryKey) → None#

Delete a user config.

Parameters:: pk – UserConfig ID string

Example

>>> api.user_configs.delete('config-id')

class air_sdk.endpoints.Worker[source]#

Bases: air_sdk.air_model.AirModel

Helper class that provides a standard way to create an ABC using inheritance.

id: str#

created: datetime.datetime#

modified: datetime.datetime#

fqdn: str#

fleet: air_sdk.endpoints.fleets.Fleet#

cpu: int#

memory: int#

storage: int#

available: bool#

cpu_arch: str#

ip_address: str#

registration_token: str | None#

classmethod get_model_api() → type[WorkerEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

update( *, fqdn: str | dataclasses._MISSING_TYPE = MISSING, ip_address: str | dataclasses._MISSING_TYPE = MISSING, cpu: int | dataclasses._MISSING_TYPE = MISSING, memory: int | dataclasses._MISSING_TYPE = MISSING, storage: int | dataclasses._MISSING_TYPE = MISSING, available: bool | dataclasses._MISSING_TYPE = MISSING, ) → None[source]#

Update specific fields of the worker.

Example

>>> worker = api.workers.get('123e4567-e89b-12d3-a456-426614174000')
>>> worker.update(cpu=16)

issue_certificate() → PEMCertificateData[source]#

Issue a new client certificate for the worker.

Example

>>> worker = api.workers.get('123e4567-e89b-12d3-a456-426614174000')
>>> cert_data = worker.issue_certificate()
>>> cert, key = cert_data['certificate'], cert_data['private_key']

class air_sdk.endpoints.WorkerClientCertificate[source]#

Bases: air_sdk.air_model.AirModel

Helper class that provides a standard way to create an ABC using inheritance.

id: str#

worker: Worker#

worker_fqdn: str#

usable: bool#

expires: datetime.datetime#

fingerprint: str#

last_used: datetime.datetime | None#

revoked: bool#

classmethod get_model_api() → type[WorkerClientCertificateEndpointAPI][source]#: Returns the respective AirModelAPI type for this model.

revoke() → None[source]#

Revoke this client certificate. Once a certificate is revoked, it may no longer be used!

Example

>>> certificates = api.worker_client_certificates.list()
>>> for certificate in certificates:
...     if certificate.fingerprint == '...':
...         certificate.revoke()

class air_sdk.endpoints.WorkerClientCertificateEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.endpoints.mixins.ListApiMixin[WorkerClientCertificate], air_sdk.endpoints.mixins.GetApiMixin[WorkerClientCertificate], air_sdk.air_model.BaseEndpointAPI[WorkerClientCertificate]

Returns an iterable of model objects.

Handles pagination in the background.

API_PATH = 'infra/workers/certificates/'#

REVOKE_PATH = 'revoke'#

model#

list(

worker: Worker | air_sdk.air_model.PrimaryKey | dataclasses._MISSING_TYPE = MISSING,

search: str | dataclasses._MISSING_TYPE = MISSING,

ordering: str | dataclasses._MISSING_TYPE = MISSING,

**params: Any,

) → Iterator[WorkerClientCertificate][source]#

List all worker certificates.

Example

>>> worker = api.workers.get('123e4567-e89b-12d3-a456-426614174000')
>>> for certificate in api.worker_client_certificates.list(worker=worker):
...     print(certificate.fingerprint)

class air_sdk.endpoints.WorkerEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.endpoints.mixins.ListApiMixin[Worker], air_sdk.endpoints.mixins.CreateApiMixin[Worker], air_sdk.endpoints.mixins.GetApiMixin[Worker], air_sdk.endpoints.mixins.PatchApiMixin[Worker], air_sdk.endpoints.mixins.DeleteApiMixin, air_sdk.air_model.BaseEndpointAPI[Worker]

Returns an iterable of model objects.

Handles pagination in the background.

API_PATH = 'infra/workers/'#

ISSUE_CERTIFICATE_PATH = 'issue-certificate'#

model#

list(

fqdn: str | dataclasses._MISSING_TYPE = MISSING,

search: str | dataclasses._MISSING_TYPE = MISSING,

ordering: str | dataclasses._MISSING_TYPE = MISSING,

**params: Any,

) → Iterator[Worker][source]#

List all workers.

Example

>>> for worker in api.workers.list(ordering='fqdn'):
...     print(worker.fqdn)

create( *, fleet: air_sdk.endpoints.fleets.Fleet | air_sdk.air_model.PrimaryKey, ip_address: str, fqdn: str, cpu_arch: str | dataclasses._MISSING_TYPE = MISSING, ) → Worker[source]#

Create a new worker.

Example

>>> fleet = api.fleets.get('123e4567-e89b-12d3-a456-426614174000')
>>> image = api.images.get('456e89ab-cdef-0123-4567-89abcdef0123')
>>> node = api.nodes.create(simulation=sim, image=image, name='my-node')

class air_sdk.endpoints.ZTPScript[source]#

Bases: air_sdk.air_model.AirModel

A ZTP (Zero Touch Provisioning) script for a simulation.

ZTPScript objects should not be created directly. Use simulation.create_ztp_script(content=’…’) instead.

Note

When printed, the content field is truncated to 50 characters with newlines collapsed to avoid flooding console/logs with large scripts. Access the full content via ztp_script.content.

created#: When the script was created.

modified#: When the script was last modified.

content#: The script content.

simulation#: The simulation this script belongs to.

Examples

>>> # Create a ZTP script
>>> content = '#!/bin/bash\n# CUMULUS-AUTOPROVISIONING\necho Hi'
>>> ztp_script = simulation.create_ztp_script(content=content)

>>> # Update the script
>>> content = '#!/bin/bash\n# CUMULUS-AUTOPROVISIONING\necho "Updated!"'
>>> ztp_script.update(content=content)

>>> # Delete the script
>>> ztp_script.delete()
>>> print(simulation.ztp_script)
None

created: datetime.datetime#

modified: datetime.datetime#

content: str#

simulation: air_sdk.endpoints.simulations.Simulation#

classmethod get_model_api() → type[ZTPScriptEndpointAPI][source]#: Returns the respective AirModelAPI type for this model

property model_api: ZTPScriptEndpointAPI#: The current model API instance.

update(*, content: str) → None[source]#

Update the content of this ZTP script.

Parameters:: content – The new script content to use.

Example

>>> ztp_script = simulation.ztp_script
>>> content = '#!/bin/bash\n# CUMULUS-AUTOPROVISIONING\n'
>>> content += 'echo "Updated script"'
>>> ztp_script.update(content=content)

delete() → None[source]#

Delete the instance.

After deletion, accessing simulation.ztp_script will return None.

Example

>>> ztp_script = simulation.ztp_script
>>> ztp_script.delete()
>>> print(simulation.ztp_script)
None

class air_sdk.endpoints.ZTPScriptEndpointAPI( api: air_sdk.AirApi, default_filters: dict[str, Any] | None = None, )[source]#

Bases: air_sdk.air_model.BaseEndpointAPI[ZTPScript]

Retrieve, update, and delete ZTP scripts for simulations.

ZTPScripts should be created during import or off of Simulation objects:

Examples

>>> # Create a ZTP script
>>> content = '#!/bin/bash\n# CUMULUS-AUTOPROVISIONING\necho "Hello!"'
>>> simulation.create_ztp_script(content=content)
<ZTPScript(content='#!/bin/bash...')>

>>> # Get a ZTP script
>>> ztp_script = api.ztp_scripts.get(simulation)
>>> print(ztp_script.content)
#!/bin/bash
# CUMULUS-AUTOPROVISIONING
echo "Hello!"

>>> # Update a ZTP script
>>> content = '#!/bin/bash\n# CUMULUS-AUTOPROVISIONING\necho "Updated!"'
>>> api.ztp_scripts.patch(simulation, content=content)

>>> # Delete a ZTP script
>>> api.ztp_scripts.delete(simulation)

API_PATH: str#

model: type[ZTPScript]#

get( *, simulation: air_sdk.endpoints.simulations.Simulation | air_sdk.air_model.PrimaryKey, ) → ZTPScript[source]#

Get the ZTP script for the simulation if it exists.

Parameters:: simulation – The simulation object or simulation ID.
Returns:: The ZTP script for the simulation.
Raises:: AirUnexpectedResponse – If the simulation doesn’t have a ZTP script.

Example

>>> # Using simulation object
>>> ztp_script = api.ztp_scripts.get(simulation)

>>> # Using simulation ID
>>> ztp_script = api.ztp_scripts.get('simulation-uuid')

>>> print(ztp_script.content)
#!/bin/bash
# CUMULUS-AUTOPROVISIONING
echo "Hello, world!"

patch( *, simulation: air_sdk.endpoints.simulations.Simulation | air_sdk.air_model.PrimaryKey, content: str, ) → ZTPScript[source]#

Update the content of the ZTPScript.

Parameters:

simulation – The simulation object or simulation ID.
content – The new script content.

Returns:

The updated ZTP script.

Examples

>>> # Using simulation object
>>> with open('ztp_script.sh', 'r') as f:
...     content = f.read()
>>> updated_script = api.ztp_scripts.patch(simulation, content=content)

>>> # Using simulation ID
>>> content = '#!/bin/bash\n# CUMULUS-AUTOPROVISIONING\necho Hi'
>>> api.ztp_scripts.patch('simulation-uuid', content=content)

update( *, simulation: air_sdk.endpoints.simulations.Simulation | air_sdk.air_model.PrimaryKey, content: str, ) → ZTPScript[source]#

Update the content of the ZTPScript.

This is an alias for patch() provided for consistency with other endpoints.

Parameters:

simulation – The simulation object or simulation ID.
content – The new script content.

Returns:

The updated ZTP script.

Examples

>>> # Using simulation object
>>> content = '#!/bin/bash\n# CUMULUS-AUTOPROVISIONING'
>>> api.ztp_scripts.update(simulation=simulation, content=content)

>>> # Using simulation ID
>>> api.ztp_scripts.update(simulation='sim-id', content=content)

delete( *, simulation: air_sdk.endpoints.simulations.Simulation | air_sdk.air_model.PrimaryKey, ) → None[source]#

Delete the ZTP script for the simulation.

After deletion, simulation.ztp_script will return None.

Parameters:: simulation – The simulation object or simulation ID.

Examples

>>> # Using simulation object
>>> api.ztp_scripts.delete(simulation)

>>> # Using simulation ID
>>> api.ztp_scripts.delete('simulation-uuid')

>>> print(simulation.ztp_script)
None

air_sdk.endpoints.ServiceAPI#

air_sdk.endpoints.ServiceEndpointApi#

air_sdk.endpoints.SimulationApi#

air_sdk.endpoints.SimulationEndpointApi#

air_sdk.endpoints.SimulationNodeApi#

air_sdk.endpoints.ImageApi#

air_sdk.endpoints.ImageEndpointApi#

air_sdk.endpoints.MarketplaceDemoApi#

air_sdk.endpoints.MarketplaceDemoEndpointApi#

air_sdk.endpoints.InterfaceApi#

air_sdk.endpoints.InterfaceEndpointApi#

air_sdk.endpoints.SimulationInterfaceApi#

air_sdk.endpoints.NodeInstructionsEndpointApi#

air_sdk.endpoints.UserConfigAPI#

air_sdk.endpoints.UserConfigEndpointApi#

air_sdk.endpoints.ResourceBudget#

air_sdk.endpoints.ResourceBudgetEndpointAPI#