runway.utils package

Utility functions.

class runway.utils.BaseModel

Bases: pydantic.main.BaseModel

Base class for Runway models.

get(name: str, default: Any = None) → Any

Safely get the value of an attribute.

Parameters

• name – Attribute name to return the value for.

• default – Value to return if attribute is not found.

__contains__(name: object) → bool

Implement evaluation of ‘in’ conditional.

Parameters

name – The name to check for existence in the model.

__getitem__(name: str) → Any

Implement evaluation of self[name].

Parameters

name – Attribute name to return the value for.

Returns

The value associated with the provided name/attribute name.

Raises

AttributeError – If attribute does not exist on this object.

__setitem__(name: str, value: Any) → None

Implement item assignment (e.g. self[name] = value).

Parameters

• name – Attribute name to set.

• value – Value to assign to the attribute.

__init__(**data: Any) → None

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

__iter__() → TupleGenerator

so dict(model) works

__new__(**kwargs)

__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any, None, None]

Used by devtools (https://python-devtools.helpmanual.io/) to provide a human readable representations of objects

__repr_name__() → str

Name of the instance’s class, used in __repr__.

__rich_repr__() → RichReprResult

Get fields for Rich library

classmethod __try_update_forward_refs__(**localns: Any) → None

Same as update_forward_refs but will not raise exception when forward references are not defined.

classmethod construct(_fields_set: Optional[SetStr] = None, **values: Any) → Model

Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

copy(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, update: Optional[DictStrAny] = None, deep: bool = False) → Model

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters

• include – fields to include in new model

• exclude – fields to exclude from new model, as with values this takes precedence over include

• update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data

• deep – set to True to make a deep copy of the model

Returns

new model instance

dict(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) → DictStrAny

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

json(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Optional[Callable[[Any], Any]] = None, models_as_dict: bool = True, **dumps_kwargs: Any) → str

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().

classmethod update_forward_refs(**localns: Any) → None

Try to update ForwardRefs on fields based on this Model, globalns and localns.

class runway.utils.JsonEncoder

Bases: json.encoder.JSONEncoder

Encode Python objects to JSON data.

This class can be used with json.dumps() to handle most data types that can occur in responses from AWS.

Usage:

>>> json.dumps(data, cls=JsonEncoder)

default(o: Any) → Any

Encode types not supported by the default JSONEncoder.

Parameters

o – Object to encode.

Returns

JSON serializable data type.

Raises

TypeError – Object type could not be encoded.

__init__(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Constructor for JSONEncoder, with sensible defaults.

If skipkeys is false, then it is a TypeError to attempt encoding of keys that are not str, int, float or None. If skipkeys is True, such items are simply skipped.

If ensure_ascii is true, the output is guaranteed to be str objects with all incoming non-ASCII characters escaped. If ensure_ascii is false, the output can contain non-ASCII characters.

If check_circular is true, then lists, dicts, and custom encoded objects will be checked for circular references during encoding to prevent an infinite recursion (which would cause an RecursionError). Otherwise, no such check takes place.

If allow_nan is true, then NaN, Infinity, and -Infinity will be encoded as such. This behavior is not JSON specification compliant, but is consistent with most JavaScript based encoders and decoders. Otherwise, it will be a ValueError to encode such floats.

If sort_keys is true, then the output of dictionaries will be sorted by key; this is useful for regression tests to ensure that JSON serializations can be compared on a day-to-day basis.

If indent is a non-negative integer, then JSON array elements and object members will be pretty-printed with that indent level. An indent level of 0 will only insert newlines. None is the most compact representation.

If specified, separators should be an (item_separator, key_separator) tuple. The default is (’, ‘, ‘: ‘) if indent is None and (‘,’, ‘: ‘) otherwise. To get the most compact JSON representation, you should specify (‘,’, ‘:’) to eliminate whitespace.

If specified, default is a function that gets called for objects that can’t otherwise be serialized. It should return a JSON encodable version of the object or raise a TypeError.

__new__(**kwargs)

encode(o)

Return a JSON string representation of a Python data structure.

>>> from json.encoder import JSONEncoder
>>> JSONEncoder().encode({"foo": ["bar", "baz"]})
'{"foo": ["bar", "baz"]}'

iterencode(o, _one_shot=False)

Encode the given object and yield each string representation as available.

For example:

for chunk in JSONEncoder().iterencode(bigobject):
    mysocket.write(chunk)

class runway.utils.MutableMap

Bases: MutableMapping[str, Any]

Base class for mutable map objects.

__init__(**kwargs: Any) → None

Initialize class.

Provided kwargs are added to the object as attributes.

Example

property data: Dict[str, Any]

Sanitized output of __dict__.

Removes anything that starts with _.

clear_found_cache() → None

Clear _found_cache.

find(query: str, default: Any = None, ignore_cache: bool = False) → Any

Find a value in the map.

Previously found queries are cached to increase search speed. The cached value should only be used if values are not expected to change.

Parameters

• query – A period delimited string that is split to search for nested values

• default – The value to return if the query was unsuccessful.

• ignore_cache – Ignore cached value.

get(key: str, default: Any = None) → Any

Implement evaluation of self.get.

Parameters

• key – Attribute name to return the value for.

• default – Value to return if attribute is not found.

__bool__() → bool

Implement evaluation of instances as a bool.

__contains__(value: Any) → bool

Implement evaluation of ‘in’ conditional.

__getitem__(key: str) → Any

Implement evaluation of self[key].

Parameters

key – Attribute name to return the value for.

Returns

The value associated with the provided key/attribute name.

Raises

AttributeError – If attribute does not exist on this object.

Example

__setitem__(key: str, value: Any) → None

Implement assignment to self[key].

Parameters

• key – Attribute name to associate with a value.

• value – Value of a key/attribute.

Example

__delitem__(key: str) → None

Implement deletion of self[key].

Parameters

key – Attribute name to remove from the object.

Example

__len__() → int

Implement the built-in function len().

Example

__iter__() → Iterator[Any]

Return iterator object that can iterate over all attributes.

Example

__str__() → str

Return string representation of the object.

__new__(**kwargs)

clear() → None.  Remove all items from D.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.

If key is not found, d is returned if given, otherwise KeyError is raised.

popitem() → (k, v), remove and return some (key, value) pair

as a 2-tuple; but raise KeyError if D is empty.

setdefault(k[, d]) → D.get(k,d), also set D[k]=d if k not in D

update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

values() → an object providing a view on D's values

class runway.utils.SafeHaven

Bases: ContextManager[SafeHaven]

Context manager that caches and resets important values on exit.

Caches and resets os.environ, sys.argv, sys.modules, and sys.path.

__init__(argv: Optional[Iterable[str]] = None, environ: Optional[Dict[str, str]] = None, sys_modules_exclude: Optional[Iterable[str]] = None, sys_path: Optional[Iterable[str]] = None) → None

Instantiate class.

Parameters

• argv – Override the value of sys.argv.

• environ – Update os.environ.

• sys_modules_exclude – A list of modules to exclude when reverting sys.modules to its previous state. These are checked as module.startswith(name).

• sys_path – Override the value of sys.path.

reset_all() → None

Reset all values cached by this context manager.

reset_os_environ() → None

Reset the value of os.environ.

reset_sys_argv() → None

Reset the value of sys.argv.

reset_sys_modules() → None

Reset the value of sys.modules.

reset_sys_path() → None

Reset the value of sys.path.

__enter__() → runway.utils.SafeHaven

Enter the context manager.

Returns

Instance of the context manager.

Return type

SafeHaven

__exit__(exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], traceback: Optional[types.TracebackType]) → None

Exit the context manager.

__new__(**kwargs)

class runway.utils.YamlDumper

Bases: yaml.dumper.Dumper

Custom YAML Dumper.

This Dumper allows for YAML to be output to follow YAML spec 1.2, example 2.3 of collections (2.1). This provides an output that is more humanreadable and complies with yamllint.

Example

>>> print(yaml.dump({'key': ['val1', 'val2']}, Dumper=YamlDumper))

Note

YAML 1.2 Specification: https://yaml.org/spec/1.2/spec.html used for reference.

increase_indent(flow: bool = False, indentless: bool = False) → None

Override parent method.

__init__(stream, default_style=None, default_flow_style=False, canonical=None, indent=None, width=None, allow_unicode=None, line_break=None, encoding=None, explicit_start=None, explicit_end=None, version=None, tags=None, sort_keys=True)

__new__(**kwargs)

runway.utils.argv(*args: str) → Iterator[None]

Context manager for temporarily changing sys.argv.

runway.utils.change_dir(newdir: Union[pathlib.Path, str]) → Iterator[None]

Change directory.

Adapted from http://stackoverflow.com/a/24176022

Parameters

newdir – Path to change directory into.

runway.utils.ensure_file_is_executable(path: str) → None

Exit if file is not executable.

Parameters

path – Path to file.

Raises

SystemExit – file is not executable.

runway.utils.ensure_string(value: Any) → str

Ensure value is a string.

runway.utils.environ(env: Optional[Dict[str, str]] = None, **kwargs: str) → Iterator[None]

Context manager for temporarily changing os.environ.

The original value of os.environ is restored upon exit.

Parameters

env – Dictionary to use when updating os.environ.

runway.utils.json_serial(obj: Any) → Any

JSON serializer for objects not serializable by default json code.

runway.utils.load_object_from_string(fqcn: str, try_reload: bool = False) → Union[type, Callable[[...], Any]]

Convert “.” delimited strings to a python object.

Parameters

• fqcn – A “.” delimited string representing the full path to an object (function, class, variable) inside a module

• try_reload – Try to reload the module so any global variables set within the file during import are reloaded. This only applies to modules that are already imported and are not builtin.

Returns

Object being imported from the provided path.

Example:

load_object_from_string("os.path.basename")
load_object_from_string("logging.Logger")
load_object_from_string("LocalClassName")

runway.utils.merge_dicts(dict1: Dict[Any, Any], dict2: Dict[Any, Any], deep_merge: bool = True) → Dict[str, Any]

runway.utils.merge_dicts(dict1: List[Any], dict2: List[Any], deep_merge: bool = True) → List[Any]

Merge dict2 into dict1.

runway.utils.snake_case_to_kebab_case(value: str) → str

Convert snake_case to kebab-case.

Parameters

value – The string value to convert.

runway.utils.extract_boto_args_from_env(env_vars: Dict[str, str]) → Dict[str, str]

Return boto3 client args dict with environment creds.

runway.utils.flatten_path_lists(env_dict: Dict[str, Any], env_root: Optional[str] = None) → Dict[str, Any]

Join paths in environment dict down to strings.

runway.utils.merge_nested_environment_dicts(env_dicts: Dict[str, Any], env_name: Optional[str] = None, env_root: Optional[str] = None) → Dict[str, Any]

Return single-level dictionary from dictionary of dictionaries.

runway.utils.find_cfn_output(key: str, outputs: List[OutputTypeDef]) → Optional[str]

Return CFN output value.

Parameters

• key – Name of the output.

• outputs – List of Stack outputs.

runway.utils.get_embedded_lib_path() → str

Return path of embedded libraries.

runway.utils.get_hash_for_filename(filename: str, hashfile_path: str) → str

Return hash for filename in the hashfile.

runway.utils.ignore_exit_code_0() → Iterator[None]

Capture exit calls and ignore those with exit code 0.

runway.utils.fix_windows_command_list(commands: List[str]) → List[str]

Return command list with working Windows commands.

npm on windows is npm.cmd, which will blow up subprocess.check_call([‘npm’, ‘…’])

Similar issues arise when calling python apps that will have a windows-only suffix applied to them.

runway.utils.run_commands(commands: List[Union[str, List[str], Dict[str, Union[str, List[str]]]]], directory: Union[pathlib.Path, str], env: Optional[Dict[str, str]] = None) → None

Run list of commands.

runway.utils.get_file_hash(filename: str, algorithm: Literal['blake2b', 'md5', 'sha1', 'sha224', 'sha256', 'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512', 'sha384', 'sha512', 'shake_128', 'shake_256']) → str

Return cryptographic hash of file.

Deprecated since version 2.4.0: Use runway.utils.FileHash instead.

runway.utils.md5sum(filename: str) → str

Return MD5 hash of file.

Deprecated since version 2.4.0: Use runway.utils.FileHash instead.

runway.utils.sha256sum(filename: str) → str

Return SHA256 hash of file.

Deprecated since version 2.4.0: Use runway.utils.FileHash instead.

runway.utils.use_embedded_pkgs(embedded_lib_path: Optional[str] = None) → Iterator[None]

Temporarily prepend embedded packages to sys.path.

runway.utils.which(program: str) → Optional[str]

Mimic ‘which’ command behavior.