clowder.util package¶

Submodules¶

clowder.util.connectivity module¶

Network connectivity

clowder.util.connectivity.is_offline(host: str = '8.8.8.8', port: int = 53, timeout: int = 3) → bool¶

Returns True if offline, False otherwise

Service: domain (DNS/TCP)

Note

Implementation source https://stackoverflow.com/a/33117579

Parameters:	host (str) – Host to check. Default is 8.8.8.8 (google-public-dns-a.google.com) port (int) – Port number. Default is 53/tcp timeout (int) – Seconds to wait until timeout
Returns:	True, if offline
Return type:	bool
Raises:	ClowderError –

clowder.util.connectivity.network_connection_required(func)¶: If no network connection, print offline message and exit

clowder.util.decorators module¶

Decorators

clowder.util.decorators.clowder_git_repo_required(func)¶: If no clowder git repo exists, print clowder git repo not found message and exit

clowder.util.decorators.clowder_repo_required(func)¶: If no clowder repo exists, print clowder repo not found message and exit

clowder.util.decorators.print_clowder_name(func)¶: Print clowder name

clowder.util.decorators.print_clowder_repo_status(func)¶: Print clowder repo status

clowder.util.decorators.print_clowder_repo_status_fetch(func)¶: Print clowder repo status

clowder.util.decorators.valid_clowder_yaml_required(func)¶: If clowder yaml file is invalid, print invalid yaml message and exit

clowder.util.execute module¶

Subprocess execution utilities

clowder.util.execute.execute_command(command: Union[str, List[str]], path: pathlib.Path, env: Optional[dict] = None, print_output: bool = True) → None¶

Execute command via subprocess

