message module

Classes for capturing messages from controls and other sources like REST APIs.

Messages can be appended to an instance of MessageBuffer before actually printing them to the console and/or a log file. An individual message is captured with multiple attributes, not just with its human-readable text. This has two benefits:

  • Messages can be translated into another language in one go and on-demand. See addTranslations().

  • Message semantics can be queried programatically, e.g. by checking for a certain message type via hasKey(). Another example is checking for a certain message severity like ERROR via hasSeverity().

Classes

class ctlbase.message.Message(textEn, key, arguments=(), severity: Severity = Severity.NORMAL, timestamp_iso=None, ns=None, correlationId=None)

A message captured with several attributes.

__eq__(other)

Compares this message to another object by all attributes except the str values of the message texts.

__init__(textEn, key, arguments=(), severity: Severity = Severity.NORMAL, timestamp_iso=None, ns=None, correlationId=None)

Creates a message.

Parameters:
static fromDict(msgDict)

Creates a Message instance from a dict.

The dict must correspond to the JSON representation defined via fromJson() and toJson(). Hence, if these two methods are overridden, fromDict() must be overridden, too.

Parameters:

msgDict (dict) – A message in its dict representation.

Returns:

The created message.

Return type:

Message

static fromJson(jsonStr)

Creates a Message instance from a JSON str.

It may be overridden to deserialize a message from a custom JSON representation. In this case, toJson() and fromDict() must be overridden, too.

Parameters:

jsonStr (str) – A message in its JSON representation, enclosed in { ... }.

Returns:

The created message.

Return type:

Message

getTextLngs()

Determines the languages for which message texts are already rendered and available.

Message texts for additional languages can be added by calling addTranslations(). For that purpose, the message must be appended to a message buffer.

Returns:

A sequence of two-letter language codes for the available message texts.

Return type:

list[str]

toColoredText(lng='en')

A variant of toText() that encloses the message text in ANSI control characters coloring the text according to severity.

toJson()

Represents this message as a JSON str.

It may be overridden to serialize the message to a custom JSON representation. In this case, fromJson() and fromDict() must be overridden, too.

The following is an example for the default JSON representation:

{
    "NAMESPACE": "SCREEN",
    "KEY": "SET_MODE_SUCCEEDED",
    "ARGUMENTS": [
        "PREVENT_DOWN",
        30
    ],
    "SEVERITY": "SUCCESS",
    "TEXT_EN": "Set mode prevent-down for 30 seconds.",
    "TEXT_DE": "Modus prevent-down für 30 Sekunden gesetzt."
}
Returns:

The message in its JSON representation, enclosed in { ... }.

Return type:

str

toText(lng='en')

Looks up the message text in a certain language. It must already be available as a message attribute, as this method does not perform any on-demand translation.

Parameters:

lng (str) – Two-letter language code for the message text to be returned.

Returns:

The message text for language lng, if available. Otherwise, the English message text is returned as a fallback.

Return type:

str

arguments: list | tuple | dict

The message arguments, i.e. language-neutral literals that can be embedded in the message texts of various languages. Examples:

  • an int like 5 for the number of failed retries

  • a str representing a user-supplied value, e.g. /tmp/command.log for a log file path

  • a float like 1.532 representing the elapsed time in seconds

If the arguments are specified as a dict instead of an ordered list or tuple, the dict keys must be semantic identifiers for the arguments. Examples:

  • retryCount for the number of failed retries

  • log for a log file path

  • elapsed for the elapsed time

correlationId: str | None

An optional correlation ID, i.e. an alphanumeric identifier used to group several messages. As an example, it can be an identifier for a batch command that produces several messages across several message namespaces.

key: str
An alphanumeric key for the message semantics. It is used to check for messages that have occurred, as well as a key into translation files. It must be unique within the message namespace.
Example: FILE_NOT_FOUND
ns: str
An alphanumeric identifier for the message namespace within which all message keys must be unique. It usually represents the application or domain from which the message originates.
If None, when appending the message to a message buffer, the default namespace of the message buffer is assumed. See nsDefault.
Examples: HOME_THEATER or ENGINE_CONTROL
severity: Severity

The message severity.

textEn: str

The human-readable message text in English (en). It serves multiple purposes:

  • At runtime: If the message is to be shown to a user whose preferred language is en, the message text is used “as is”.

  • If the user has a different preferred language, the English message text is used as a fallback if no appropriate translation is available.

  • At development time: If the English message text occurs as a literal str in the source code, it serves as an additional documentation of the source code, e.g. a literal error message for a specific case.

  • Also, it serves as the primary reference when translating the message text into other languages.

