aea.helpers.base
Miscellaneous helpers.
yaml_load
@_ordered_loading
yaml_load(*args, **kwargs) -> Dict[str, Any]
Load a yaml from a file pointer in an ordered way.
Returns:
the yaml
yaml_load_all
@_ordered_loading
yaml_load_all(*args, **kwargs) -> List[Dict[str, Any]]
Load a multi-paged yaml from a file pointer in an ordered way.
Returns:
the yaml
yaml_dump
@_ordered_dumping
yaml_dump(*args, **kwargs) -> None
Dump multi-paged yaml data to a yaml file in an ordered way.
:return None
yaml_dump_all
@_ordered_dumping
yaml_dump_all(*args, **kwargs) -> None
Dump multi-paged yaml data to a yaml file in an ordered way.
:return None
locate
locate(path: str) -> Any
Locate an object by name or dotted path, importing as necessary.
load_module
load_module(dotted_path: str, filepath: Path) -> types.ModuleType
Load a module.
Arguments:
dotted_path: the dotted path of the package/module.filepath: the file to the package/module.
Returns:
None
Raises:
ValueError: if the filepath provided is not a module.Exception: if the execution of the module raises exception.
load_env_file
load_env_file(env_file: str)
Load the content of the environment file into the process environment.
Arguments:
env_file: path to the env file.
Returns:
None.
sigint_crossplatform
sigint_crossplatform(process: subprocess.Popen) -> None
Send a SIGINT, cross-platform.
The reason is because the subprocess module doesn't have an API to send a SIGINT-like signal both on Posix and Windows with a single method.
However, a subprocess.Popen class has the method 'send_signal' that gives more flexibility in this terms.
Arguments:
process: the process to send the signal to.
Returns:
None
win_popen_kwargs
win_popen_kwargs() -> dict
Return kwargs to start a process in windows with new process group.
Help to handle ctrl c properly. Return empty dict if platform is not win32
send_control_c
send_control_c(process: subprocess.Popen, kill_group: bool = False) -> None
Send ctrl-C crossplatform to terminate a subprocess.
Arguments:
process: the process to send the signal to.
Returns:
None
RegexConstrainedString Objects
class RegexConstrainedString(UserString)
A string that is constrained by a regex.
The default behaviour is to match anything. Subclass this class and change the 'REGEX' class attribute to implement a different behaviour.
__init__
| __init__(seq)
Initialize a regex constrained string.
cd
@contextlib.contextmanager
cd(path)
Change working directory temporarily.
get_logger_method
get_logger_method(fn: Callable, logger_method: Union[str, Callable]) -> Callable
Get logger method for function.
Get logger in fn definion module or creates logger is module.name.
Or return logger_method if it's callable.
Arguments:
fn: function to get logger for.logger_method: logger name or callable.
Returns:
callable to write log with
try_decorator
try_decorator(error_message: str, default_return=None, logger_method="error")
Run function, log and return default value on exception.
Does not support async or coroutines!
Arguments:
error_message: message template with one{}for exceptiondefault_return: value to return on exception, by default Nonelogger_method: name of the logger method or callable to print logs
MaxRetriesError Objects
class MaxRetriesError(Exception)
Exception for retry decorator.
retry_decorator
retry_decorator(number_of_retries: int, error_message: str, delay: float = 0, logger_method="error")
Run function with several attempts.
Does not support async or coroutines!
Arguments:
number_of_retries: amount of attemptserror_message: message template with one{}for exceptiondelay: num of seconds to sleep between retries. default 0logger_method: name of the logger method or callable to print logs
exception_log_and_reraise
@contextlib.contextmanager
exception_log_and_reraise(log_method: Callable, message: str)
Run code in context to log and re raise exception.
Arguments:
log_method: function to print logmessage: message template to add error text.