chaintool package

Submodules

chaintool.cli module

Parse command-line args and invoke the requested operation.

chaintool.cli.main(forced_progname=None)[source]

Parse command-line args from sys.argv, and dispatch to handlers.

Call chaintool.init() to make sure the chaintool files and configurations exist and are updated to the current version. Then parse the commandline and dispatch the appropriate handler.

Parameters

forced_progname (str | None, optional) – program name to use in help output; if None, then sys.argv[0] will be used; defaults to None

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command module

Top-level logic for “cmd” operations.

Called from cli module. Handles locking and shortcuts/completions; delegates to command_impl_* modules for most of the work.

Note that most locks acquired here are released only when the program exits. Operations are meant to be invoked one per program instance, using the CLI.

chaintool.command.cli_del(delcmds, ignore_seq_usage)[source]

Delete one or more commands.

If ignore_seq_usage is False, acquire the seq inventory readlock and item readlocks on all sequences.

Acquire the cmd inventory writelock and item writelocks on the commands to delete.

If ignore_seq_usage is False, check all the given commands to make sure that they are not currently contained in any sequence (reject if so).

Delete each command (via command_impl_op.delete()), and tear down its shortcut (shortcuts.delete_cmd_shortcut()) and autocompletion behavior (completions.delete_completion()).

Parameters
  • delcmds (list[str]) – names of commands to delete

  • ignore_seq_usage (bool) – if True, don’t validate that commands are unused by current sequences

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command.cli_edit(cmd, print_after_set)[source]

Interactively create or update a command.

Acquire the seq inventory readlock, the cmd inventory writelock, and the cmd item writelock. Read the current command’s commandline (if it exists).

If we’re creating a new command, check to see whether a sequence of this same name already exists (reject if so). Then create a temporary empty command that we can edit.

Release the inventory locks and let the user interactively edit any existing commandline. The delegate to command_impl_op.define() to create/update the command.

Finally: if we successfully created a new command, set up its shortcut (shortcuts.create_cmd_shortcut()) and autocompletion behavior (completions.create_completion()).

Parameters
  • cmd (str) – name of command to create/update

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command.cli_list(column)[source]

Print the list of current command names.

No locking needed. Just read a directory list and print it.

Parameters

column (bool) – if True, print as one command name per line

Returns

exit status code; currently always returns 0

Return type

int

chaintool.command.cli_print(cmd)[source]

Pretty-print the info for a command.

Delegate to command_impl_print.print_one().

Parameters

cmd (str) – name of command to print

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command.cli_print_all()[source]

Pretty-print the info for all commands.

Acquire the cmd inventory readlock and get the list of all commands. Readlock those commands and delegate to command_impl_print.print_multi() to pretty-print the info for that list of commands.

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command.cli_run(cmd, quiet, args)[source]

Run a command.

Acquire the cmd item readlock. Create a temporary directory using a context manager. While the temp directory exists, grab its name for the value of the “tempdir” reserved placeholder, and delegate to command_impl_op.run() to execute the command. Finally, print a warning if any of the given placeholder args were irrelevant for this command.

Note that command_impl_op.run() may modify args (for use with subsequent commands in the sequence).

Parameters
  • cmd (str) – name of command to run

  • quiet (bool) – whether to print only the command output

  • args (list[str]) – placeholder arguments for this run; to modify

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command.cli_set(cmd, cmdline, overwrite, print_after_set)[source]

Create or update a command to consist of the given commandline.

Acquire the seq inventory readlock, the cmd inventory writelock, and the cmd item writelock. If we’re creating a new command, check to see whether a sequence of this same name already exists (reject if so).

Delegate to command_impl_op.define() to create/update the command.

Finally: if we successfully created a new command, set up its shortcut (shortcuts.create_cmd_shortcut()) and autocompletion behavior (completions.create_completion()).

Parameters
  • cmd (str) – name of command to create/update

  • cmdline (str) – commandline

  • overwrite (bool) – whether to allow if command already exists

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command.cli_vals(cmd, args, print_after_set)[source]

Update placeholder values for a command.

Acquire the cmd item writelock. Delegate to command_impl_op.vals() to update this command. Finally, print a warning if any of the given placeholder args were irrelevant for this command.

Parameters
  • cmd (str) – name of command to update

  • args (list[str]) – new placeholder value settings

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command.cli_vals_all(placeholder_args)[source]

Update placeholder values for all commands.

Acquire the cmd inventory readlock and get the list of all commands. Writelock those commands and delegate to command_impl_op.vals() to update each command. Finally, print a warning if any of the given placeholder args were irrelevant for all commands.

Parameters
  • cmd (str) – name of command to update

  • args (list[str]) – new placeholder value settings

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command_impl_core module

Utilities called by various modules to read/write command files.

chaintool.command_impl_core.all_names()[source]

Get the names of all current commands.

Return the filenames in the commands directory.

Returns

current command names

Return type

list[str]

chaintool.command_impl_core.create_temp(cmd)[source]

