ten8t package#
Submodules#
ten8t.rule_pathlib module#
ten8t.rule_fs module#
ten8t.rule_ndf module#
ten8t.rule_pdf module#
ten8t.rule_ping module#
ten8t.rule_sqlachemy module#
ten8t.rule_webapi module#
ten8t.rule_xlsx module#
ten8t.ten8t_attribute module#
ten8t.ten8t_checker module#
ten8t.ten8t_exception module#
Basic exception classes for ten8t.
- exception ten8t.ten8t_exception.Ten8tException[source]#
Bases:
ExceptionSpecialized exception for ten8t.
- exception ten8t.ten8t_exception.Ten8tTypeError[source]#
Bases:
TypeErrorType errors associated with setting up ten8t
When bad types are sent to ten8t, this exception will be raised and should not be confused the TypeError that (should) indicate an unexpected lower level error.
- exception ten8t.ten8t_exception.Ten8tValueError[source]#
Bases:
ValueErrorValue Error associated with setting up ten8t
These exceptions will occur when setting up parameters on ten8t attributes and basic setup. For example a negative weight.
ten8t.ten8t_function module#
ten8t.ten8t_immutable module#
ten8t.ten8t_logging module#
Logging support for Ten8t.
The purpose of this module is to set up package logging in a general way that integrates with other packages and loggers.
The key feature is that you can call ten8t_setup_logging AFTER you have configured other packages’ loggers’ and it will just work along with them.
If the only logging in the system is set up with this tool then you can still use this and provide a target stream (e.g., stdout) and a file name.
This method is set up to manage a single global logger object named ten8t_logger that should be used by all the ten8t modules.
- ten8t.ten8t_logging.ten8t_reset_logging()[source]#
Reset the logger to its initial state by removing all handlers and reinstating a NullHandler.
- NOTE: This function is intended for testing purposes ONLY.
Using this in production code may disrupt logging configurations.
- ten8t.ten8t_logging.ten8t_setup_logging(level: int = 30, propagate: bool = True, file_name: str | None = None, stream_=None, format_string: str = '%(asctime)s - %(name)s - %(levelname)s - %(message)s') None[source]#
Configure the global logger for the ‘ten8t’ package.
- Parameters:
level (int) – Logging level (default=logging.WARNING).
propagate (bool) – If True, propagates logs to the parent logger (default=True).
file_name (str) – Optional file path to save logs.
stream (io.TextIOWrapper | None) – Optional stream (e.g., sys.stdout, sys.stderr).
format_string (str) – Format string for log messages (default:time, name, level, and msg).
- Returns:
Configures the global ten8t_logger.
- Return type:
None
ten8t.ten8t_module module#
ten8t.ten8t_package module#
ten8t.ten8t_rc_factory module#
ten8t.ten8t_result module#
This module contains the Ten8tResult class and some common result transformers.
- ten8t.ten8t_result.TR#
alias of
Ten8tResult
- class ten8t.ten8t_result.Ten8tResult(status: bool | None = False, func_name: str = '', pkg_name: str = '', module_name: str = '', msg: str = '', msg_rendered: str = '', msg_text: str = '', info_msg: str = '', info_msg_rendered: str = '', info_msg_text: str = '', warn_msg: str = '', warn_msg_rendered: str = '', warn_msg_text: str = '', doc: str = '', runtime_sec: float = 0.0, except_: Exception | None = None, traceback: str = '', skipped: bool = False, weight: float = 100.0, tag: str = '', level: int = 1, phase: str = '', count: int = 0, ruid: str = '', ttl_minutes: float = 0.0, mit_msg: str = '', owner_list: list[str] = <factory>, attempts: int = 1, skip_on_none: bool = False, fail_on_none: bool = False, summary_result: bool = False, thread_id: str = '', cached: bool = False)[source]#
Bases:
objectRepresents the outcome and metadata of a Ten8tFunction execution.
This class is designed to encapsulate detailed information about the execution of a function or rule in the Ten8t framework, including its status, runtime details, messages, and various metadata. It provides a structured way to handle execution results while optimizing memory and performance through the use of __slots__.
### Key Features: - Captures detailed results, including status, messages (info, warnings, errors),
runtime, and exceptions.
Supports efficient memory use and faster attribute access by enabling __slots__. This is especially useful when managing large numbers of result objects.
Prevents unintended dynamic attribute assignment, ensuring predictable behavior.
Provides utility methods like as_dict() for easy serialization and integration.
### Why Use __slots__? - Memory Efficiency:
By defining a fixed set of attributes, memory overhead from per-instance dictionaries (__dict__) is avoided.
Performance Improvements: Attribute access and assignment are faster due to reduced indirection overhead.
Error Prevention: Ensures that only predefined attributes can be added, reducing potential runtime mistakes.
While __slots__ is beneficial here, it limits dynamic attribute addition. If the need arises for dynamic attributes in the future, __slots__ can be disabled by removing slots=True from the @dataclass.
- ### Attributes:
status (bool | None): Indicates the execution status (pass, fail, or None). func_name (str): The name of the executed function. pkg_name (str): Package where the function resides. module_name (str): Module name where the function resides. msg (str): User-facing message summarizing the result. info_msg (str): Informational message about the execution. warn_msg (str): Warning message, if applicable. doc (str): Docstring of the executed function or rule. runtime_sec (float): The runtime of the function in seconds. except_ (Exception | None): Exception raised during execution, if any. traceback (str): String representation of the exception traceback. skipped (bool): Indicates whether the execution was skipped. weight (float): Relative importance or weight assigned to the result. tag (str): Descriptive tag for grouping results (for example, HR, Ops, Engr, Production) level (int): Numerical category for ordering results. phase (str): “Phase” category, perhaps (r&d,proto,production). count (int): Number of return values or individual steps. ruid (str): Unique identifier for this result. ttl_minutes (float): Time-to-live duration for the result, in minutes. mit_msg (str): Mitigation suggestion or message, if applicable. owner_list (list[str]): A list of owners or responsible parties. skip_on_none (bool): Whether to skip function execution if encountering None. fail_on_none (bool): Whether to fail function execution if encountering None. summary_result (bool): Indicates this result is a summary of multiple outcomes. thread_id (str): The identifier of the thread where this function ran.
- ### Methods:
- as_dict() -> dict:
Converts the Ten8tResult instance into a dictionary representation for serialization or hashing purposes.
- ### Usage Example:
```python # Example instantiation result = Ten8tResult(
status=True, func_name=”my_function”, msg=”Function executed successfully”, runtime_sec=0.45,
)
# Accessing attributes print(result.status) # True print(result.runtime_sec) # 0.45
# Converting to dictionary result_dict = result.as_dict() print(result_dict) ```
- classmethod from_dict(data: dict) Ten8tResult[source]#
Create a Ten8tResult instance from a dictionary (that presumably was saved with as_dict())
- Parameters:
data (dict) – Dictionary representation of a Ten8tResult instance.
- Returns:
A new Ten8tResult instance populated with values from the dictionary.
- Return type:
- mu = <ten8t.render._markup.Ten8tMarkup object>#
- class ten8t.ten8t_result.Ten8tResultDictFilter(ruid_patterns: Sequence[str] | str | None = None, tag_patterns: Sequence[str] | str | None = None, phase_patterns: Sequence[str] | str | None = None, func_name_patterns: Sequence[str] | str | None = None, summary_results: bool = None, status_results: bool = None)[source]#
Bases:
objectAdvanced filter capable of filtering result dictionary on multiple fields (ruid, tag, and phase).
- filter(results: dict, ruid_patterns: Sequence[str] | str | None = None, tag_patterns: Sequence[str] | str | None = None, phase_patterns: Sequence[str] | str | None = None, func_name_patterns: Sequence[str] | str | None = None, summary_results: bool = None, status_results: bool = None) dict[source]#
Filters data on ruid, tag, and phase fields using the provided patterns.
- ten8t.ten8t_result.fails_only(sr: Ten8tResult)[source]#
Filters out successful results.
- Parameters:
sr (Ten8tResult) – The result to check.
- Returns:
The result if it has failed, otherwise None.
- Return type:
- ten8t.ten8t_result.group_by(results: Sequence[Ten8tResult], keys: Sequence[str]) dict[str, Any][source]#
Groups a list of Ten8tResult by a list of keys.
This function allows for arbitrary grouping of Ten8tResult using the keys of the Ten8tResult as the grouping criteria. You can group in any order or depth with any number of keys.
- Parameters:
results (Sequence[Ten8tResult]) – The list of results to group.
keys (Sequence[str]) – The list of keys to group by.S
- ten8t.ten8t_result.overview(results: list[Ten8tResult]) str[source]#
Returns an overview of the results.
- Parameters:
results (list[Ten8tResult]) – The list of results to summarize.
- Returns:
A summary of the results.
- Return type:
- ten8t.ten8t_result.passes_only(sr: Ten8tResult)[source]#
Return only results that have pass status
- ten8t.ten8t_result.remove_info(sr: Ten8tResult)[source]#
Filter out messages tagged as informational
- Parameters:
sr (Ten8tResult) – The result to check.
- Returns:
The result if it has failed, otherwise None.
- Return type:
- ten8t.ten8t_result.results_as_dict(results: list[Ten8tResult])[source]#
Converts a list of Ten8tResult to a list of dictionaries.
- Parameters:
results (list[Ten8tResult]) – The list of results to convert.
- Returns:
The list of dictionaries.
- Return type:
list[Dict]
- ten8t.ten8t_result.warn_as_fail(sr: Ten8tResult)[source]#
Treats results with a warning message as failures.
- Parameters:
sr (Ten8tResult) – The result to check.
- Returns:
The result with its status set to False if there’s a warning message.
- Return type:
ten8t.ten8t_ruid module#
ten8t.ten8t_thread module#
ten8t.ten8t_util module#
This is the sad place for lonely functions that don’t have a place
- ten8t.ten8t_util.FloatListOrNone#
Type alias for a sequence of floats or None.
- ten8t.ten8t_util.IntListOrNone#
Type alias for a sequence of integers or None.
- class ten8t.ten8t_util.NextIntValue[source]#
Bases:
objectI had to create this class in order to make mypy happy. Mypy does not know how to handle dynamic functions and playing games
- ten8t.ten8t_util.StrListOrNone#
Type alias for a sequence of strings or None.
- ten8t.ten8t_util.any_to_int_list(param: Sequence[int] | int | None, sep=' ') Sequence[int][source]#
Convert a string to a list of integers or if a list is given make sure it is all integers. :param param: list of integers or string to convert to list of integers :param sep: separator character.
- Returns:
list of integers
- ten8t.ten8t_util.any_to_path_list(param: Sequence[str | Path] | None, sep=' ') Sequence[Path][source]#
Flexibly take a list of strings are pathlib objects and make a uniform list of pathlib objects. This is useful for normalizing data read from different sources without have a bunch of point of use parsing.
The assumption is that this data could come from a config file, a command line parameter, a UI element that returns strings, or code. This should make all code just “fix” the data with this call.
- Parameters:
param – StrOrPathListOrNone Data to normalize
sep – Separator character. Should almost always be ‘ ‘
Returns:
- ten8t.ten8t_util.any_to_str_list(param: Sequence[str] | str | None, sep=' ') Sequence[str][source]#
Convert a string to a list of strings or if a list is given make sure it is all strings. :param param: list of strings or string to convert to list of strings :param sep: separator character.
Returns:
- ten8t.ten8t_util.clean_dict(d: dict, remove_nulls=True, keep_keys=None, remove_keys=None, empty_values=None)[source]#
Recursively processes a dictionary to remove keys based on specified rules.
- Parameters:
d (dict) – The input dictionary to process.
remove_nulls (bool, optional) – If True, removes keys with “empty” values (as defined in empty_values), unless the keys are in keep_keys. Defaults to True.
keep_keys (list[str], optional) – A list of keys to preserve, ensuring they are not removed, even if their values are defined as empty. Defaults to None.
remove_keys (list[str], optional) – A list of keys to forcibly remove, regardless of their values. Defaults to None.
empty_values (list, optional) – A list of values that should be considered empty. By default, this is [‘’, [], {}].
- Returns:
- A new dictionary where keys are removed based on remove_nulls, keep_keys,
remove_keys, and empty_values.
- Return type:
- Raises:
Ten8tTypeError – If the input is not a dictionary.
Ten8tValueError – If there are conflicts between keep_keys and remove_keys.
- ten8t.ten8t_util.cwd_here(file_: str | Path | None = None) Path[source]#
Change current working directory to the provided folder or to the parent folder if a file is provided.
- If ‘file_’ is:
None: the current file (__file__) location is used to set cwd.
a directory: the provided directory itself becomes the current working directory.
a file: the parent directory containing that file becomes the current working directory.
Why? This is useful for running scripts in different locations without having to worry about setting the current working directy. Simplifies setup of demos and examples and allows you to NOT set the CWD in a configuration launch files.
python foo/fum/demo.py
And have code call cwd_here() and know that relative paths for demo.py will be correct.
- Parameters:
file (str | pathlib.Path | None) – The path to a file or directory, or None. Default is None.
- Returns:
Explicitly returns path of the new current working directory.
- Return type:
Example usage:#
>>> cwd_here('/path/to/directory') # sets cwd to '/path/to/directory'
>>> cwd_here('/path/to/file.txt') # sets cwd to '/path/to/'
>>> cwd_here() # sets the cwd to currently running script (__FILE__)