air_sdk.endpoints.workers

Classes

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
PEMCertificateData	Certificate data issued for the worker in PEM format.
WorkerEndpointAPI	Returns an iterable of model objects.
WorkerClientCertificateEndpointAPI	Returns an iterable of model objects.

Module Contents

class air_sdk.endpoints.workers.Worker

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]

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

Update specific fields of the worker.

Example

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

issue_certificate() → PEMCertificateData

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.workers.WorkerClientCertificate

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]

Returns the respective AirModelAPI type for this model.

revoke() → None

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.workers.PEMCertificateData

Bases: TypedDict

Certificate data issued for the worker in PEM format.

certificate: str

private_key: str

class air_sdk.endpoints.workers.WorkerEndpointAPI(

api: air_sdk.AirApi,

default_filters: dict[str, Any] | None = None,

)

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]

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

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.workers.WorkerClientCertificateEndpointAPI(

api: air_sdk.AirApi,

default_filters: dict[str, Any] | None = None,

)

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]

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)