Create an empty command used to “reserve the name” during edit-create.

If the command is being created by interactive edit, an empty-valued temporary YAML document is first created via this function, so that the inventory lock doesn’t need to be held during the edit.

Parameters

cmd (str) – name of command to make a temp document for

chaintool.command_impl_core.exists(cmd)[source]

Test whether the given command already exists.

Return whether a file of name cmd exists in the commands directory.

Parameters

cmd (str) – name of command to check

Returns

whether the given command exists

Return type

bool

chaintool.command_impl_core.init(_prev_version, _cur_version)[source]

Initialize module.

Called when chaintool runs. Creates the commands directory, inside the data appdir, if necessary.

Parameters
  • _prev_version (str) – version string of previous chaintool run; not used

  • _cur_version (str) – version string of current chaintool run; not used

chaintool.command_impl_core.read_dict(cmd)[source]

Fetch the contents of a command as a dictionary.

From the commands directory, load the YAML for the named command. Return its properties as a dictionary.

Parameters

cmd (str) – name of command to read

Raises

FileNotFoundError if the command does not exist

Returns

dictionary of command properties/values

Return type

dict[str, str]

chaintool.command_impl_core.write_dict(cmd, cmd_dict, mode)[source]

Write the contents of a command as a dictionary.

Dump the command dictionary into a YAML document and write it into the commands directory.

Parameters
  • cmd (str) – name of command to write

  • cmd_dict (dict[str, str]) – dictionary of command properties/values

  • mode ("w" | "x") – mode used in the open-to-write

Raises

FileExistsError if mode is “x” and the command exists

chaintool.command_impl_op module

Low-level logic for “cmd” operations (other than pretty-printing).

Called from command, sequence, sequence_impl_op, and xfer modules. Does the bulk of the work for creating/modifying/executing/deleting command definitions.

class chaintool.command_impl_op.ReservedPlaceholdersCtx(stdout: Optional[str] = None, stdout_requested: bool = False, tempdir: Optional[str] = None)[source]

Bases: object

Info shared among commands to implement reserved placeholders.

stdout: str = None
stdout_requested: bool = False
tempdir: str = None
chaintool.command_impl_op.define(cmd, cmdline, overwrite, print_after_set, compact)[source]

Create or update a command to consist of the given commandline.

Do some initial validation of cmd and cmdline to check that they are non-empty and consist of legal characters.

Make a wrapper for handle_set_placeholder() to capture the mutable containers that we’ll be updating. This results in a function that can accumulate placeholder and error info while processing the placeholder tokens. Pass the input commandline and the wrapper function to process_cmdline() to get the format string for the commandline.

Call print_errors() to print about any detected errors, and if there are any, bail out with error status.

Store the input commandline, the generated format string, and the accumulated placeholder info in the command dictionary.

Finally, if print_after_set is True, pretty-print the command that we just created/updated.

Parameters
  • cmd (str) – name of command to create/update

  • cmdline (str) – commandline

  • overwrite (bool) – whether to allow if command already exists

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

  • compact (bool) – whether to reduce the use of newlines (used when caller is processing many commands)

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command_impl_op.delete(cmd, is_not_found_ok)[source]

Delete a command.

Delete the file of name cmd in the commands directory.

If that file does not exist, and is_not_found_ok is False, then raise a FileNotFoundError exception.

Parameters
  • cmd (str) – names of command to delete

  • is_not_found_ok (bool) – whether to silently accept already-deleted case

Raises

FileNotFoundError if the command does not exist and is_not_found_ok is False

chaintool.command_impl_op.run(cmd, quiet, args, unused_args, rsv_ctx)[source]

Run a command.

Apply the placeholder values from the args list (as well as any special reserved-placeholder values) to the relevant values of the command dictionary, and update unused_args, by calling command_with_values(). If that fails, bail out with error status.

Generate the commandline to execute by using the keys/values from this command dictionary with the command’s format string. Invoke virtual_tools.dispatch() to see whether the command is a “virtual tool” that should be executed internally (and do so). If so, then return the status from the virtual tool. If not, then execute the commandline via subprocess.run() and return its exit status.

Note that if rsv_ctx.stdout_requested is False, the output of the command will be printed as it is generated, and rsv_ctx.stdout will be set to None. On the other hand if it is True, the output of the command will be captured. Then it will be printed and assigned to rsv_ctx.stdout.

Note also that virtual_tools.dispatch() may modify args.

Parameters
  • cmd (str) – name of command to run

  • quiet (bool) – whether to print only the command output

  • args (list(str)) – placeholder arguments for this run; to modify

  • unused_args (list[str]) – placeholder arguments unused by any command in current sequence; to modify

  • rsv_ctx (ReservedPlaceholdersCtx) – contains stdout from prev cmd (if needed here) and will contain stdout for next (if requested); to modify

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command_impl_op.vals(cmd, args, unused_args, print_after_set, compact)[source]

