Response

This script defines the Response and ResponseEvent classes, used for receiving messages from the control interface.

Classes

Response:
Represents a response from the control interface.
ResponseEvent:
Represents an event used to wait for a response.
UpdateStateSuccess:
Represents a response for a successful update state request.

Enums

ResponseType:
Enumeration for the different types of response. It includes ERROR, COMPLETE_STATE, and UPDATE_STATE_SUCCESS.

Usage

Get response content:

response = Response()
(content_type, content) = response.get_content()

Check if the request_id matches:

response = Response()
if response.check_request_id("1234"):
    print("Request ID matches")

Convert the update state success to a dictionary:

update_state_success = UpdateStateSuccess()
update_state_success.to_dict()

Response Class

class ankaios_sdk._components.response.Response(message_buffer)[source]

Bases: object

Represents a response received from the Ankaios system.

buffer

The received message buffer.

Type:: bytes

content_type

The type of the response content (e.g., “error”, “complete_state”, “update_state_success”).

Type:: str

content: The content of the response, which can be a string, CompleteState, or UpdateStateSuccess.

__init__(message_buffer)[source]

Initializes the Response object with the received message buffer.

Parameters:: message_buffer (bytes) – The received message buffer.

get_request_id()[source]

Gets the request id of the response.

Returns:: The request id of the response.
Return type:: str

check_request_id(request_id)[source]

Checks if the request id of the response matches the given request id.

Parameters:: request_id (str) – The request id to check against.
Returns:: True if the request_id matches, False otherwise.
Return type:: bool

get_content()[source]

Gets the content of the response. It can be either a string (if error), a CompleteState instance, or a UpdateStateSuccess instance.

Returns:: the content type and the content.
Return type:: tuple[ResponseType, any]

ResponseEvent Class

class ankaios_sdk._components.response.ResponseEvent(response=None)[source]

Bases: Event

Represents an event that holds a Response object.

__init__(response=None)[source]

Initializes the ResponseEvent with an optional Response object.

Parameters:: Optional (response) – The response to associate with the event. Defaults to None.

set_response(response)[source]

Sets the response and triggers the event.

Parameters:: response (Response) – The response to set.
Return type:: None

get_response()[source]

Gets the response associated with the event.

Returns:: The response associated with the event.
Return type:: Response

wait_for_response(timeout)[source]

Waits for the response to be set, with a specified timeout.

Parameters:: timeout (int) – The maximum time to wait for the response, in seconds.
Returns:: The response associated with the event.
Return type:: Response
Raises:: TimeoutError – If the response is not set within the specified timeout.

UpdateStateSuccess Class

class ankaios_sdk._components.response.UpdateStateSuccess[source]

Bases: object

Represents an object that holds the added and deleted workloads. This is automatically returned whenever a state update is successful.

__init__()[source]: Initializes the UpdateStateSuccess.

to_dict()[source]

Converts the UpdateStateSuccess to a dictionary.

Returns:: The dictionary representation.
Return type:: dict

__str__()[source]

Converts the UpdateStateSuccess to a string.

Returns:: The string representation.
Return type:: str

ResponseType Enum

class ankaios_sdk._components.response.ResponseType(value)[source]

Bases: Enum

Enumeration for the different types of response.

ERROR = 1

Got an error from Ankaios.

Type:: (int)

COMPLETE_STATE = 2

Got the complete state.

Type:: (int)

UPDATE_STATE_SUCCESS = 3

Got a successful update state response.

Type:: (int)

__str__()[source]

Converts the ResponseType to a string.

Returns:: The string representation of the ResponseType.
Return type:: str