Base Modules

Base module for handling the configuration for a given Action Trigger.

Note: This is not used in the current implementation. It is a work in progress and a placeholder for future work.

class action_triggers.base.config.ActionTriggerActionBase(key: str, conn_details: dict | None, params: dict | None, **kwargs)

Bases: ABC

Base class for triggering an action. This class should be subclassed to implement the specific action for an action trigger.

Parameters:

• key – The key for the action trigger (must exist in the settings.ACTION_TRIGGERS[self.action_trigger_type] dictionary).

• conn_details – The connection parameters to use for establishing the connection.

• params – Additional parameters to use for the action trigger.

• kwargs – Additional keyword arguments to pass to the subclass.

abstract property action_trigger_type: ActionTriggerType

The type of action trigger.

abstract property conn_class: Type[ConnectionBase]

The connection class to use for establishing a connection to the service which the action will interact with.

async send_message(message: str) → None

Starts a connection with the message broker and sends a message.

Parameters:

message – The message to send to the message broker.

class action_triggers.base.config.ConnectionBase(config: dict, conn_details: dict, params: dict)

Bases: ABC

Base class for establishing a connection. This class provides the capability to marry the configuration provided in the settings with the connection details and parameters provided by the user. However, this enforces a one-sided relationship where the user cannot overwrite the base configuration as defined in the settings as the base configuration always takes precedence.

Parameters:

• config – The configuration as defined in settings.ACTION_TRIGGERS for a given action trigger type.

• conn_details – Additional connection parameters to use for establishing the connection provided by the user.

• params – Additional parameters to use for the action trigger provided by the user.

abstract async close() → None

Close the connection to the message broker.

property conn_details: dict

Lazy load the the merged connection details. When merging the connection details, the user provided base connection details take precedence over the base configuration.

Returns:

The merged connection details.

abstract async connect() → None

Establish a connection to the message broker.

abstract property error_class: Type[ErrorBase]

The error class to use for storing errors.

property params: dict

Lazy load the the merged parameters. When merging the parameters, the user provided base parameters take precedence over the base configuration.

Returns:

The merged parameters.

abstract property required_conn_detail_fields: Sequence[RequiredFieldBase]

The required connection detail fields that must be provided by the user.

abstract property required_params_fields: Sequence[RequiredFieldBase]

The required parameters fields that must be provided by the user.

Base module containing the logic for handling of bespoke error classes. Note: An error class is different from an Exception class. An Exception class is for handling exceptions and they can be raised. This module contains what are fundamentally classes for storing and managing error messages.

class action_triggers.base.error.ErrorBase

Bases: object

A base class for storing errors for a set of fields. For each field, an error can be added using the add_<field_name>_error method.

Parameters:

error_class – The associated Exception class to raise when there are errors. (default: Exception)

Example:

class MyError(ErrorBase):
    field_1 = ErrorField()
    field_2 = ErrorField()

error = MyError()
error.add_field_1_error("key_1", "message_1")
error.add_field_1_error("key_1", "message_2")
error.add_field_2_error("key_2", "message_3")

error.as_dict()

# Output:

{
    "field_1": {
        "key_1": ["message_1", "message_2"]
    },
    "field_2": {
        "key_2": ["message_3"]
    }
}

as_dict() → Dict[str, Dict[str, List[str]]]

Convert the errors to a dictionary.

Returns:

The errors as a dictionary.

error_class

alias of Exception

is_valid(raise_exception: bool = False) → bool

Check if the error is valid.

Parameters:

raise_exception – Whether to raise an exception if raise_exception is True and there are errors.

Returns:

True if the error is valid, False otherwise.

class action_triggers.base.error.MetaError(name, bases, dct)

Bases: type

A metaclass for the Error class. This metaclass generates and attaches methods to aid in adding errors to the fields of the class.

static make_add_error(field_name: str) → Callable[[Any, str, str], None]

Generate a method for adding an error to a field.

Parameters:

field_name – The name of the field to add the error to.

Returns:

A method for adding an error to a field.