air_sdk#

Submodules#

Exceptions#

AirModelAttributeError

Common base class for all non-exit exceptions.

Classes#

`Client`	A session client for managing the execution of API requests.
`DockerRunParameters`	Docker run parameters for simulator containers.
`DockerRunTmpfsParameter`	Docker tmpfs mount configuration.
`EmulationParams`	Emulation parameters for a simulator/platform.
`Platform`	Emulated platform configuration.
`ResourceBudgetUsage`	Current resource usage within an organization's budget.
`Resources`	Resource overhead for a simulator/platform.
`SimState`	Simulation state constants.
`AirApi`
`MarketplaceDemoEndpointAPI`	API client for marketplace demo endpoints.
`MarketplaceDemoTagEndpointAPI`	API client for marketplace demo tag endpoints.
`Organization`	Represents an organization and its resource budget in the Air platform.
`OrganizationEndpointAPI`	API client for Organization / ResourceBudget endpoints.
`SSHKey`	SSH Key model representing a user's SSH public key.
`SSHKeyEndpointAPI`	API client for SSH key endpoints.

Package Contents#

class air_sdk.Client(api_url: str)[source]#

Bases: requests.Session

A session client for managing the execution of API requests.

api_url#

connect_timeout#

read_timeout#

pagination_page_size = 200#

get_basic_headers() → dict[str, str][source]#: Return generic headers for all requests.

static get_telemetry_headers() → dict[str, str][source]#: Return telemetry-specific headers for all requests.

get_user_agent_header_value() → str[source]#: Return the user agent header value.

get_http_adapter() → requests.adapters.HTTPAdapter[source]#: Return an HTTP adapter for all requests.

get_retry_strategy() → urllib3.util.Retry[source]#: Return a retry strategy for all requests. The following retry strategy will attempt to retry the request only if the connection fails.

hunt_for_sak() → str[source]#

ngc_sak_login() → None[source]#

No args. Client must have directory ~/.ngc/config

ngc_device_login(email: str, ngc_org_name: str) → None[source]#: Log in with device login - does not require a NGC API Key/SAK.

request(*args: Any, **kwargs: Any) → requests.Response[source]#: Override request method to pass the timeout

exception air_sdk.AirModelAttributeError( message: str = 'An error occurred while accessing an AirModel attribute.', status_code: int | None = None, )[source]#

Bases: AirError

Common base class for all non-exit exceptions.

message = 'An error occurred while accessing an AirModel attribute.'#

class air_sdk.DockerRunParameters[source]#

Bases: TypedDict

Docker run parameters for simulator containers.

tmpfs: list[DockerRunTmpfsParameter]#

cap_add: list[str]#

devices: list[str]#

volumes: list[str]#

environment: dict[str, str]#

class air_sdk.DockerRunTmpfsParameter[source]#

Bases: TypedDict

Docker tmpfs mount configuration.

path: str#

size_gb: int#

class air_sdk.EmulationParams[source]#

Bases: TypedDict

Emulation parameters for a simulator/platform.

direct_link_emulation: bool#

max_network_pci: int#

class air_sdk.Platform[source]#

Bases: TypedDict

Emulated platform configuration.

cpu: int#

memory: int#

default_port_type: str#

default_port_count: int#

port_count_options: list[int]#

class air_sdk.ResourceBudgetUsage[source]#

Bases: TypedDict

Current resource usage within an organization’s budget.

cpu#: Number of CPU cores currently in use

memory#: Memory currently in use, in MiB

disk_storage#: Disk storage currently in use, in GB

image_storage#: Image storage currently in use, in GB

userconfigs#: User configs content currently in use, in bytes

cpu: float#

memory: float#

disk_storage: float#

image_storage: int#

userconfigs: int#

class air_sdk.Resources[source]#

Bases: TypedDict

Resource overhead for a simulator/platform.

cpu: int#

memory: int#

class air_sdk.SimState[source]#

Bases: str, enum.Enum

Simulation state constants.

Use these instead of raw strings for type safety and IDE autocomplete.

Example

>>> from air_sdk import SimState
>>> while sim.state != SimState.ACTIVE:
...     sleep(5)
...     sim.refresh()

CLONING = 'CLONING'#

CREATING = 'CREATING'#

IMPORTING = 'IMPORTING'#

INVALID = 'INVALID'#

INACTIVE = 'INACTIVE'#

REQUESTING = 'REQUESTING'#

PROVISIONING = 'PROVISIONING'#

PREPARE_BOOT = 'PREPARE_BOOT'#

BOOTING = 'BOOTING'#

ACTIVE = 'ACTIVE'#

PREPARE_SHUTDOWN = 'PREPARE_SHUTDOWN'#

SHUTTING_DOWN = 'SHUTTING_DOWN'#

SAVING = 'SAVING'#

PREPARE_TEARDOWN = 'PREPARE_TEARDOWN'#

TEARING_DOWN = 'TEARING_DOWN'#

