yabs package

yabs.task_runner

yabs.util module

exception yabs.util.CheckError

Bases: ExitingYabsError

–check condition failed.

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

code

exception code

exception yabs.util.ConfigError

Bases: ExitingYabsError

Invalid yabs.yaml or command line (terminates without stacktrace).

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

code

exception code

exception yabs.util.ExitingYabsError

Bases: SystemExit, YabsError

YabsError the will cause sys.exit(). Raised for errors that don’t need a stack trace.

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

code

exception code

exception yabs.util.YabsError

Bases: RuntimeError

Base class for all exception that we deliberatly throw.

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class yabs.util.NO_DEFAULT

Bases: object

Used as default parameter to distinguish from None.

yabs.util.assert_always(condition, msg=None)

assert even in production code.

yabs.util.check_arg(argument, allowed_types, condition=<class 'yabs.util.NO_DEFAULT'>, *, or_none=False)

Check if argument has the expected type and value.

Note: the exception’s traceback is manipulated, so that the back frame points to the check_arg() line, instead of the actual raise.

Parameters:

• argument (any) – value of the argument to check

• allowed_types (type or tuple of types)

• condition (bool, optional) – additional condition that must be true

• or_none (bool, optional) – defaults to false

Returns:

None

Raises:

• TypeError – if argument is of an unexpected type

• ValueError – if argument does not fulfill the optional condition

• AssertionError – if check_arg was called with a wrong syntax, i.e. allowed_types does not contain types, e.g. if 1 was passed instead of int.

Examples:

def foo(name, amount, options=None):
    check_arg(name, str)
    check_arg(amount, (int, float), amount > 0)
    check_arg(options, dict, or_none=True)

yabs.util.check_cli_verbose(default=3)

Check for presence of –verbose/–quiet or -v/-q without using argparse.

yabs.util.check_dict_keys(d: dict, *, known: set, mandatory: set, prefix: str, key_prefix: str = '') → List[str]

Validate a dict fo missing or unknown keys.

yabs.util.datetime_to_iso(dt=None, *, microseconds: bool = False) → str

Return current UTC datetime as ISO formatted string.

yabs.util.format_elap(seconds, *, count=None, unit='items', high_prec=False, short_suffix=False)

Return elapsed time as H:M:S.h string with reasonable precision.

yabs.util.format_rate(count, time, unit=None, high_prec=False)

Return count / time with reasonable precision.

yabs.util.get_dict_attr(d, key_path, default=<class 'yabs.util.NO_DEFAULT'>)

Return the value of a nested dict using dot-notation path.

Parameters:

• d (dict)

• key_path (str)

Raises:

• KeyError –

• ValueError –

• IndexError –

Examples:

...

Todo

• k[1] instead of k.[1]

• default arg

yabs.util.get_folder_file_names(folder)

Return folder files names as set.

yabs.util.init_logging(verbose=3, path=None)

CLI calls this.

yabs.util.lstrip_string(s, prefix, ignore_case=False)

Remove leading string from s.

Note: This is different than s.lstrip(‘bar’) which would remove all leading ‘a’, ‘b’, and ‘r’ chars.

yabs.util.plural_s(value) → str

Return ‘s’ if value is not singular.

Example

from util import plural_s as ps

v = [1, 2] print(f”Found value{ps(v)}: {v}”)

yabs.util.resolve_path(root, path, must_exist=True, check_root=False)

Return an absolute path, assuming relative to the root file or folder.

yabs.util.run_process_streamed(process, name=None, on_output=None, log_alive=5.0, timeout=None, poll_interval=0.1, flush_min_interval=0.2, prefix_chunks=False)

Read and print output of a running process.

Parameters:

• process (subprocess.Popen) – A running process instance

• name (str) – Descriptive name of the process

• on_output (stream) – Call this method for output (default: logger.info)

• log_alive (float) – Print a simple message if not new output was received for X seconds (default: 5.0)

• timeout (float) – Kill process after X seconds (default: None)

• poll_interval (float) – Poll output buffer every X seconds (default: 0.1)

• flush_min_interval (float) – Minimal log interval: Wait X seconds for more lines (default: 0.2)

• prefix_chunks (bool) – Prefix output chunks with <name> (default: False)