timestamp_iso: str | datetime.datetime | None
An optional timestamp in ISO 8601 format or an instance of datetime.datetime directly. It should represent the point in time when the message occurred originally.
Example: 2024-08-01T14:38:32.499588
class ctlbase.message.ErrorMessage(textEn, key, arguments=(), timestamp_iso=None, ns=None, correlationId=None, exitCode=None)

Bases: Exception, Message

A variant of Message that works like an Exception.

It has an implicit severity of ERROR and can be “raised”.

Attributes with the same name have the same meaning as with Message.

__init__(textEn, key, arguments=(), timestamp_iso=None, ns=None, correlationId=None, exitCode=None)

Creates an error message.

Parameters:
exitCode: int | None

An optional numeric code to be used as exit code of the process. See getExitCode().

class ctlbase.message.MessageBuffer(nsDefault, lngDefault='en', isColoringDefault=None)

A buffer for messages.

Messages can be appended without writing them to an output stream directly. Appended messages can be queried programatically until the buffer is cleared via clear(). For instance, to check for a certain message severity like ERROR, use the hasSeverity() method.

To eventually print the messages to stdout, stderr or another output stream, use the printText() method.

The COLOR_PRINT environment variable determines the default behavior for coloring in text output functions if isColoringDefault was initialized as None. The allowed values for the environment variable are true and false.

__init__(nsDefault, lngDefault='en', isColoringDefault=None)

Creates a new message buffer.

Parameters:

  • nsDefault (str) – See nsDefault.

  • lngDefault (str) – See lngDefault.

  • isColoringDefault (bool | None) – If True, text is colored by default when printed to stdout or stederr. If None, a possibly existing COLOR_PRINT_ENVKEY environment value is used. If the latter does not exist, a default value of True is assumed.

addTranslations(isOkLoadError=True, isOkKeyError=True, isOkInterpolationError=False)

Adds translations for the message texts already in the buffer.

The target language is set in the global scope of the Python process via changeLanguage().

Parameters:

  • isOkLoadError (bool) – If False, a message with key TRANSLATION_LOAD_FAILED and severity WARNING is appended to the buffer if a translation file for the target language cannot be loaded. If True, the message is appended with severity DEBUG.

  • isOkKeyError (bool) – If False, a message with key TRANSLATION_KEY_NOT_FOUND and severity WARNING is appended to the buffer if a specific translation key cannot be found. If True, the message is appended with severity DEBUG.

  • isOkInterpolationError – If False, a message with key TRANSLATION_INTERPOLATION_FAILED and severity WARNING is appended to the buffer if the value for a variable cannot be resolved. If True, the message is appended with severity DEBUG.

append(msg, key=None, arguments=(), severity: Severity = Severity.NORMAL, timestamp_iso=None, ns=None, correlationId=None, isDeduplicate=True)

Appends a message to the buffer. The message may be given in one of two forms:

  1. As an instance of Message or ErrorMessage, provided in msg. Example:

    msg = Message("Configuration file not found", 'CONFIG_NOT_FOUND', severity=Severity.ERROR)
msgBuffer.append(msg)

  2. As individual attributes, of which the English message text must be provided in msg. What is more, key must be provided. All other attributes are optional and assume default values if not provided. Example:

    msgBuffer.append("Configuration file not found", 'CONFIG_NOT_FOUND', severity=Severity.ERROR)

Regardless of the chosen form, if the message namespace is None, the value of nsDefault is used.

Parameters:

  • msg (Message | ErrorMessage | str) – The message to append, depending on the chosen form (see above).

  • key (str) – Only relevant if form 2 is chosen. See key.

  • arguments (list | tuple | dict) – Only relevant if form 2 is chosen. See arguments.

  • severity (Severity) – Only relevant if form 2 is chosen. See severity.

  • timestamp_iso (str) – Only relevant if form 2 is chosen. See timestamp_iso.

  • ns (str) – Only relevant if form 2 is chosen. See ns.

  • correlationId (str | None) – Only relevant if form 2 is chosen. See correlationId.

  • isDeduplicate (bool) – If True, appends the message only if it is not yet contained in the buffer, according to __eq__().

appendBuffer(otherBuffer, correlationIdFilter=None)

Appends the messages from another buffer to this buffer.

Parameters:

  • otherBuffer (MessageBuffer) – The source buffer from which to append messages.

  • correlationIdFilter (str | None) – An optional filter for messages by correlation ID. See correlationIdFilter.

appendJson(jsonStr, correlationIdFilter=None)

Parses the messages in a JSON str and appends them to the buffer.

Helpful if messages from another control or REST service were retrieved as JSON and need to be appended to the buffer.

The supported JSON representation is defined by fromJson(). Multiple messages must be enclosed in [ ... ].

