air_sdk.endpoints.workers | DSX Air Documentation

Classes

Name	Description
`Worker`	None
`WorkerClientCertificate`	None
`PEMCertificateData`	Certificate data issued for the worker in PEM format.
`WorkerEndpointAPI`	None
`WorkerClientCertificateEndpointAPI`	None

Module Contents

class air_sdk.endpoints.workers.Worker

Bases: air_sdk.air_model.AirModel

id: str = field(...)

created: datetime.datetime = field(...)

modified: datetime.datetime = field(...)

fqdn: str

fleet: Fleet = field(...)

cpu: int = field(...)

memory: int = field(...)

storage: int = field(...)

available: bool

cpu_arch: str

ip_address: str

registration_token: str | None = field(...)

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

id: str = field(...)

worker: Worker = field(...)

worker_fqdn: str

usable: bool = field(...)

expires: datetime.datetime

fingerprint: str = field(...)

last_used: datetime.datetime | None = field(...)

revoked: bool

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: typing.TypedDict

Certificate data issued for the worker in PEM format.

certificate: str

private_key: str

class air_sdk.endpoints.workers.WorkerEndpointAPI

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

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: Fleet | 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

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

API_PATH = infra/workers/certificates/

REVOKE_PATH = revoke

model

list(
    worker: Worker | 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)