Update placeholder values for a command.

Apply the placeholder values from the args list to the relevant values of the command dictionary, and update unused_args, by calling command_with_values(). If that fails, bail out with error status.

Call update_cmdline() to update the stored commandline to match the new placeholder values, and write back the new command dictionary.

Finally, if print_after_set is True, pretty-print the command that we just updated.

Parameters
  • cmd (str) – name of command to update

  • args (list(str)) – placeholders to update, with values

  • unused_args (list[str]) – placeholder arguments unused by any command in current sequence; to modify

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.command_impl_print module

Low-level logic for “cmd” operations related to pretty-printing.

Called from command, sequence, command_impl, and sequence_impl_op modules.

chaintool.command_impl_print.print_multi(commands, ignore_env)[source]

Pretty-print the info for multiple commands.

Call init_print_info_collections() to build the info and associations that will be used for pretty-printing. At the coarsest level, placeholders will be bucketed as “required values”, “optional values”, and “toggles”.

Define a sort function that will be used when arranging placeholders by “command group” (commands that use those placeholders)… we want to list larger groups first, and among command-groups of the same size, order them by how early their first command appears in the list of commands.

For each of the top-level buckets of placeholder types (if non-empty), pretty-print a header and call print_placeholders_set() to print the info for those placeholders.

Parameters
  • commands (list[str]) – names of commands to print

  • ignore_env (bool) – whether to ignore the effects of chaintool-env; will be True for non-sequence prints

Returns

exit status code; currently always returns 0

Return type

int

chaintool.command_impl_print.print_one(cmd)[source]

Pretty-print the info for a command.

Read the command dictionary, and bail out if it does not exist.

Pretty-print the placeholder info separated into “required values” (no default), “optional values”, and “toggles”.

Parameters

cmd (str) – name of command to print

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.completions module

Create/delete bash completions for commands and sequences.

The bash function for the autocompletions of the main “chaintool” utility, and the “complete” command that associates that function with that utility, are contained in a “main script” file that is placed in the “completions” directory under the data appdir.

The bash function for shortcut-script autocompletions is defined in a “helper script” file in the same “completions” directory.

The “completions/shortcuts” directory contains individual files, one per shortcut, that contain the “complete” command associating that function with that shortcut.

An “omnibus” file that sources all of the above files is also placed in the “completions” directory.

If the user has chosen to do old-style (non-dynamic) completions loading, then the “omnibus” file can be sourced from their shell startup script. This will define all the necessary functions and run all necessary “complete” commands, but only at shell startup time (so changes will not be picked up until a new shell starts).

For dynamic completions loading, the user must specify a directory from which completion scripts will be lazy-loaded. A file will be placed there which sources the “main script” file. A file will also be placed there for each shortcut; each such file will source the “helper script” if necessary and also run the “complete” command for that shortcut.

chaintool.completions.create_completion(item_name)[source]

Create a completion for a shortcut.

Invoke create_static() to create the always-generated file for old-style completions. Then, if dynamic completions are enabled, also invoke create_lazyload() to create the file in the user dir for lazy-load scripts.

Parameters

item_name (str) – shortcut name

chaintool.completions.create_lazyload(item_name)[source]

Create the per-shortcut file in the user dir for lazy-load scripts.

Called when creating a new shortcut if dynamic completions are enabled. Also called when enabling dynamic completions when some shortcuts already exist.

Read the user dir location from the USERDIR_LOCATION choicefile, and create the per-shortcut file there that will: source the “main script” if necessary, source the “helper script” if necessary, then invoke the “complete” command.

Parameters

item_name (str) – shortcut name

chaintool.completions.delete_completion(item_name)[source]

Delete a completion for a shortcut.

Invoke delete_static() to delete the always-generated file for old-style completions. Then, if dynamic completions are enabled, also invoke delete_lazyload() to delete the file in the user dir for lazy-load scripts.

Parameters

item_name (str) – shortcut name

chaintool.completions.delete_lazyload(item_name)[source]

Delete the per-shortcut file in the user dir for lazy-load scripts.

Parameters

item_name (str) – shortcut name

chaintool.completions.init(prev_version, cur_version)[source]

Initialize module.

Called when chaintool runs. Creates the completions directory, inside the data appdir, if necessary. Also creates the shortcuts completions directory inside that, if necessary.

Once those directories are ensured, the “main script” and “helper script” files can be extracted from the package resources and placed in the completions directory. The “omnibus” file is also created there. These three files will be (re)created if missing or if the chaintool version has changed since last run.

Parameters
  • prev_version (str) – version string of previous chaintool run

  • cur_version (str) – version string of current chaintool run

chaintool.completions_setup module

Handle configuring or disabling the bash completions feature.

chaintool.completions_setup.configure()[source]

Set up or disable bash autocompletions.

If there is no current bash autocompletions setup for chaintool, let the user choose the style of setup to use (if any). For “dynamic” completions with lazy script loads, the default for the necessary user directory is taken from get_userdir_path(); for old-style completions the default path of script-to-modify is taken from shared.get_startup_script_path(). (But in either case the user can enter the path of their choice to edit/replace the default.)

