Skip to content

aea.helpers.base

Miscellaneous helpers.

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.

SimpleId Objects

class SimpleId(RegexConstrainedString)

A simple identifier.

The allowed strings are all the strings that: - have at least length 1 - have at most length 128 - the first character must be between a-z,A-Z or underscore - the other characters must be either the above or digits.

Examples of allowed strings:

SimpleId("an_identifier") 'an_identifier'

Examples of not allowed strings:

SimpleId("0an_identifier") Traceback (most recent call last): ... ValueError: Value 0an_identifier does not match the regular expression re.compile('[a-zA-Z_][a-zA-Z0-9_]{0,127}')

SimpleId("") Traceback (most recent call last): ... ValueError: Value does not match the regular expression re.compile('[a-zA-Z_][a-zA-Z0-9_]{0,127}')

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 exception
  • default_return: value to return on exception, by default None
  • logger_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 attempts
  • error_message: message template with one {} for exception
  • delay: num of seconds to sleep between retries. default 0
  • logger_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 log
  • message: message template to add error text.

recursive_update

recursive_update(to_update: Dict, new_values: Dict) -> None

Update a dictionary by replacing conflicts with the new values.

It does side-effects to the first dictionary.

to_update = dict(a=1, b=2, subdict=dict(subfield1=1)) new_values = dict(b=3, subdict=dict(subfield1=2)) recursive_update(to_update, new_values) to_update {'a': 1, 'b': 3, 'subdict': {'subfield1': 2}}

Arguments:

  • to_update: the dictionary to update.
  • new_values: the dictionary of new values to replace.

Returns:

None

find_topological_order

find_topological_order(adjacency_list: Dict[T, Set[T]]) -> List[T]

Compute the topological order of a graph (using Kahn's algorithm).

Arguments:

  • adjacency_list: the adjacency list of the graph.

Returns:

the topological order for the graph (as a sequence of nodes)

Raises:

  • ValueError: if the graph contains a cycle.

reachable_nodes

reachable_nodes(adjacency_list: Dict[T, Set[T]], starting_nodes: Set[T]) -> Dict[T, Set[T]]

Find the reachable subgraph induced by a set of starting nodes.

Arguments:

  • adjacency_list: the adjacency list of the full graph.
  • starting_nodes: the starting nodes of the new graph.

Returns:

the adjacency list of the subgraph.