DELETING = 'DELETING'#

PREPARE_PURGE = 'PREPARE_PURGE'#

PURGING = 'PURGING'#

DEMO = 'DEMO'#

TRAINING = 'TRAINING'#

PREPARE_REBUILD = 'PREPARE_REBUILD'#

REBUILDING = 'REBUILDING'#

class air_sdk.AirApi( api_url: str = const.AIR_API_URL, authenticate: bool = True, api_key: str | None = None, username: str | None = None, password: str | None = None, bearer_token: str | None = None, )[source]#

client#

classmethod with_api_key(api_key: str, api_url: str = const.AIR_API_URL) → AirApi[source]#

Initialize API with an explicit NGC API Key.

The api_key is also known as a Starfleet API Key, or ‘SAK’.

classmethod with_device_login( email: str, org_num: str, api_url: str = const.AIR_API_URL, ) → AirApi[source]#: Initialize API with device login authentication.

classmethod with_ngc_config(api_url: str = const.AIR_API_URL) → AirApi[source]#: Initialize API using NGC config for authentication.

auth_with_ngc_config() → None[source]#

Authenticate with a local NGC config.

This method is called by the AirApi constructor if authenticate is True.

property histories: endpoints.HistoryEndpointAPI#

property images: endpoints.ImageEndpointAPI#

property image_shares: endpoints.ImageShareEndpointAPI#

property simulations: endpoints.SimulationEndpointAPI#

property systems: endpoints.SystemEndpointAPI#

property nodes: endpoints.NodeEndpointAPI#

property interfaces: endpoints.InterfaceEndpointAPI#

property services: endpoints.ServiceEndpointAPI#

property simulation_interfaces: endpoints.InterfaceEndpointAPI#

property node_instructions: endpoints.NodeInstructionEndpointAPI#

property simulation_nodes: endpoints.NodeEndpointAPI#

property ztp_scripts: endpoints.ZTPScriptEndpointAPI#

property workers: endpoints.WorkerEndpointAPI#

property worker_client_certificates: endpoints.WorkerClientCertificateEndpointAPI#

property fleets: endpoints.FleetEndpointAPI#

property marketplace_demos: endpoints.MarketplaceDemoEndpointAPI#

property marketplace_demo_tags: endpoints.MarketplaceDemoTagEndpointAPI#

property ssh_keys: endpoints.SSHKeyEndpointAPI#

property manifests: endpoints.ManifestEndpointAPI#

property cloud_inits: bc.CloudInitEndpointAPI#: Cloud-init endpoint for V2 backwards compatibility.

property user_configs: endpoints.UserConfigEndpointAPI#

property organizations: endpoints.OrganizationEndpointAPI#: Organization / Resource Budget endpoint.

property resource_budgets: endpoints.OrganizationEndpointAPI#: Resource Budget endpoint (alias for organizations).

property simulation: endpoints.SimulationEndpointAPI#: Deprecated alias for simulations (v1 backward compatibility).

property service: endpoints.ServiceEndpointAPI#: Deprecated alias for services (v1 backward compatibility).

set_connect_timeout(t_delta: datetime.timedelta) → None[source]#

set_read_timeout(t_delta: datetime.timedelta) → None[source]#

set_page_size(n: int) → None[source]#: Set the page size of paginated responses.

class air_sdk.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.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.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.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)

air_sdk.ResourceBudget#

air_sdk.ResourceBudgetEndpointAPI#

class air_sdk.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.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')

air_sdk.SimulationApi#

air_sdk.SimulationEndpointApi#

air_sdk.ImageApi#

air_sdk.ImageEndpointApi#

air_sdk.ServiceAPI#

air_sdk.ServiceEndpointApi#

air_sdk.MarketplaceDemoApi#

air_sdk.MarketplaceDemoEndpointApi#

air_sdk.MarketplaceDemoTagApi#

air_sdk.InterfaceApi#

air_sdk.InterfaceEndpointApi#

air_sdk.SimulationInterfaceApi#

air_sdk.SimulationNodeApi#

air_sdk.NodeEndpointApi#

air_sdk.NodeApi#

air_sdk.UserConfigAPI#

air_sdk.UserConfigEndpointApi#

air_sdk.OrganizationApi#

`ResourceBudget`
`ResourceBudgetEndpointAPI`
`SimulationApi`
`SimulationEndpointApi`
`ImageApi`
`ImageEndpointApi`
`ServiceAPI`
`ServiceEndpointApi`
`MarketplaceDemoApi`
`MarketplaceDemoEndpointApi`
`MarketplaceDemoTagApi`
`InterfaceApi`
`InterfaceEndpointApi`
`SimulationInterfaceApi`
`SimulationNodeApi`
`NodeEndpointApi`
`NodeApi`
`UserConfigAPI`
`UserConfigEndpointApi`
`OrganizationApi`

air_sdk#

Submodules#

Attributes#

Exceptions#

Classes#

Package Contents#