On the other hand if bash autocompletions are currently configured, give the user the option of undoing that configuration.

Returns

exit status code; currently always returns 0

Return type

int

chaintool.completions_setup.probe_config(ask_to_change)[source]

Determine existing completions setup; optionally offer to change.

If a completions configuration does not exist, or does not check out as valid (via check_dynamic() or :func`check_oldstyle`), then return False.

Then if ask_to_change is False, immediately return True.

Otherwise, ask the user if they want to preserve the existing setup. If they do, return True. Otherwise use disable_dynamic() or disable_oldstyle() to attempt removing the current setup, returning the boolean inverse or whatever that disable operation returns.

Parameters

ask_to_change (bool) – whether to offer to change a current valid setup

Returns

whether there is an existing setup that has been preserved

Return type

bool

chaintool.info module

Dump info about current configuration.

chaintool.info.dump()[source]

Print current appdirs and shortcuts/completions configuration.

Returns

exit status code; currently always returns 0

Return type

int

chaintool.locks module

Locking system to preserve consistency under simultaneous operations.

Acquire WRITE inventory lock to create or delete item-of-type. Any inventory lock prevents other create/delete for item-of-type.

Acquire WRITE item lock to create, delete, or modify an item. Any item lock prevents other create/delete/modify for that item.

This simple R/W lock implementation does not enforce all the guardrails necessary to prevent deadlock. Because its usage is pretty simple in this program, we just have to follow conventions to avoid deadlock (knock on wood). The conventions are:

  • lock acquisition order: seq inventory, seq item, cmd inventory, cmd item

  • for holding multiple item locks, acquire in sorted item name order (this is actually enforced as long as you use multi_item_lock to do it)

Also note that item locks (and in some cases inventory locks) are released only when the program exits, using an atexit handler. Operations are meant to be invoked one per program instance, using the CLI.

class chaintool.locks.LockType(value)[source]

Bases: enum.Enum

Enum used to differentiate readlocks and writelocks.

READ = 'read'
WRITE = 'write'
chaintool.locks.init()[source]

Initialize module.

Called when chaintool runs. Creates the locks directory, inside the cache appdir, if necessary.

chaintool.locks.inventory_lock(item_type, lock_type)[source]

Create an inventory lock.

Delegate to lock_internal() with a prefix indicating an inventory lock for this item type.

Parameters
  • item_type ("cmd" | "seq") – whether this is for commands or sequences

  • lock_type (LockType.WRITE | LockType.READ) – whether this is writelock or readlock

chaintool.locks.item_lock(item_type, item_name, lock_type)[source]

Create an individual item lock.

Delegate to lock_internal() with a prefix indicating an item lock for this item type and specific item name.

Parameters
  • item_type ("cmd" | "seq") – whether this is for commands or sequences

  • item_name (str) – name of the command or sequence to lock

  • lock_type (LockType.WRITE | LockType.READ) – whether this is writelock or readlock

chaintool.locks.multi_item_lock(item_type, item_name_list, lock_type)[source]

Create multiple item locks.

Sort the list of item names and then lock each via item_lock().

Parameters
  • item_type ("cmd" | "seq") – whether this is for commands or sequences

  • item_name_list (list[str]) – names of the commands or sequences to lock

  • lock_type (LockType.WRITE | LockType.READ) – whether this is writelock or readlock

chaintool.locks.release_inventory_lock(item_type, lock_type)[source]

Remove an inventory lock.

Delete the lockfile with matching prefix and lock type, and with a PID suffix matching the current process PID.

Parameters
  • item_type ("cmd" | "seq") – whether this is for commands or sequences

  • lock_type (LockType.WRITE | LockType.READ) – whether this is writelock or readlock

chaintool.sequence module

Top-level logic for “seq” operations.

Called from cli module. Handles locking and shortcuts/completions; delegates to sequence_impl_* modules for most of the work.

Note that most locks acquired here are released only when the program exits. Operations are meant to be invoked one per program instance, using the CLI.

chaintool.sequence.cli_del(delseqs)[source]

Delete one or more sequences.

Acquire the seq inventory writelock, and item writelocks on the sequences to delete. Delete each sequence (via sequence_impl_op.delete()), and tear down its shortcut (shortcuts.delete_seq_shortcut()) and autocompletion behavior (completions.delete_completion()).

Parameters

delseqs (list[str]) – names of sequences to delete

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.sequence.cli_edit(seq, ignore_undefined_cmds, print_after_set)[source]

Interactively create or update a sequence.

Acquire the seq inventory and item writelocks. Read the current sequence command list (if it exists).

If we’re creating a new sequence, also acquire the cmd inventory lock and check to see whether a command of this same name already exists (reject if so). Then create a temporary empty sequence that we can edit.