Returns:

tuple (ret_code, output)

yabs.util.search_file_upward(root: Union[str, Path], filename: str, *, max_level=0, or_none=False) → Path

.

yabs.util.set_console_ctrl_handler(ctrl_handler, do_ctrl_c=True, do_ctrl_break=False, do_ctrl_close=False)

The do_ctrl_* functions could simply be sys.exit(1), which will ensure that atexit handlers get called. See https://bugs.python.org/issue35935

Raises:

ctypes.WinError –

Returns:

False if not on Windows or could not register the handler.

yabs.util.shorten_string(long_string, max_chars, min_tail_chars=0, place_holder='[...]')

Return string, shortened to max_chars characters.

long_string = “This is a long string, that will be truncated.” trunacated_string = truncate_string(long_string, max_chars=26, min_tail_chars=11, place_holder=”[…]”) print trunacated_string >> This is a […] truncated.

@param long_string: string to be truncated @param max_chars: max chars of returned string @param min_tail_chars: minimum of tailing chars @param place_holder: place holder for striped content @return: truncated string

yabs.util.timetag(seconds=True, *, ms=False)

Return a time stamp string that can be used as (part of a) filename (also sorts well).

yabs.util.to_list(obj, or_none=False)

Convert a single object to a list.

yabs.util.to_set(obj, or_none=False)

Convert a single object to a set.

yabs.task package

yabs.task.common module

class yabs.task.common.TaskContext(task_runner)

Bases: object

Context information that is passed by the task runner to all tasks. This instance is used by tasks to pass information to downstream tasks.

artifacts: dict

(dict) all files that ‘pypi_release’ created, e.g. {"sdist": <path>, "bdist_msi": <path>}

dry_run: bool

(bool) true if --dry-run was passed

gh_auth_token: str

(str) GitHub authentication token

inc: str

(str) value of --inc argument (‘major’, ‘minor’, ‘patch’, ‘postrelease’)

org_tag_name: str

(str) the repo’s latest tag name (before ‘bump’)

org_version: Version

(semantic_version.Version) latest version (before ‘bump’) E.g. str(context.org_version) == ‘1.2.3-a1’

repo: str

(str) GitHub repo name, e.g. ‘USER/PROJECT’

repo_obj: Repo

(git.repo.base.Repo)

repo_path: Path

(str) Root folder

repo_short: str

(str) Short repo name, without user/ prefixe.g. ‘PROJECT’

tag_name: str

(str) the current tag name (after ‘bump’)

task_runner: TaskRunner

(TaskRunner)

version: Version

(semantic_version.Version) current version (after ‘bump’) E.g. str(context.version) == ‘1.2.3’

version_manager: VersionFileManager

(VersionManager)

class yabs.task.common.WorkflowTask(task_inst: TaskInstance)

Bases: ABC

Common base class for all yabs tasks.

abstract classmethod check_task_def(task_inst: TaskInstance)

Check task definition for errors.

This allows static pre-checks before the actual workflow starts.

Returns:

(str|list|bool) Error message(s)

cli_arg(key: str, default=None)

Return a value from command line args.

Tasks mus handle tha case that args is not undefined when not running as CLI or the command is “info” instead of “running” for example.

get_setup_metadata(extra_args: Optional[list] = None) → dict

‘Query setup.py for project name and version.

abstract classmethod register_cli_command(subparsers, parents, run_parser)

Let tasks add a sub-command and/or arguments to the ‘run’ command.

COMMON_OPTS = frozenset({'dry_run', 'verbose'})

(frozenset) Task options shared by all task

DEFAULT_OPTS: dict = None

(dict) define all supported arguments and their default values. This attribute must be defined by derived classes.

KNOWN_TARGETS = frozenset({'bdist_msi', 'bdist_wheel', 'sdist'})

(frozenset) Task options shared by all task

MANDATORY_OPTS: set = None

(set) mandatory task options. ‘task’ is implicitly mandatory. This is validated by the task runner before starting the workflow.

dry_run: bool

(bool) true if –dry-run was passed to the CLI

opts: dict

(dict) The actual arguments, i.e. the default values merged with (dict) passed options

task_inst: TaskInstance

(TaskInstance)

verbose: int

(int, default=3) 0..5