whylogs.api.logger.experimental.logger.actor.actor#

Module Contents#

Classes#

QueueWrapper

Helper class that provides a standard way to create an ABC using

CloseMessage

QueueConfig

Configuration for the queue used by the actor.

Actor

Simple actor inspired interface designed for batch processing messages.

Attributes#

QueueMessageType

DEFAULT_TIMEOUT

MessageType

whylogs.api.logger.experimental.logger.actor.actor.QueueMessageType#
whylogs.api.logger.experimental.logger.actor.actor.DEFAULT_TIMEOUT = 0.1#
class whylogs.api.logger.experimental.logger.actor.actor.QueueWrapper#

Bases: abc.ABC, Generic[QueueMessageType]

Helper class that provides a standard way to create an ABC using inheritance.

abstract send(message: QueueMessageType, timeout: float = DEFAULT_TIMEOUT) → None#
Parameters

  • message (QueueMessageType) –

  • timeout (float) –

Return type

None

abstract send_many(messages: List[QueueMessageType], timeout: float = DEFAULT_TIMEOUT) → None#
Parameters

  • messages (List[QueueMessageType]) –

  • timeout (float) –

Return type

None

abstract get(timeout: float = DEFAULT_TIMEOUT) → Optional[QueueMessageType]#
Parameters

timeout (float) –

Return type

Optional[QueueMessageType]

abstract get_many(timeout: float = DEFAULT_TIMEOUT, max: Optional[int] = None) → List[QueueMessageType]#
Parameters

  • timeout (float) –

  • max (Optional[int]) –

Return type

List[QueueMessageType]

abstract size() → int#
Return type

int

abstract close() → None#
Return type

None

class whylogs.api.logger.experimental.logger.actor.actor.CloseMessage#
class whylogs.api.logger.experimental.logger.actor.actor.QueueConfig#

Configuration for the queue used by the actor.

Parameters

  • max_batch_size (int) – Maximum number of messages to process in a single batch.

  • message_accumualtion_duration (float) – How long to accumulate messages before processing a batch

  • message_poll_wait (float) – How long to block when polling for messages in a loop.

max_batch_size: int = 50000#
message_accumualtion_duration: float = 1.0#
message_poll_wait: float = 0.1#
max_buffer_bytes: int = 100000000#
whylogs.api.logger.experimental.logger.actor.actor.MessageType#
class whylogs.api.logger.experimental.logger.actor.actor.Actor(queue_wrapper: QueueWrapper[Union[MessageType, CloseMessage]], queue_config: QueueConfig = QueueConfig())#

Bases: abc.ABC, Generic[MessageType]

Simple actor inspired interface designed for batch processing messages. This was created with thread and process based whylogs logging in mind and it has all of the common message passing logic needed.

Parameters
abstract close_message_handled() → bool#
Return type

bool

abstract set_close_message_handled() → None#
Return type

None

abstract close_message_wait() → None#
Return type

None

abstract is_done() → bool#
Return type

bool

abstract done_wait() → None#
Return type

None

abstract set_done() → None#

Set this actor as done, meaning it has finished processing all messages.

Return type

None

abstract is_closed() → bool#
Return type

bool

abstract set_closed() → None#

Sets this actor as closed, meaning it should no longer accept messages.

Return type

None

abstract process_batch(batch: List[Union[MessageType, CloseMessage]], batch_type: Type[Union[MessageType, CloseMessage]]) → None#
Parameters
Return type

None

send(message: Union[MessageType, CloseMessage]) → None#
Parameters

message (Union[MessageType, CloseMessage]) –

Return type

None

send_many(messages: List[Union[MessageType, CloseMessage]]) → None#
Parameters

messages (List[Union[MessageType, CloseMessage]]) –

Return type

None

close() → None#
Return type

None

process_messages() → None#
Return type

None

run() → None#
Return type

None