Release the inventory locks and let the user interactively edit any existing list of cmds. The delegate to sequence_impl_op.define() to create/update the sequence.

Finally: if we successfully created a new sequence, set up its shortcut (shortcuts.create_seq_shortcut()) and autocompletion behavior (completions.create_completion()).

Parameters
  • seq (str) – name of sequence to create/update

  • ignore_undefined_cmds (bool) – if True, don’t validate that commands exist

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.sequence.cli_list(column)[source]

Print the list of current sequence names.

No locking needed. Just read a directory list and print it.

Parameters

column (bool) – if True, print as one sequence name per line

Returns

exit status code; currently always returns 0

Return type

int

chaintool.sequence.cli_print(seq)[source]

Pretty-print the info for all commands in a sequence.

Acquire the seq item readlock and cmd inventory readlock. Read the sequence’s command list. Readlock those commands and delegate to command_impl_print.print_multi() to pretty-print the info for that list of commands.

Parameters

seq (str) – name of sequence to print

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.sequence.cli_run(seq, quiet, args, ignore_errors, skip_cmdnames)[source]

Run a sequence.

Acquire the seq item readlock and cmd inventory readlock. Read the sequence’s command list, then acquire readlocks on those commands and release the cmd inventory readlock.

Do a pre-pass through the commands to see which ones want to consume the stdout of the previous command; we’ll use that info to correctly populate stdout_requested in the reserved-placeholders context passed to each command execution.

Create a temporary directory using a context manager. While the temp directory exists, grab its name for the value of the “tempdir” reserved placeholder, and delegate to command_impl_op.run() to execute each command in the list that is not a member of skip_cmdnames. If a command returns an error status and ignore_errors is false, bail out.

In the success case, finally print a warning if any of the given placeholder args were irrelevant for all the executed commands.

Note that args may be modified during the process of running commands.

Parameters
  • seq (str) – name of sequence to run

  • quiet (bool) – whether to print only the command output

  • args (list[str]) – placeholder arguments for this run; to modify

  • ignore_errors (bool) – if True, a command error does not stop the run

  • skip_cmdnames (list[str]) – list of command names to not execute

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.sequence.cli_set(seq, cmds, ignore_undefined_cmds, overwrite, print_after_set)[source]

Create or update a sequence to consist of the given list of commands.

Acquire the seq inventory and item writelocks. If we’re creating a new sequence, also acquire the cmd inventory readlock and check to see whether a command of this same name already exists (reject if so).

Delegate to sequence_impl_op.define() to create/update the sequence.

Finally: if we successfully created a new sequence, set up its shortcut (shortcuts.create_seq_shortcut()) and autocompletion behavior (completions.create_completion()).

Parameters
  • seq (str) – name of sequence to create/update

  • cmds (list[str]) – list of command names to form the sequence

  • ignore_undefined_cmds (bool) – if True, don’t validate that commands exist

  • overwrite (bool) – whether to allow if sequence already exists

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.sequence.cli_vals(seq, args, print_after_set)[source]

Update placeholder values for all commands in a sequence.

Acquire the seq item writelock and cmd inventory readlock. Read the sequence’s command list, then acquire writelocks on those commands and release the cmd inventory readlock.

Delegate to command_impl_op.vals() to update each command in the list. If any change results from this, and print_after_set is True, then pretty-print the new sequence.

Finally, print a warning if any of the given placeholder args were irrelevant for all of the sequence’s commands.

Parameters
  • seq (str) – name of sequence to process

  • args (list[str]) – new placeholder value settings

  • print_after_set (bool) – whether to automatically trigger “print” operation if any change results

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.sequence_impl_core module

Utilities called by various modules to read/write sequence files.

chaintool.sequence_impl_core.all_names()[source]

Get the names of all current sequences.

Return the filenames in the sequences directory.

Returns

current sequence names

Return type

list[str]

chaintool.sequence_impl_core.create_temp(seq)[source]

Create an empty sequence used to “reserve the name” during edit-create.

If the sequence is being created by interactive edit, an empty-valued temporary YAML document is first created via this function, so that the inventory lock doesn’t need to be held during the edit.

Parameters

seq (str) – name of sequence to make a temp document for

chaintool.sequence_impl_core.exists(seq)[source]

Test whether the given sequence already exists.

Return whether a file of name seq exists in the sequences directory.

Parameters

seq (str) – name of sequence to check

Returns

whether the given sequence exists

Return type

bool

chaintool.sequence_impl_core.init(_prev_version, _cur_version)[source]

Initialize module.

Called when chaintool runs. Creates the sequences directory, inside the data appdir, if necessary.

Parameters
  • _prev_version (str) – version string of previous chaintool run; not used

  • _cur_version (str) – version string of current chaintool run; not used

chaintool.sequence_impl_core.read_dict(seq)[source]

Fetch the contents of a sequence as a dictionary.

From the sequences directory, load the YAML for the named sequence. Return its properties as a dictionary.

Parameters

seq (str) – name of sequence to read

