aea.helpers.base¶

Miscellaneous helpers.

locate¶

def locate(path: str) -> Any

Locate an object by name or dotted save_path, importing as necessary.

load`_`module¶

def load_module(dotted_path: str, filepath: Path) -> types.ModuleType

Load a module.

Arguments:

dotted_path: the dotted save_path of the package/module.
filepath: the file to the package/module.

Raises:

ValueError: if the filepath provided is not a module. # noqa: DAR402
Exception: if the execution of the module raises exception. # noqa: DAR402

Returns:

module type

load`_`env`_`file¶

def load_env_file(env_file: str) -> None

Load the content of the environment file into the process environment.

Arguments:

env_file: save_path to the env file.

sigint`_`crossplatform¶

def 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.

win`_`popen`_`kwargs¶

def 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

Returns:

windows popen kwargs

send`_`control`_`c¶

def send_control_c(process: subprocess.Popen,
                   kill_group: bool = False) -> None

Send ctrl-C cross-platform to terminate a subprocess.

Arguments:

process: the process to send the signal to.
kill_group: whether or not to kill group

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``¶

def __init__(seq: Union[UserString, str]) -> None

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("an identifier") Traceback (most recent call last): ... ValueError: Value an 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
def cd(path: PathLike) -> Generator

Change working directory temporarily.

get`_`logger`_`method¶

def get_logger_method(fn: Callable,
                      logger_method: Union[str, Callable]) -> Callable

Get logger method for function.

Get logger in fn definition 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¶

def try_decorator(error_message: str,
                  default_return: Callable = None,
                  logger_method: Any = "error") -> Callable

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

Returns:

the callable

MaxRetriesError Objects¶

class MaxRetriesError(Exception)

Exception for retry decorator.

retry`_`decorator¶

def retry_decorator(number_of_retries: int,
                    error_message: str,
                    delay: float = 0,
                    logger_method: str = "error") -> Callable

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: number of seconds to sleep between retries. default 0
logger_method: name of the logger method or callable to print logs

Returns:

the callable

exception`_`log`_`and`_`reraise¶

@contextlib.contextmanager
def exception_log_and_reraise(log_method: Callable, message: str) -> Generator

Run code in context to log and re raise exception.

Arguments:

log_method: function to print log
message: message template to add error text.

Returns:

the generator

recursive`_`update¶

def recursive_update(to_update: Dict,
                     new_values: Dict,
                     allow_new_values: bool = False) -> 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.
allow_new_values: whether or not to allow new values.

find`_`topological`_`order¶

def 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.

Raises:

ValueError: if the graph contains a cycle.

Returns:

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

reachable`_`nodes¶

def 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.

cached`_`property Objects¶

class cached_property()

Cached property from python3.8 functools.

``init``¶

def __init__(func: Callable) -> None

Init cached property.

`__`set`_`name`__`¶

def __set_name__(_: Any, name: Any) -> None

Set name.

``get``¶

def __get__(instance: Any, _: Optional[Any] = None) -> Any

Get instance.

ensure`_`dir¶

def ensure_dir(dir_path: str) -> None

Check if dir_path is a directory or create it.

dict`_`to`_`path`_`value¶

def dict_to_path_value(
        data: Mapping,
        path: Optional[List] = None) -> Iterable[Tuple[List[str], Any]]

Convert dict to sequence of terminal path build of keys and value.

parse`_`datetime`_`from`_`str¶

def parse_datetime_from_str(date_string: str) -> datetime.datetime

Parse datetime from string.

CertRequest Objects¶

class CertRequest()

Certificate request for proof of representation.

``init``¶

def __init__(public_key: str, identifier: SimpleIdOrStr,
             ledger_id: SimpleIdOrStr, not_before: str, not_after: str,
             message_format: str, save_path: str) -> None

Initialize the certificate request.

Arguments:

public_key: the public key, or the key id.
identifier: certificate identifier.
ledger_id: ledger identifier the request is referring to.
not_before: specify the lower bound for certificate validity. If it is a string, it must follow the format: 'YYYY-MM-DD'. It will be interpreted as timezone UTC-0.
not_after: specify the lower bound for certificate validity. If it is a string, it must follow the format: 'YYYY-MM-DD'. It will be interpreted as timezone UTC-0.
message_format: message format used for signing
save_path: the save_path where to save the certificate.

public`_`key¶

@property
def public_key() -> Optional[str]

Get the public key.

ledger`_`id¶

@property
def ledger_id() -> str

Get the ledger id.

key`_`identifier¶

@property
def key_identifier() -> Optional[str]

Get the key identifier.

identifier¶

@property
def identifier() -> str

Get the identifier.

not`_`before`_`string¶

@property
def not_before_string() -> str

Get the not_before field as string.

not`_`after`_`string¶

@property
def not_after_string() -> str

Get the not_after field as string.

not`_`before¶

@property
def not_before() -> datetime.datetime

Get the not_before field.

not`_`after¶

@property
def not_after() -> datetime.datetime

Get the not_after field.

message`_`format¶

@property
def message_format() -> str

Get the message format.