Parameters:

  • jsonStr (str) – The source JSON str from which to append messages.

  • correlationIdFilter (str | None) – An optional filter for messages by correlation ID. See correlationIdFilter.

clear()

Clears all messages in the buffer.

The default values of the buffer are retained.

getExitCode()

Determines the most appropriate numeric exit code for the process, based on the messages in the buffer.

If instances of ErrorMessage were appended to this buffer, the exitCode of the instance appended last is returned.
Otherwise, the exit code depends on the message in the buffer with the “highest” severity:
4 if at least one message has the severity ERROR.
3 if at least one message has the severity WARNING.
0 otherwise.
Returns:

The numeric exit code.

Return type:

int

getJson(correlationIdFilter=None)

Creates a formatted JSON str from the messages in the buffer.

The supported JSON representation is defined by toJson().

Parameters:

correlationIdFilter (str | None) – An optional filter for messages by correlation ID. See correlationIdFilter.

Returns:

The messages in their JSON representation, as an ordered list enclosed in [ ... ].

Return type:

str

getList(correlationIdFilter=None)

Creates an ordered sequence of message objects in this buffer.

Parameters:

correlationIdFilter (str | None) – An optional filter for messages by correlation ID. See correlationIdFilter.

Returns:

An ordered sequence of message objects.

Return type:

list[Message]

getText(lngForText=None, severityFilter=(Severity.NORMAL, Severity.SUCCESS, Severity.WARNING, Severity.ERROR), correlationIdFilter=None, isColoring=None)

Concatenates the message texts in the buffer into one str.

Parameters:

  • lngForText (str) – Two-letter language code for the message texts to be concatenated. If None, the value of lngDefault is used. Defaults to en if a corresponding translation cannot be found.

  • severityFilter (Iterable | None) – An optional filter for messages by severity. See severityFilter.

  • correlationIdFilter (str | None) – An optional filter for messages by correlation ID. See correlationIdFilter.

  • isColoring (bool | None) – If True, the returned text is colored according to message severity. If None, the value of isColoringDefault is used.

Returns:

The concatenated str, with individual messages separated by newline characters.

Return type:

str

hasKey(key, argumentFilter=None, correlationId=None)

Searches the buffer for a message with a certain key.

Parameters:
Returns:

True if at least one message has been found.

Return type:

bool

hasSeverity(severities, argumentFilter=None, correlationId=None)

Searches the buffer for a message with one of several possible severities.

Parameters:

  • severities (list | tuple) – One or more severities to search for. It is sufficient for a message to match one of the given severities.

  • argumentFilter (object | dict | None) – An optional filter for messages by argument. See argumentFilter.

  • correlationIdFilter (str | None) – An optional filter for messages by correlation ID. See correlationIdFilter.

Returns:

True if at least one message has been found.

Return type:

bool

printText(lngForText=None, severityFilter=(Severity.NORMAL, Severity.SUCCESS, Severity.WARNING, Severity.ERROR), correlationIdFilter=None, isColoring=None, stdStream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, errStream=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>)

Writes the messages in the buffer to an output stream as plain text. The messages are separated by newline characters.

Parameters:

  • lngForText (str) – Two-letter language code for the message texts to be printed. If None, the value of lngDefault is used. Defaults to en if a corresponding translation cannot be found.

  • severityFilter (Iterable | None) – An optional filter for messages by severity. See severityFilter.

  • correlationIdFilter (str | None) – An optional filter for messages by correlation ID. See correlationIdFilter.

  • isColoring (bool | None) – If True, the text is colored according to message severity. If None, the value of isColoringDefault is used.

  • stdStream (io.TextIOBase) – The output stream to which all messages are printed by default. The only exceptions are messages with severities WARNING and ERROR. These are printed to the errStream, if it is not None.

  • errStream (io.TextIOBase | None) – An optional output stream to which messages with severities WARNING and ERROR are printed.

isColoringDefault

If True, colored text is colored by default when printed to stdout or stderr.

lngDefault: str

Two-letter language code to be used as default when printing text to stdout or stderr.

nsDefault: str

The default namespace to be assigned to a message if its namespace is None.

class ctlbase.message.LoggingMessageBuffer(loop, nsDefault, lngDefault='en', isColoringDefault=None, logPath=None, firstColumnWidth=24, isLogImmediately=True, isLogToConsole=False)

Bases: MessageBuffer

A message buffer capable of logging messages to a file.

__init__(loop, nsDefault, lngDefault='en', isColoringDefault=None, logPath=None, firstColumnWidth=24, isLogImmediately=True, isLogToConsole=False)

Creates a new logging message buffer.

Parameters:
append(msg, key=None, arguments=(), severity: Severity = Severity.NORMAL, timestamp_iso=None, ns=None, correlationId=None, isDeduplicate=None)