Raises

FileNotFoundError if the sequence does not exist

Returns

dictionary of sequence properties/values

Return type

dict[str, str]

chaintool.sequence_impl_core.write_dict(seq, seq_dict, mode)[source]

Write the contents of a sequence as a dictionary.

Dump the sequence dictionary into a YAML document and write it into the sequences directory.

Parameters
  • seq (str) – name of sequence to write

  • seq_dict (dict[str, str]) – dictionary of sequence properties/values

  • mode ("w" | "x") – mode used in the open-to-write

Raises

FileExistsError if mode is “x” and the sequence exists

chaintool.sequence_impl_op module

Low-level logic for “seq” operations.

Called from sequence, command, and xfer modules. Does the bulk of the work for creating/modifying/deleting sequence definitions.

chaintool.sequence_impl_op.define(seq, cmds, undefined_cmds, overwrite, print_after_set, compact)[source]

Create or update a sequence to consist of the given commands.

Do some initial validation of seq and cmds to check that they are non-empty and consist of legal characters.

If undefined_cmds is non-empty, print an error and bail out. (This list was already generated for us by the caller to avoid the need for inventory lock acquisition in this module).

Store the commands list in the sequence dictionary.

Finally, if print_after_set is True, pretty-print the sequence that we just created/updated.

Parameters
  • seq (str) – name of sequence to create/update

  • cmds (list[str]) – names of commands to make up the sequence

  • undefined_cmds (list[str]) – names of commands specified for the sequence that do not exist (if not-exist is an error condition)

  • overwrite (bool) – whether to allow if sequence already exists

  • print_after_set (bool) – whether to automatically trigger “print” operation at the end

  • compact (bool) – whether to reduce the use of newlines (used when caller is processing many sequences)

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.sequence_impl_op.delete(seq, is_not_found_ok)[source]

Delete a sequence.

Delete the file of name seq in the sequences directory.

If that file does not exist, and is_not_found_ok is False, then raise a FileNotFoundError exception.

Parameters
  • seq (str) – names of sequence to delete

  • is_not_found_ok (bool) – whether to silently accept already-deleted case

Raises

FileNotFoundError if the sequence does not exist and is_not_found_ok is False

chaintool.shared module

Constants and utility functions shared by the package’s modules.

chaintool.shared.check_shell()[source]

Attempt to determine the user’s login shell.

The first bool in the returned tuple simply indicates whether the SHELL environment variable is set. The second bool is true only if SHELL is set to a string that ends in “/bash”.

Returns

a tuple of: whether there is a SHELL, and whether it is bash

Return type

tuple[bool, bool]

chaintool.shared.delete_if_exists(filepath)[source]

Delete a file, silently succeeding if it is already gone.

Delegate to os.remove() and swallow any FileNotFoundError exception.

Parameters

filepath (str) – file to delete

chaintool.shared.editline(prompt, oldline, all_completions=None)[source]

Interactively edit a string.

Print the prompt and the old/default string value (oldline), and allow the user to readline-edit that string value. When the user presses enter, return that edited value.

If all_completions is set to some list of valid autocompletions, also set up the appropriate readline hooks to implement tab-completion for those strings. Otherwise block tab-completion.

Just before returning the result string, make sure to revert any hooks we set.

Parameters
  • prompt (str) – prompt to print before the editable string

  • oldline (str) – original value to present for editing

  • all_completions (list[str] | None, optional) – valid autocompletions, if any; defaults to None

Returns

user-edited new version of the string

Return type

str

chaintool.shared.errprint(msg)[source]

Print an error message.

Print msg to stderr, in red.

Parameters

msg (str) – error message

chaintool.shared.get_last_chaintool_version()[source]

Return the last-used chaintool version.

Returns

chaintool version from previous run

Return type

str

chaintool.shared.get_last_python_version()[source]

Return the last-used Python version.

Returns

Python version from previous run

Return type

str

chaintool.shared.get_last_schema_version()[source]

Return the last-used schema version.

Note that if the choicefile for schema version is absent, that corresponds to version 0.

Returns

schema version from previous run

Return type

int

chaintool.shared.get_startup_script_path()[source]

Interactively get a shell startup script path from the user

Starting with a suggested value from default_startup_script(), ask the user to enter/edit the path. Return what the user provides, or None if that filepath does not exist.

Returns

startup script pathm, if valid

Return type

str | None

chaintool.shared.init()[source]

Initialize module.

Called when chaintool runs. Creates the locations directory, inside the config appdir, if necessary. Also configure readline to allow tab-completion during seq edit.

chaintool.shared.is_valid_name(name)[source]

Check that the given string is valid as a cmd or seq name.

Return False if name is emptystring or contains whitespace.

Parameters

name (str) – name to check

Returns

whether the name is valid

Return type

bool

chaintool.shared.read_choicefile(choicefile_path)[source]

Return the file contents (choice string), if the file exists.

Return a string containing the contents of the given file, or None if that file does not exist.

