runway.util module

Utility functions.

class runway.util.BaseModel[source]

Bases: pydantic.main.BaseModel

Base class for Runway models.

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.

get(name: str, default: Optional[Any] = None) → Any[source]

Implement evaluation of self.get.

  • name – Attribute name to return the value for.

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

class runway.util.JsonEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

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.

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

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 OverflowError). 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.

default(o: Any) → Any[source]

Encode types not supported by the default JSONEncoder.


o – Object to encode.


JSON serializable data type.


TypeError – Object type could not be encoded.

class runway.util.MutableMap(*args, **kwds)[source]

Bases:, typing.Generic

Base class for mutable map objects.

Initialize class.

Provided kwargs are added to the object as attributes.


property data

Sanitized output of __dict__.

Removes anything that starts with _.


Clear _found_cache.

find(query: str, default: Optional[Any] = None, ignore_cache: bool = False) → Any[source]

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.

  • 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: Optional[Any] = None) → Any[source]

Implement evaluation of self.get.

  • key – Attribute name to return the value for.

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

class runway.util.SafeHaven(*args, **kwds)[source]

Bases: contextlib.AbstractContextManager, typing.Generic

Context manager that caches and resets important values on exit.

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

Instantiate class.

  • 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 values cached by this context manager.


Reset the value of os.environ.


Reset the value of sys.argv.


Reset the value of sys.modules.


Reset the value of sys.path.

class runway.util.YamlDumper(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)[source]

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.


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


YAML 1.2 Specification: used for reference.

increase_indent(flow: bool = False, indentless: bool = False)None[source]

Override parent method.

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

Context manager for temporarily changing sys.argv.

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

Change directory.

Adapted from


newdir – Path to change directory into.

runway.util.ensure_file_is_executable(path: str)None[source]

Exit if file is not executable.


path – Path to file.


SystemExit – file is not executable.

runway.util.ensure_string(value: Any)str[source]

Ensure value is a string.

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

Context manager for temporarily changing os.environ.

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


env – Dictionary to use when updating os.environ.

runway.util.json_serial(obj: Any) → Any[source]

JSON serializer for objects not serializable by default json code.

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

Convert “.” delimited strings to a python object.

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


Object being imported from the provided path.


runway.util.merge_dicts(dict1: Dict[Any, Any], dict2: Dict[Any, Any], deep_merge: bool = '...') → Dict[str, Any][source]
runway.util.merge_dicts(dict1: List[Any], dict2: List[Any], deep_merge: bool = '...') → List[Any]

Merge dict2 into dict1.

runway.util.snake_case_to_kebab_case(value: str)str[source]

Convert snake_case to kebab-case.


value – The string value to convert.

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

Return boto3 client args dict with environment creds.

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

Join paths in environment dict down to strings.

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

Return single-level dictionary from dictionary of dictionaries.

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

Return CFN output value.

  • key – Name of the output.

  • outputs – List of Stack outputs.


Return path of embedded libraries.

runway.util.get_hash_for_filename(filename: str, hashfile_path: str)str[source]

Return hash for filename in the hashfile.

runway.util.ignore_exit_code_0() → Iterator[None][source]

Capture exit calls and ignore those with exit code 0.

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

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 like pipenv that will have a windows-only suffix applied to them

runway.util.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[source]

Run list of commands.

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

Return cryptographic hash of file.

runway.util.md5sum(filename: str)str[source]

Return MD5 hash of file.

runway.util.sha256sum(filename: str)str[source]

Return SHA256 hash of file.

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

Temporarily prepend embedded packages to sys.path.

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

Mimic ‘which’ command behavior.