save`_`path¶

@property
def save_path() -> Path

Get the save path for the certificate.

Note: if the path is not absolute, then the actual save path might depend on the context.

Returns:

the save path

get`_`absolute`_`save`_`path¶

def get_absolute_save_path(path_prefix: Optional[PathLike] = None) -> Path

Get the absolute save path.

If save_path is an absolute path, then the prefix is ignored. Otherwise, the path prefix is prepended.

Arguments:

path_prefix: the (absolute) path to prepend to the save path.

Returns:

the actual save path.

public`_`key`_`or`_`identifier¶

@property
def public_key_or_identifier() -> str

Get the public key or identifier.

get`_`message¶

def get_message(public_key: str) -> bytes

Get the message to sign.

construct`_`message¶

@classmethod
def construct_message(cls, public_key: str, identifier: SimpleIdOrStr,
                      not_before_string: str, not_after_string: str,
                      message_format: str) -> bytes

Construct message for singning.

Arguments:

public_key: the public key
identifier: identifier to be signed
not_before_string: signature not valid before
not_after_string: signature not valid after
message_format: message format used for signing

Returns:

the message

get`_`signature¶

def get_signature(path_prefix: Optional[PathLike] = None) -> str

Get signature from save_path.

Arguments:

path_prefix: the path prefix to be prepended to save_path. Defaults to cwd.

Returns:

the signature.

json¶

@property
def json() -> Dict

Compute the JSON representation.

from`_`json¶

@classmethod
def from_json(cls, obj: Dict) -> "CertRequest"

Compute the JSON representation.

``eq``¶

def __eq__(other: Any) -> bool

Check equality.

compute`_`specifier`_`from`_`version¶

def compute_specifier_from_version(version: Version) -> str

Compute the specifier set from a version.

version specifier is: >=major.minor.0, <next_major.0.0

Arguments:

version: the version

Returns:

the specifier set

decorator`_`with`_`optional`_`params¶

def decorator_with_optional_params(decorator: Callable) -> Callable

Make a decorator usable either with or without parameters.

In other words, if a decorator "mydecorator" is decorated with this decorator, It can be used both as:

@mydecorator def myfunction(): ...

or as:

@mydecorator(arg1, kwarg1="value") def myfunction(): ...

Arguments:

decorator: a decorator callable

Returns:

a decorator callable

delete`_`directory`_`contents¶

def delete_directory_contents(directory: Path) -> None

Delete the content of a directory, without deleting it.

prepend`_`if`_`not`_`absolute¶

def prepend_if_not_absolute(path: PathLike, prefix: PathLike) -> PathLike

Prepend a path with a prefix, but only if not absolute

Arguments:

path: the path to process.
prefix: the path prefix.

Returns:

the same path if absolute, else the prepended path.

aea.helpers.base¶

locate¶

load_module¶

load_env_file¶

sigint_crossplatform¶

win_popen_kwargs¶

send_control_c¶

RegexConstrainedString Objects¶

__init__¶

SimpleId Objects¶

cd¶

get_logger_method¶

try_decorator¶

MaxRetriesError Objects¶

retry_decorator¶

exception_log_and_reraise¶

recursive_update¶

find_topological_order¶

reachable_nodes¶

cached_property Objects¶

__init__¶

__set_name__¶

__get__¶

ensure_dir¶

dict_to_path_value¶

parse_datetime_from_str¶

CertRequest Objects¶

__init__¶

public_key¶

ledger_id¶

key_identifier¶

identifier¶

not_before_string¶

not_after_string¶

not_before¶

not_after¶

message_format¶

save_path¶

get_absolute_save_path¶

public_key_or_identifier¶

get_message¶

construct_message¶

get_signature¶

json¶

from_json¶

__eq__¶

compute_specifier_from_version¶

decorator_with_optional_params¶

delete_directory_contents¶

prepend_if_not_absolute¶

load`_`module¶

load`_`env`_`file¶

sigint`_`crossplatform¶

win`_`popen`_`kwargs¶

send`_`control`_`c¶

``init``¶

get`_`logger`_`method¶

try`_`decorator¶

retry`_`decorator¶

exception`_`log`_`and`_`reraise¶

recursive`_`update¶

find`_`topological`_`order¶

reachable`_`nodes¶

cached`_`property Objects¶

``init``¶

`__`set`_`name`__`¶

``get``¶

ensure`_`dir¶

dict`_`to`_`path`_`value¶

parse`_`datetime`_`from`_`str¶

``init``¶

public`_`key¶

ledger`_`id¶

key`_`identifier¶

not`_`before`_`string¶

not`_`after`_`string¶

not`_`before¶

not`_`after¶

message`_`format¶

save`_`path¶

get`_`absolute`_`save`_`path¶

public`_`key`_`or`_`identifier¶

get`_`message¶

construct`_`message¶

get`_`signature¶

from`_`json¶

``eq``¶

compute`_`specifier`_`from`_`version¶

decorator`_`with`_`optional`_`params¶

delete`_`directory`_`contents¶

prepend`_`if`_`not`_`absolute¶