Parameters

choicefile_path (str) – file to read

Returns

file contents if it exists

Return type

str | None

chaintool.shared.remove_script_additions(script_path, begin_mark, end_mark, expected_lines)[source]

Strip lines delineated by comment markers from a script file.

Open the given script file and look for the marker comments. See if removing lines from the beginning marker to the end marker would result in the expected number of lines removed. If so: make a backup copy of the script file, write back the updated script file with lines removed, and return True.

Otherwise leave the script file unmodified and return False.

Parameters
  • script_path (str) – path to the scriptfile to edit

  • begin_mark (str) – comment line marking start of section-to-remove

  • end_mark (str) – comment line marking end of section-to-remove

  • expected_lines (int) – number of lines that should be removed, including the marker comments

Returns

whether lines were removed from the script file

Return type

bool

chaintool.shared.set_last_chaintool_version(version_str)[source]

Update the stored last-used chaintool version.

Parameters

version_str (str) – chaintool version to write

chaintool.shared.set_last_python_version(version_str)[source]

Update the stored last-used Python version.

Parameters

version_str (str) – Python version to write

chaintool.shared.set_last_schema_version(version)[source]

Update the stored last-used schema version.

Parameters

version (int) – schema version to write

chaintool.shared.write_choicefile(choicefile_path, choice)[source]

Write the choice string to the file, or delete the file if no choice.

Write the given choice string to the given file, or delete that file (via delete_if_exists()) if choice is None.

Parameters
  • choicefile_path (str) – file to write/delete

  • choice (str | None) – file contents if any

chaintool.shortcuts module

Create/delete “shortcut” scripts for commands and sequences.

These scripts allow doing a command or sequence run operation with less typing. They are placed in a directory that can be added to the user’s PATH.

chaintool.shortcuts.create_cmd_shortcut(cmd_name)[source]

Create a shortcut script for a command.

Delegate to create_shortcut() with the “cmd” type and the given name.

Parameters

cmd_name (str) – name of the command to make a shortcut for

chaintool.shortcuts.create_seq_shortcut(seq_name)[source]

Create a shortcut script for a sequence.

Delegate to create_shortcut() with the “seq” type and the given name.

Parameters

seq_name (str) – name of the sequence to make a shortcut for

chaintool.shortcuts.delete_cmd_shortcut(cmd_name)[source]

Delete a shortcut script for a command.

Remove the file with the given name in the shortcuts directory.

Parameters

cmd_name (str) – name of the command to delete a shortcut for

chaintool.shortcuts.delete_seq_shortcut(seq_name)[source]

Delete a shortcut script for a sequence.

Remove the file with the given name in the shortcuts directory.

Parameters

seq_name (str) – name of the sequence to delete a shortcut for

chaintool.shortcuts.init(_prev_version, _cur_version)[source]

Initialize module.

Called when chaintool runs. Creates the shortcuts directory, inside the data appdir, if necessary.

Parameters
  • _prev_version (str) – version string of previous chaintool run; not used

  • _cur_version (str) – version string of current chaintool run; not used

chaintool.shortcuts_setup module

Handle setting/unsetting the PATH modification for shortcuts.

chaintool.shortcuts_setup.configure()[source]

Set or remove a PATH modification to include the shortcuts directory.

If the PATH needs to be modified, give the user a chance to set that modification in the startup script file of their choice (default script path taken from shared.get_startup_script_path()). If it is already modified, give the user a chance to undo that modification.

Returns

exit status code; currently always returns 0

Return type

int

chaintool.shortcuts_setup.probe_config(ask_to_change)[source]

Determine existing PATH modification; optionally offer to change.

Work through various possibilities of whether the current PATH setting includes the desired shortcuts directory, and whether there is an existing script file that has already been modified to set that. If there is no current (valid) configuration for such a modified script file, and we are at (or can get to) a clean state for doing future modifications, then do any necessary cleanup and return False.

In the case where the dir is already in the PATH but we don’t have a record of modifying a script file … we’re not sure how to proceed, so return True to indicate we won’t be auto-changing things.

If there is already a modified script file setting PATH appropriately, inform the user. If ask_to_change is False, immediately return True. Otherwise, call unconfigure() to see if they want to undo the script modification. Return the boolean inverse of whatever unconfigure() returns.

Parameters

ask_to_change (bool) – whether to offer to change a current valid config

Returns

whether there is an existing config that has been preserved

Return type

bool

chaintool.virtual_tools module

Implementation of commands that run internally to chaintool.

Implements chaintool-copy, chaintool-del, and chaintool-env “virtual tools” that run in Python code here rather than in a subprocess shell on the OS.

chaintool.virtual_tools.copytool(copy_args, _run_args)[source]

Implement chaintool-copy for platform-independent file copy.

Bail out with error if copy_args has other than 2 elements.

Otherwise, treat first element as copy source and second element as copy dest. Delegate to shutil.copy2 to do the copy. If copy2 raises any exception, return an error.