Immediately writes a message to the log file if isLogImmediately is True and a a log file has been opened via openLog(). In this case, messages are not buffered. Hence, they cannot be queried programatically via hasKey() or hasSeverity() or any other method.

In all other cases, the call is delegated to the super method append().

close()

Closes the buffer, eventually writing all messages in the buffer to the log file if isLogImmediately is False.

async openLog(logPath=None, firstColumnWidth=None)

Actually opens the log file. If it exists already, log entries will be appended to it. Otherwise the file is created.

Parameters:

  • logPath (str) – The log file path to open and to set as the value of logPath. If None, the existing value of logPath is used.

  • firstColumnWidth (int) – The width of the first column in the log output, as a number of characters. If None, the value given in firstColumnWidth is retained.

Raises:

ValueError – If no log file path is given.

firstColumnFormatStr: str

The printf-style string used to format the first column in the log output.

The first column contains the timestamp of the corresponding log entry, right-padded with spaces to match the number of characters given in firstColumnWidth.

isLogImmediately: bool

If True, any message is written to the log file as soon as it is appended to the buffer. If False, all the messages in the buffer are written to the log file when the buffer is closed via its close() method.

isLogToConsole: bool

If True, messages are additionally printed to stdout at the same time when they are written to the log file.

logPath: str

The log file path. It is only opened when openLog() is called.

class ctlbase.message.MessageIterator(messages, format=Format.OBJECT, startIndex=0, isImmutable=False, argumentFilter=None, severityFilter=(Severity.NORMAL, Severity.SUCCESS, Severity.WARNING, Severity.ERROR), correlationIdFilter=None, lngForText='en')

An iterator over an ordered sequence of messages.

While iterating, messages may still be appended to the buffer.

class Format(value)

The output format of messages.

OBJECT = 0

Instances of Message.

TEXT = 2

The message texts as str.

TEXT_COLORED = 3

The message texts as str enclosed in ANSI control characters coloring the text according to severity.

__init__(messages, format=Format.OBJECT, startIndex=0, isImmutable=False, argumentFilter=None, severityFilter=(Severity.NORMAL, Severity.SUCCESS, Severity.WARNING, Severity.ERROR), correlationIdFilter=None, lngForText='en')
Parameters:

  • messages (MessageBuffer | list[Message] | tuple[Message]) – The messages to iterate over.

  • format (Format) – The output format of the messages.

  • startIndex (int) – The (zero-based) index of the first message to be included in the output. It must be less than len(messages). As an example, if startIndex is 2, iteration starts with the third message in messages.

  • isImmutable (bool) – If True, the iterator uses an immutable copy of the message sequence. In this case, messages appended afterwards are not reflected in the iterator.

  • argumentFilter (object | dict | None) – An optional filter for message arguments. The filter is specified as a scalar value, e.g. as a str, int or bool. The value must exactly match the value of one of a message’s arguments to be included in the output. Alternatively, the filter is specified as a dict with a single key/value pair. As an example, argumentFilter={'retryCount':5} matches message arguments specified as a dict, containing a semantic key retryCount with a value of 5.If None, messages are not filtered by arguments.

  • severityFilter (Iterable | None) – An optional filter for message severities. The filter is specified as an Iterable with one or more severities. At least one of them must match the severity of a message for it to be included in the output. If None, messages are not filtered by severity.

  • correlationIdFilter (str | None) – An optional filter for correlation IDs. The value must exactly match the correlation ID of a message for it to be included in the output. If None, messages are not filtered by correlation ID.

  • lngForText (str) – Two-letter language code for the message texts to be included in the output. Defaults to en, if a corressponding translation cannot be found. Only relevant for output formats TEXT and TEXT_COLORED.

Raises:

Constants and defaults

class ctlbase.message.Severity(value)

The severity of a single message.

The member values denote the ANSI color codes for corresponding messages printed to the console and/or a log file.

DEBUG = 90

Technical message intended for debugging.

ERROR = 91

Error message, e.g. for a failed action.

NORMAL = 39

Informative message with no special coloring.

SUCCESS = 92

Success message, e.g. for a completed action.

VERBOSE = 37

Verbose message providing more detail than NORMAL.

WARNING = 93

Warning message, e.g. for an action that completed, yet in an unexpected way.

ctlbase.message.SEVERITY_FILTER_DEFAULT = (Severity.NORMAL, Severity.SUCCESS, Severity.WARNING, Severity.ERROR)

Default filter for message severities when iterating over messages. The value is in effect for the entire Python process.

ctlbase.message.COLOR_PRINT_ENVKEY = 'COLOR_PRINT'

Environment configuration key for the coloring of text printed to stdout or stderr.