Response

This script defines the Response class and its associated types for handling responses from the control interface. It includes methods for parsing the received messages and converting them into appropriate Python objects.

Classes

Response:
Represents a response from the control interface.
UpdateStateSuccess:
Represents a response for a successful update state request.
LogEntry:
Represents a log entry from a workload instance.
LogsStopResponse:
Represents a response for marking the end of the log stream from a workload instance.
EventEntry:
Represents an event.

Enums

ResponseType:
Enumeration for the different types of response. It includes responses like ERROR, CONNECTION_CLOSED, COMPLETE_STATE and so on.

Union Types

LogResponse:
Union type for log responses, which can be either LogEntry or LogsStopResponse.

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.

Variables:

buffer (bytes) – The received message buffer.
content_type (ResponseType) – The type of the response content.
content – The content of the response, which can store any response type.

__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

get_content()[source]

Gets the content of the response. It can be either:

a string (error / connection closed)
a CompleteState object
an UpdateStateSuccess object
a list of log entires
a log stop response

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

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.

Variables:

added_workloads (list[WorkloadInstanceName]) – The list of added workloads.
deleted_workloads (list[WorkloadInstanceName]) – The list of deleted workloads.

__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)

CONTROL_INTERFACE_ACCEPTED = 2

Control interface connection accepted.

Type:: (int)

COMPLETE_STATE = 3

Got the complete state.

Type:: (int)

UPDATE_STATE_SUCCESS = 4

Got a successful update state response.

Type:: (int)

LOGS_ENTRY = 5

Got logs entry.

Type:: (int)

LOGS_REQUEST_ACCEPTED = 6

Logs request accepted, waiting for logs.

Type:: (int)

LOGS_STOP_RESPONSE = 7

Got logs stop response.

Type:: (int)

LOGS_CANCEL_ACCEPTED = 8

Logs cancel request accepted.

Type:: (int)

EVENT_RESPONSE = 9

Got an event response.

Type:: (int)

EVENT_CANCEL_ACCEPTED = 10

Event cancel request accepted.

Type:: (int)

CONNECTION_CLOSED = 20

Connection closed by the server.

Type:: (int)

__str__()[source]

Converts the ResponseType to a string.

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

LogResponse Union Type

ankaios_sdk._components.response.LogResponse: alias of LogEntry | LogsStopResponse

LogEntry Class

class ankaios_sdk._components.response.LogEntry(workload_instance_name, message)[source]

Bases: object

Represents a log entry from a workload instance.

workload_instance_name: WorkloadInstanceName: The name of the workload instance from which the log entry was received.

message: str: The log message.

__str__()[source]

Converts the LogEntry to a string.

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

LogsStopResponse Class

class ankaios_sdk._components.response.LogsStopResponse(workload_instance_name)[source]

Bases: object

Represents a response for marking the end of the log stream from a workload instance.

workload_instance_name: WorkloadInstanceName: The name of the workload instance from which no more logs will be sent.

__str__()[source]

Converts the LogsStopResponse to a string.

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

EventEntry Class

class ankaios_sdk._components.response.EventEntry(complete_state, added_fields, updated_fields, removed_fields)[source]

Bases: object

Represents an event notification.

complete_state: CompleteState: The complete state of the event indicating the state changes.

added_fields: list[str]: The list of added fields of the state.

updated_fields: list[str]: The list of updated fields of the state.

removed_fields: list[str]: The list of removed fields of the state.

__str__()[source]

Converts the EventEntry to a string.

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