Parameters
  • copy_args (list[str]) – arguments to chaintool-copy

  • _run_args (list[str]) – arguments to “seq/cmd run”, not used in this function

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.virtual_tools.deltool(del_args, _run_args)[source]

Implement chaintool-del for platform-independent file delete.

Bail out with error if del_args has other than 1 element.

Otherwise, treat that element as the filepath to delete. Delegate to shared.delete_if_exists() to do the delete. If that raises any exception, return an error.

Parameters
  • del_args (list[str]) – arguments to chaintool-del

  • _run_args (list[str]) – arguments to “seq/cmd run”, not used in this function

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.virtual_tools.dispatch(cmdline, run_args)[source]

Run a “virtual tool” for a commandline, if appropriate.

If the first word of the given commandline is not a key in VTOOL_DISPATCH, return None.

Otherwise pass the remaining words from that commandline, as well as any runtime-specified placeholder args, to the virtual tool function selected by that first word.

Parameters
  • cmdline (str) – commandline for the command to run

  • run_args (list[str]) – arguments to “seq/cmd run”; to modify

Returns

exit status code, or None if no virtual tool

Return type

int | None

chaintool.virtual_tools.envtool(env_args, run_args)[source]

Implement chaintool-env to modify runtime placeholder values.

Parse the given chaintool-env arguments (in env_args) to get a list of environment ops. Bail out with error if any are invalid.

Iterate through the list of ops. Since the run_args will be applied sequentially for subsequent commands, it is sufficient just to add the var assignment generated by each op as a new element at the end of run_args.

Parameters
  • env_args (list[str]) – arguments to chaintool-env

  • run_args (list[str]) – arguments to “seq/cmd run”; to modify

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.virtual_tools.update_env(cmdline, env_values)[source]

Get the optional placeholder values set by a commandline.

This utility function is invoked during printing placedholder info for commands in a sequence; it determines whether a command affects how placeholder values will be shown for subsequent commands in the sequence.

Only the “chaintool-env” command can affect subsequent commands in this way, so return immediately if the first cmdline word doesn’t match that. Also return if there is any error parsing the remaining words of a “chaintool-env” command into a list of environment ops.

Iterate through the list of ops and add each op’s placeholder name/value to the env_values dict.

Parameters
  • cmdline (str) – commandline for the command to examine

  • env_values (dict[str, str]) – dict of optional placeholder values, keyed by placeholder name; to modify

chaintool.xfer module

Top-level logic for “export” and “import” operations.

Called from cli module. Handles locking and shortcuts/completions; delegates to command_impl_* and sequence_impl_* modules for most of the work.

Note that most locks acquired here are released only when the program exits. Operations are meant to be invoked one per program instance, using the CLI.

chaintool.xfer.cli_export(export_file)[source]

Export all current commands and sequences to a file.

Acquire the seq and cmd inventory readlocks, get all sequence and command names, and readlock all those items.

Open the given file and write a YAML doc to it. Commands (from command_impl_core.read_dict()) are written to a list value for the “commands” property, and sequences (from sequence_impl_core.read_dict()) similarly to the “sequences” property. The “schema_version” is also written, to help interpret this file if its format changes in the future.

Parameters

export_file (str) – filepath to write to

Returns

exit status code (0 for success, nonzero for error)

Return type

int

chaintool.xfer.cli_import(import_file, overwrite)[source]

Import commands and sequences from a filepath or an http/https URL.

Acquire the seq and cmd inventory writelocks. If overwrite is True, get all sequence and command names, and writelock all those items.

Open the given file or URL and read a YAML doc from it. Commands are read from a list value for the “commands” property, and sequences similary from the “sequences” property. The overwrite argument is passed along to command and sequence creation (via command_impl_op.define() and sequence_impl_op.define()) to control whether an imported item is allowed to replace an existing item of the same name.

For each successfully created item, also set up its shortcut (shortcuts.create_seq_shortcut() or shortcuts.create_cmd_shortcut()) and autocompletion behavior (completions.create_completion()).

Parameters
  • import_file (str) – filepath or http/https URL to read from

  • overwrite (bool) – whether to allow replacing existing items (note this does NOT allow conflict between command name and sequence name)

Returns

exit status code; currently always returns 0

Return type

int

Module contents

Initialize the package’s modules.

chaintool.current_export_schema_ver()[source]

Return the export schema version understood by this chaintool version.

Delegate to schema_ver_for_package_ver() using the current chaintool version and the schema-change info from EXPORT_SCHEMA_CHANGE_VERSIONS.

Returns

export schema version for the given package version, or None if the current chaintool version string could not be evaluated

Return type

int | None

chaintool.init()[source]

Idempotent initialization of chaintool’s files and configurations.

This function must be called at least once before using chaintool for the first time, and after any upgrade to a newer chaintool version. It is automatically invoked with every use of the chaintool command; you would only need to explicitly invoke it if you are calling functions in the chaintool modules from other code.