Parameters:	List[str]] command (Union[str,) – Command to run path (Path) – Path to set as `cwd` env (Optional[dict]) – Enviroment to set as `env` print_output (bool) – Whether to print output
Raises:	ClowderError –

clowder.util.execute.execute_forall_command(command: Union[str, List[str]], path: pathlib.Path, forall_env: dict, print_output: bool) → None¶

Execute forall command with additional environment variables and display continuous output

Parameters:	List[str]] command (Union[str,) – Command to run path (Path) – Path to set as `cwd` forall_env (dict) – Enviroment to set as `env` print_output (bool) – Whether to print output
Raises:	ClowderError –

clowder.util.file_system module¶

File system utilities

clowder.util.file_system.create_backup_file(file: pathlib.Path) → None¶

Copy file to {file}.backup

Parameters:	file (Path) – File path to copy
Raises:	OSError –

clowder.util.file_system.make_dir(directory: pathlib.Path) → None¶

Make directory if it doesn’t exist

Parameters:	directory (str) – Directory path to create
Raises:	ClowderError –

clowder.util.file_system.remove_directory(dir_path: pathlib.Path) → None¶

Remove directory at path

Parameters:	dir_path (str) – Path to directory to remove
Raises:	ClowderError –

clowder.util.file_system.remove_file(file: pathlib.Path) → None¶

Remove file

Parameters:	file (Path) – File path to remove
Raises:	OSError –

clowder.util.file_system.restore_from_backup_file(file: pathlib.Path) → None¶

Copy {file}.backup to file

Parameters:	file (Path) – File path to copy
Raises:	OSError –

clowder.util.file_system.symlink_clowder_yaml(source: pathlib.Path, target: pathlib.Path) → None¶

Force symlink creation

Parameters:	source (Path) – File to create symlink pointing to target (Path) – Symlink location
Raises:	ClowderError –

clowder.util.formatting module¶

String formatting utilities

clowder.util.formatting.check_for_duplicates(list_of_elements: List[str]) → Optional[str]¶

Check if given list contains any duplicates

Parameters:	list_of_elements (List[str]) – List of strings to check for duplicates
Returns:	First duplicate encountered, or None if no duplicates found
Return type:	Optional[str]

clowder.util.formatting.clowder_command(cmd: str) → str¶

Return formatted clowder command name

Parameters:	cmd (str) – Clowder command name
Returns:	Formatted clowder command name
Return type:	str

clowder.util.formatting.clowder_name(name: str) → str¶

Return formatted clowder name

Parameters:	name (str) – Clowder name
Returns:	Formatted clowder name
Return type:	str

clowder.util.formatting.command(cmd: Union[str, List[str]]) → str¶

Return formatted command name

Parameters:	List[str]] cmd (Union[str,) – Clowder command name
Returns:	Formatted clowder command name
Return type:	str

clowder.util.formatting.error(err: Exception) → str¶

Return error message for generic error

Parameters:	err (Exception) – Generic error
Returns:	Formatted generic error
Return type:	str

clowder.util.formatting.error_ambiguous_clowder_yaml() → str¶

Return formatted error string for ambiguous clowder yaml file

Returns:	Formatted ambigious clowder yaml error
Return type:	str

clowder.util.formatting.error_clone_missing_projects() → str¶

Format error message for clone missing projects

Returns:	Formatted error message for clone missing projects
Return type:	str

clowder.util.formatting.error_clowder_already_initialized() → str¶

Format error message for clowder already initialized

Returns:	Formatted message for clowder already initialized error
Return type:	str

clowder.util.formatting.error_clowder_symlink_source_missing(symlink_path: pathlib.Path) → str¶

Return formatted error string for clowder symlink source not found

Parameters:	symlink_path (Path) – Clowder yaml symlink path
Returns:	Formatted clowder symlink source not found warning
Return type:	str

clowder.util.formatting.error_command_failed(cmd: Union[str, List[str]]) → str¶

Format error message for failed command

Parameters:	List[str]] cmd (Union[str,) – Clowder command name
Returns:	Formatted clowder command name
Return type:	str

clowder.util.formatting.error_directory_exists(dir_path: str) → str¶

Format error message for already existing directory

Parameters:	dir_path (Path) – Directory path
Returns:	Formatted directory exists error
Return type:	str

clowder.util.formatting.error_duplicate_project_path(path: str, yml: pathlib.Path) → str¶

Return formatted error string for duplicate project path

Parameters:	path (str) – Duplicate project path yml (Path) – Path to yaml file
Returns:	Formatted duplicate remote upstream name error
Return type:	str

clowder.util.formatting.error_duplicate_version(version: str) → str¶

Format error message for duplicate clowder version

Parameters:	version (str) – Clowder version name
Returns:	Formatted duplicate clowder version error
Return type:	str

clowder.util.formatting.error_empty_yaml(yml: pathlib.Path, name: pathlib.Path) → str¶

Return formatted error string for empty clowder yaml file

Parameters:	yml (Path) – Path to yaml file name (Path) – Path to use in error message
Returns:	Formatted empty yaml error
Return type:	str

clowder.util.formatting.error_existing_file_at_clowder_repo_path(file_path: str) → str¶

Format error message for existing file at .clowder path

Parameters:	file_path (str) – Path to existing .clowder file
Returns:	Formatted existing file at .clowder path error
Return type:	str

clowder.util.formatting.error_existing_file_at_symlink_target_path(name: str) → str¶

Format error message for existing non-symlink file at symlink target path

Parameters:	name (str) – Path to use in error message
Returns:	Formatted existing non-symlink file at symlink target path error
Return type:	str

clowder.util.formatting.error_failed_clowder_init() → str¶

Format error message for failed clowder init

Returns:	Formatted failed clowder init error
Return type:	str

clowder.util.formatting.error_failed_create_directory(dir_path: str) → str¶

Format error message for failing to create directory

Parameters:	dir_path (str) – Directory path to create
Returns:	Formatted create directory error
Return type:	str

clowder.util.formatting.error_failed_create_parser() → str¶

Format error message for failing to create cli parsers

Returns:	Formatted failed to create parsers error
Return type:	str

clowder.util.formatting.error_failed_remove_directory(dir_path: str) → str¶

Format error message for failing to remove directory

Parameters:	dir_path (str) – Directory path to remove
Returns:	Formatted remove directory error
Return type:	str

clowder.util.formatting.error_failed_remove_file(file_path: str) → str¶

Format error message for failing to remove file

Parameters:	file_path (str) – File path
Returns:	Formatted remove file error
Return type:	str

clowder.util.formatting.error_failed_symlink_file(target: str, source: str) → str¶

Format error message for failing to symlink file

Parameters:	target (str) – Target file path source (str) – Source file path
Returns:	Formatted remove file error
Return type:	str

clowder.util.formatting.error_file_exists(file_path: str) → str¶

Format error message for already existing file

Parameters:	file_path (str) – File path name
Returns:	Formatted file exists error
Return type:	str

clowder.util.formatting.error_groups_contains_all(yml: pathlib.Path) → str¶

Return formatted error string for invalid ‘all’ entry in groups list

Parameters:	yml (Path) – Path to yaml file
Returns:	Formatted error for groups containing all
Return type:	str

clowder.util.formatting.error_invalid_config_file(file_path: str) → str¶

Return error message for invalid config file

Parameters:	file_path (str) – Invalid config file path
Returns:	Formatted invalid config file error
Return type:	str

clowder.util.formatting.error_invalid_git_config_value(key: str, value: str) → str¶

Format error message for invalid git config value

Parameters:	key (str) – Key for value with invalid git type value (str) – Value with invalid git type
Returns:	Formatted error message for invalid git config value
Return type:	str

clowder.util.formatting.error_invalid_project_state() → str¶

Format error message for invalid project state

Returns:	Formatted error message for invalid project state
Return type:	str

clowder.util.formatting.error_invalid_ref(ref: str, yml: pathlib.Path) → str¶

Return formatted error string for incorrect ref

Parameters:	ref (str) – Git reference yml (Path) – Path to yaml file
Returns:	Formatted invalid ref error
Return type:	str

clowder.util.formatting.error_invalid_yaml_file(name: str) → str¶

Return error message for invalid yaml file

Parameters:	name (str) – Invalid file’s name
Returns:	Formatted yaml error
Return type:	str

clowder.util.formatting.error_missing_clowder_git_repo() → str¶

Format error message for missing clowder git repo

Returns:	Formatted missing clowder git repo error
Return type:	str

clowder.util.formatting.error_missing_clowder_repo() → str¶

Format error message for missing clowder repo

Returns:	Formatted missing clowder repo error
Return type:	str

clowder.util.formatting.error_missing_clowder_yaml() → str¶

Format error message for missing clowder yaml file

Returns:	Formatted missing YAML error
Return type:	str

clowder.util.formatting.error_missing_default_source() → str¶

Format error message for missing default source

Returns:	Formatted error message for missing default source
Return type:	str

clowder.util.formatting.error_missing_file(yaml_file: str) → str¶

Format error message for missing linked clowder yaml file

Parameters:	yaml_file (str) – Path to missing yaml file
Returns:	Formatted missing YAML error
Return type:	str

clowder.util.formatting.error_no_clowder_found(dir_path: str) → str¶

Format error message for no clowder found

Parameters:	dir_path (str) – Missing clowder directory path
Returns:	Formatted no clowder found error
Return type:	str

clowder.util.formatting.error_offline() → str¶

Return error message for no internet connection

Returns:	Offline error message
Return type:	str

clowder.util.formatting.error_open_file(path: str) → str¶

Format error message for failing to open file

Parameters:	path (str) – File path
Returns:	Formatted file error
Return type:	str

clowder.util.formatting.error_parallel_command_failed() → str¶

Return formatted error string for parallel command failed

Returns:	Formatted parallel command failed error
Return type:	str

clowder.util.formatting.error_parallel_commands_unavailable() → str¶

Return formatted error string for parallel command unavailable

Returns:	Formatted parallel command unavailable error
Return type:	str

clowder.util.formatting.error_parallel_exception(file_path: str, *args) → str¶

Return formatted error string for parallel error

Parameters:	file_path (str) – Clowder file path args – Method arguments
Returns:	Formatted parallel exception error
Return type:	str

clowder.util.formatting.error_project_not_found() → str¶

Format error message for project not found

Returns:	Formatted error message for clone missing projects
Return type:	str

clowder.util.formatting.error_remote_already_exists(remote_name: str, remote_url: str, actual_url: str) → str¶

Format error message when remote already exists with different url

Parameters:	remote_name (str) – Remote name remote_url (str) – Remote URL actual_url (str) – Actual URL
Returns:	Formatted remote exists error
Return type:	str

clowder.util.formatting.error_remote_dup(upstream: str, project: str, remote: str, yml: pathlib.Path) → str¶

Return formatted error string for upstream with same remote as project

Parameters:	upstream (str) – Upstream name project (str) – Project name remote (str) – Remote name yml (Path) – Path to yaml file
Returns:	Formatted duplicate remote upstream name error
Return type:	str

clowder.util.formatting.error_save_default(name: str) → str¶

Format error message for trying to save disallowed version

Parameters:	name (str) – Version name
Returns:	Formatted default version error
Return type:	str

clowder.util.formatting.error_save_file(file_path: str) → str¶

Format error message for failing to save file

Parameters:	file_path (Path) – File path
Returns:	Formatted save failure error
Return type:	str

clowder.util.formatting.error_save_version_exists(version_name: str, yml: pathlib.Path) → str¶

Format error message previous existing saved version

Parameters:	version_name (str) – Version name yml (Path) – Path to yaml file
Returns:	Formatted version exists error
Return type:	str

clowder.util.formatting.error_source_default_not_found(source: str, yml: pathlib.Path) → str¶

Return formatted error string for unknown default source specified

Parameters:	source (str) – Source name yml (Path) – Path to yaml file
Returns:	Formatted source not found error
Return type:	str

clowder.util.formatting.error_source_not_defined(name: Optional[str] = None) → str¶

clowder.util.formatting.error_source_not_found(source: str, yml: pathlib.Path, project: str, upstream: Optional[str] = None) → str¶

Return formatted error string for project with unknown source specified

Parameters:	source (str) – Source name yml (Path) – Path to yaml file project (str) – Project name upstream (Optional[str]) – Upstream name
Returns:	Formatted source not found error
Return type:	str

clowder.util.formatting.error_source_not_validated() → str¶

clowder.util.formatting.error_sources_already_validated() → str¶

clowder.util.formatting.error_symlink_source_missing(source: pathlib.Path) → str¶

Return formatted error string for symlink source not found

Parameters:	source (Path) – Symlink source path
Returns:	Formatted clowder symlink source not found warning
Return type:	str

clowder.util.formatting.error_system_exit() → str¶

Format error message for system exit

Returns:	Formatted system exit error
Return type:	str

clowder.util.formatting.error_timestamp_not_found() → str¶

Return timestamp not found error message

Returns:	Formatted timestamp not found error message
Return type:	str

clowder.util.formatting.error_unknown_config_type() → str¶

Format error message for unknown config type

Returns:	Formatted error message for unknown config type
Return type:	str

clowder.util.formatting.error_unknown_error() → str¶

Format error message for unknown error

Returns:	Formatted unknown error message
Return type:	str

clowder.util.formatting.error_unknown_project(name: str) → str¶

Return formatted unknown project name error

Parameters:	name (str) – Project name
Returns:	Formatted unknown project name error
Return type:	str

clowder.util.formatting.error_user_interrupt() → str¶

Format error message for user interrupt

Returns:	Formatted user interrupt error
Return type:	str

clowder.util.formatting.error_wrong_group_type() → str¶

clowder.util.formatting.error_wrong_source_type() → str¶

Format error message for wrong source type

Returns:	Formatted error message for wrong source type
Return type:	str

clowder.util.formatting.error_wrong_submodules_type() → str¶

clowder.util.formatting.error_wrong_upstream_type() → str¶

Format error message for wrong upstream type

Returns:	Formatted error message for wrong upstream type
Return type:	str

clowder.util.formatting.options_help_message(options: Tuple[str, ...], message: str) → str¶

Help message for groups option

Parameters:	..] options (Tuple[str,) – List of options message (str) – Help message
Returns:	Formatted options help message
Return type:	str

clowder.util.formatting.path_string(path: str) → str¶

Return formatted path

Parameters:	path (Path) – Path name
Returns:	Formatted path name
Return type:	str

clowder.util.formatting.project_options_help_message(message: str) → str¶

Help message for projects/groups options

Parameters:	message (str) – Help message
Returns:	Formatted options help message
Return type:	str

clowder.util.formatting.ref_string(ref: str) → str¶

Return formatted ref name

Parameters:	ref (str) – Git reference
Returns:	Formatted ref name
Return type:	str

clowder.util.formatting.remote_string(remote: str) → str¶

Return formatted remote name

Parameters:	remote (str) – Remote name
Returns:	Formmatted remote name
Return type:	str

clowder.util.formatting.remove_prefix(text: str, prefix: str) → str¶

Remove prefix from string

Parameters:	text (str) – Text to remove prefix from prefix (str) – Prefix to remoe
Returns:	Text with prefix removed if present
Return type:	str

clowder.util.formatting.save_version_message(version: str, yml: str) → str¶

Format message for saving version

Parameters:	version (str) – Clowder version name yml (str) – Path to yaml file
Returns:	Formatted version name
Return type:	str

clowder.util.formatting.upstream_string(name: str) → str¶

Return formatted upstream name

Parameters:	name (str) – Upstream name
Returns:	Formatted upstream name
Return type:	str

clowder.util.formatting.url_string(url: str) → str¶

Return formatted url

Parameters:	url (str) – URL
Returns:	Formatted URL
Return type:	str

clowder.util.formatting.version_options_help_message(message: str, versions: Tuple[str, ...]) → str¶

Help message for projects/groups options

Parameters:	message (str) – Help message ..] versions (Tuple[str,) – Version choices
Returns:	Formatted options help message
Return type:	str

clowder.util.formatting.version_string(version_name: str) → str¶

Return formatted string for clowder yaml version

Parameters:	version_name (str) – Clowder version name
Returns:	Formatted clowder version name
Return type:	str

clowder.util.formatting.warning_clowder_repo_missing_git_dir() → str¶

Return formatted warning string for existing .clowder directory that isn’t a git repository

Returns:	Formatted warning string for existing .clowder directory that isn’t a git repository
Return type:	str

clowder.util.formatting.warning_clowder_yaml_not_symlink_with_clowder_repo(name: str) → str¶

Return formatted warning string for non-symlink clowder yaml file with an existing clowder repo

Parameters:	name (str) – Clowder yaml file name
Returns:	Formatted warning string for non-symlink clowder yaml file with an existing clowder repo
Return type:	str

clowder.util.formatting.warning_invalid_config_file(file_path: str) → str¶

Return warning message for invalid config file

Parameters:	file_path (Path) – Invalid config file path
Returns:	Formatted invalid config file warning
Return type:	str

clowder.util.parallel module¶

Clowder parallel command

clowder.util.parallel.async_callback(val) → None¶

Increment async progress bar

Parameters:	val – Dummy parameter to satisfy callback interface

clowder.util.parallel.forall_parallel(commands: List[str], projects: Tuple[clowder.data.resolved_project.ResolvedProject, ...], jobs: int, ignore_errors: bool) → None¶

Runs command or script for projects in parallel

Parameters:	commands (List[str]) – Command to run ..] projects (Tuple[ResolvedProject,) – Projects to run command for jobs (int) – Number of jobs to use running parallel commands ignore_errors (bool) – Whether to exit if command returns a non-zero exit code

clowder.util.parallel.herd_parallel(projects: Tuple[clowder.data.resolved_project.ResolvedProject, ...], jobs: int, branch: Optional[str] = None, tag: Optional[str] = None, depth: Optional[int] = None, rebase: bool = False) → None¶

Clone projects or update latest from upstream in parallel

Parameters:

..] projects (Tuple[ResolvedProject,) – Projects to herd
jobs (int) – Number of jobs to use running parallel commands
branch (Optional[str]) – Branch to attempt to herd
tag (Optional[str]) – Tag to attempt to herd
depth (Optional[int]) – Git clone depth. 0 indicates full clone, otherwise must be a positive integer
rebase (bool) – Whether to use rebase instead of pulling latest changes

clowder.util.parallel.herd_project(project: clowder.data.resolved_project.ResolvedProject, branch: str, tag: str, depth: int, rebase: bool) → None¶

Herd command wrapper function for multiprocessing Pool execution

Parameters:	project (ResolvedProject) – ResolvedProject instance branch (str) – Branch to attempt to herd tag (str) – Tag to attempt to herd depth (int) – Git clone depth. 0 indicates full clone, otherwise must be a positive integer rebase (bool) – Whether to use rebase instead of pulling latest changes

clowder.util.parallel.pool_handler(count: int)¶

Pool handler for finishing parallel jobs

Parameters:	count (int) – Total count of projects in progress bar
Raises:	ClowderError –

clowder.util.parallel.reset_parallel(projects: Tuple[clowder.data.resolved_project.ResolvedProject, ...], jobs: int, timestamp_project: Optional[str] = None) → None¶

Reset project branches to upstream or checkout tag/sha as detached HEAD in parallel

Parameters:	..] projects (Tuple[ResolvedProject,) – Project names to reset jobs (int) – Number of jobs to use running parallel commands timestamp_project (Optional[str]) – Reference project to checkout other project timestamps relative to

clowder.util.parallel.reset_project(project: clowder.data.resolved_project.ResolvedProject, timestamp: str) → None¶

Reset command wrapper function for multiprocessing Pool execution

Parameters:	project (ResolvedProject) – ResolvedProject instance timestamp (str) – If not None, reset to commit at timestamp, or closest previous commit

clowder.util.parallel.run_project(project: clowder.data.resolved_project.ResolvedProject, commands: List[str], ignore_errors: bool) → None¶

Run command wrapper function for multiprocessing Pool execution

Parameters:	project (ResolvedProject) – ResolvedProject instance commands (list[str]) – Commands to run ignore_errors (bool) – Whether to exit if command returns a non-zero exit code

clowder.util.parallel.worker_init() → None¶: Process pool terminator

Note

Implementation source https://stackoverflow.com/a/45259908

clowder.util.progress module¶

Progress bar wrapper class

class clowder.util.progress.Progress¶

Bases: object

Class wrapping progress bar

close() → None¶: Close progress bar

complete() → None¶: Complete progress bar

start(count: int) → None¶

Start progress bar

Parameters:	count (int) – Initial count

update() → None¶: Update progress bar

clowder.util.yaml module¶

Clowder command line utilities

clowder.util.yaml.link_clowder_yaml_default(clowder_dir: pathlib.Path) → None¶

Create symlink pointing to clowder yaml file

Parameters:	clowder_dir (Path) – Directory to create symlink in
Raises:	ClowderError –

clowder.util.yaml.link_clowder_yaml_version(clowder_dir: pathlib.Path, version: str) → None¶

Create symlink pointing to clowder yaml file

Parameters:	clowder_dir (Path) – Directory to create symlink in version (str) – Version name of clowder yaml file to link
Raises:	ClowderError –

clowder.util.yaml.load_yaml_file(yaml_file: pathlib.Path, relative_dir: pathlib.Path) → dict¶

Load clowder config from yaml file

Parameters:	yaml_file (Path) – Path of yaml file to load relative_dir (Path) – Directory yaml file is relative to
Returns:	YAML python object
Return type:	dict
Raises:	ClowderError –

clowder.util.yaml.print_clowder_yaml() → None¶: Print current clowder yaml

clowder.util.yaml.save_yaml_file(yaml_output: dict, yaml_file: pathlib.Path) → None¶

Save yaml file to disk

Parameters:	yaml_output (dict) – Parsed YAML python object yaml_file (Path) – Path to save yaml file
Raises:	ClowderError –

clowder.util.yaml.validate_yaml_file(parsed_yaml: dict, file_path: pathlib.Path) → None¶

Validate yaml file

Parameters:	parsed_yaml (dict) – Parsed yaml dictionary file_path (Path) – Path to yaml file
Raises:	ClowderError –

clowder.util.yaml.yaml_string(yaml_output: dict) → str¶

Return yaml string from python data structures

Parameters:	yaml_output (dict) – YAML python object
Returns:	YAML as a string
Return type:	str
Raises:	ClowderError –

Module contents¶

clowder.util module __init__