aea.helpers.base
Miscellaneous helpers.
locate
locate(path: str) -> Any
Locate an object by name or dotted save_path, importing as necessary.
load_
module
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.
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) -> None
Load the content of the environment file into the process environment.
Arguments:
env_file
: save_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: 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("") 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: PathLike) -> Generator
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: 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 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: str = "error") -> Callable
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) -> Generator
Run code in context to log and re raise exception.
Arguments:
log_method
: function to print logmessage
: message template to add error text.
recursive_
update
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.
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.
cached_
property Objects
class cached_property()
Cached property from python3.8 functools.
__
init__
| __init__(func: Callable) -> None
Init cached property.
__
set_
name__
| __set_name__(_: Any, name: Any) -> None
Set name.
__
get__
| __get__(instance: Any, _: Optional[Any] = None) -> Any
Get instance.
ensure_
dir
ensure_dir(dir_path: str) -> None
Check if dir_path is a directory or create it.
dict_
to_
path_
value
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
parse_datetime_from_str(date_string: str) -> datetime.datetime
Parse datetime from string.
CertRequest Objects
class CertRequest()
Certificate request for proof of representation.
__
init__
| __init__(public_key: str, identifier: SimpleIdOrStr, ledger_id: SimpleIdOrStr, not_before: str, not_after: str, save_path: str) -> None
Initialize the certificate request.
Arguments:
public_key
: the public key, or the key id.identifier
: certificate identifier.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.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.save_path
: the save_path where to save the certificate.
public_
key
| @property
| public_key() -> Optional[str]
Get the public key.
ledger_
id
| @property
| ledger_id() -> str
Get the ledger id.
key_
identifier
| @property
| key_identifier() -> Optional[str]
Get the key identifier.
identifier
| @property
| identifier() -> str
Get the identifier.
not_
before_
string
| @property
| not_before_string() -> str
Get the not_before field as string.
not_
after_
string
| @property
| not_after_string() -> str
Get the not_after field as string.
not_
before
| @property
| not_before() -> datetime.datetime
Get the not_before field.
not_
after
| @property
| not_after() -> datetime.datetime
Get the not_after field.
save_
path
| @property
| 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.
get_
absolute_
save_
path
| 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
| public_key_or_identifier() -> str
Get the public key or identifier.
get_
message
| get_message(public_key: str) -> bytes
Get the message to sign.
get_
signature
| get_signature(path_prefix: Optional[PathLike] = None) -> str
Get signature from save_path.
Arguments:
path_prefix
: the path prefix to be prependend to save_path. Defaults to cwd.
Returns:
the signature.
json
| @property
| json() -> Dict
Compute the JSON representation.
from_
json
| @classmethod
| from_json(cls, obj: Dict) -> "CertRequest"
Compute the JSON representation.
__
eq__
| __eq__(other: Any) -> bool
Check equality.
compute_
specifier_
from_
version
compute_specifier_from_version(version: Version) -> str
Compute the specifier set from a version, by varying only on the patch number.
I.e. from "{major}.{minor}.{patch}", return
">={major}.{minor}.0, <{major}.{minor + 1}.0"
Arguments:
version
: the version
Returns:
the specifier set
decorator_
with_
optional_
params
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(): ...
delete_
directory_
contents
delete_directory_contents(directory: Path) -> None
Delete the content of a directory, without deleting it.
prepend_
if_
not_
absolute
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.