paks package

Submodules

paks.cli module

paks.cli.get_parser()

paks.cli.run_main()

paks.cli.run module

paks.cli.run.main(args, parser, extra, subparser)

paks.cli.config module

paks.cli.config.main(args, parser, extra, subparser)

paks.client module

class paks.client.PakClient(*args, **kwargs)

Bases: object

Paks has a main controller for interacting with paks.

run(image, registry=None, shell=None, container_tech=None)

Run a paks image

paks.defaults module

paks.schemas module

paks.settings module

class paks.settings.EmptySettings

Bases: object

add(key, value)

Add a value to a list parameter

change_validate(key, value)

A courtesy function to validate a new config addition.

delete(key)

edit()

Interactively edit a config file.

get(key, default=None)

Get a settings value, doing appropriate substitution and expansion.

get_settings_file(settings_file=None)

Get the preferred used settings file.

inituser()

Create a user specific config in user’s home.

load(settings_file=None)

Load the settings file into the settings object

remove(key, value)

Remove a value from a list parameter

save(filename=None)

Save settings, but do not change order of anything.

set(key, value)

Set a setting based on key and value. If the key has :, it’s nested

validate()

Validate the loaded settings with jsonschema

class paks.settings.Settings(settings_file, validate=True)

Bases: paks.settings.EmptySettings

The settings class is a wrapper for easily parsing a settings.yml file.

We parse into a query-able class. It also gives us control to update settings, meaning we change the values and then write them to file. It’s basically a dictionary-like class with extra functions.

paks.version module

paks.logger module

class paks.logger.ColorizingStreamHandler(nocolor=False, stream=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, use_threads=False)

Bases: logging.StreamHandler

BLACK = 0

BLUE = 4

BOLD_SEQ = '\x1b[1m'

COLOR_SEQ = '\x1b[%dm'

CYAN = 6

GREEN = 2

MAGENTA = 5

RED = 1

RESET_SEQ = '\x1b[0m'

WHITE = 7

YELLOW = 3

can_color_tty()

colors = {'CRITICAL': 1, 'DEBUG': 4, 'ERROR': 1, 'INFO': 2, 'WARNING': 3}

decorate(record)

emit(record)

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

property is_tty

class paks.logger.Logger

Bases: object

cleanup()

debug(msg)

error(msg)

exit(msg, return_code=1)

handler(msg)

info(msg)

location(msg)

progress(done=None, total=None)

set_level(level)

set_stream_handler(stream_handler)

shellcmd(msg)

text_handler(msg)

The default snakemake log handler. Prints the output to the console. :param msg: the log message dictionary :type msg: dict

warning(msg)

paks.logger.setup_logger(quiet=False, printshellcmds=False, nocolor=False, stdout=False, debug=False, use_threads=False, wms_monitor=None)

paks.backends module

paks.backends.get_container_backend(name, settings=None)

paks.backends.base module

class paks.backends.base.ContainerName(raw)

Bases: object

Parse a container name into named parts

property extended_name

property name

parse(raw)

Parse a name into known pieces

property slug

class paks.backends.base.ContainerTechnology(settings=None)

Bases: object

A base class for a container technology

clean(string_input)

encode(msg)

get_history(line, openpty)

Given an input with some number of up/down and newline, derive command.

interactive_command(cmd)

Ensure we always restore original TTY otherwise terminal gets messed up

run_executor(string_input, openpty)

Given a string input, run executor

welcome(openpty)

Welcome the user and clear terminal

paks.backends.docker module

class paks.backends.docker.DockerContainer(image, settings=None)

Bases: paks.backends.base.ContainerTechnology

A Docker container controller.

add_registry(uri)

Given a “naked” name, add the registry if it’s Docker Hub

command = 'docker'

run(shell)

Interactive shell into a container image.

paks.backends.podman module

class paks.backends.podman.PodmanContainer(image, settings=None)

Bases: paks.backends.docker.DockerContainer

A Podman container controller, which is the same as Docker.

command = 'podman'

shell(image)

Interactive shell into a container image.

paks.commands module

class paks.commands.DockerCommands(container_tech)

Bases: object

get_executor(name, out=None)

Backend is required to update history

has_command(name)

property history

parse_name(cmd)

required = ['container_name', 'name']

paks.commands.command module

class paks.commands.command.Command(tech, required=None, out=None)

Bases: object

Class method to invoke shell commands and retrieve output and error. This class is inspired and derived from utils functions in https://github.com/vsoch/scif

check(**kwargs)

Check ensures that:

1. The container tech of the command matches the class
2. Required arguments are provided.

decode(line)

Given a line of output (error or regular) decode using the system default, if appropriate

do_print(line, clear=True)

encode(line)

execute(cmd)

Execute a command to the container

execute_get(runcmd, getcmd)

Execute and get runs a command inside the container (pipes to temporary file) and then loads from the outside.

execute_host(cmd)

Execute a command to the host, return out and error

get_args(cmd)

Once we get here, we only care about additional command args.

parse_command(cmd)

this is called when a new command is provided to ensure we have a list. We don’t check that the executable is on the path, as the initialization might not occur in the runtime environment.

parse_kwargs = True

pre_message = None

return_failure(message, out=None, err=None)

Return a failed result (requires a message)

return_success(message=None, out=None, err=None)

Return a successful result

run_command(cmd, output='output')

Wrapper to stream a command, which handles returning a result on error.

run_hidden(cmd)

Run a hidden command.

stream_command(cmd, output='output')

Stream a command and use output or error.

supported_tech = ['docker', 'podman', 'singularity']

class paks.commands.command.Result(out=None, err=None, retval=0, msg=None)

Bases: object

A Result objet holds output and error, and a return value and message

paks.commands.state module

class paks.commands.state.SaveContainer(tech, required=None, out=None)

Bases: paks.commands.command.Command

pre_message = 'Saving container...'

run(**kwargs)

Save a temporary container name back to the main container name Available after check - validated self.kwargs

supported_for = ['docker', 'podman']

paks.utils module

paks.utils.fileio module

paks.utils.fileio.copyfile(source, destination, force=True)

Copy a file from a source to its destination.

paks.utils.fileio.get_file_hash(image_path, algorithm='sha256')

Return an sha256 hash of the file based on a criteria level.

paks.utils.fileio.get_tmpdir(tmpdir=None, prefix='', create=True)

Get a temporary directory for an operation.

paks.utils.fileio.get_tmpfile(tmpdir=None, prefix='')

Get a temporary file with an optional prefix.

paks.utils.fileio.mkdir_p(path)

mkdir_p attempts to get the same functionality as mkdir -p :param path: the path to create.

paks.utils.fileio.mkdirp(dirnames)

Create one or more directories

paks.utils.fileio.print_json(json_obj)

Print json pretty

paks.utils.fileio.read_file(filename, mode='r')

Read a file.

paks.utils.fileio.read_json(filename)

paks.utils.fileio.read_yaml(filename)

paks.utils.fileio.recursive_find(base, pattern=None)

Find filenames that match a particular pattern, and yield them.

paks.utils.fileio.workdir(dirname)

Provide context for a working directory, e.g.,

with workdir(name):

# do stuff

paks.utils.fileio.write_file(filename, content, mode='w')

Write content to a filename

paks.utils.fileio.write_json(content, filename)

paks.utils.names module

class paks.utils.names.RobotNamer

Bases: object

generate(delim='-', length=4, chars='0123456789')

Generate a robot name. Inspiration from Haikunator, but much more

poorly implemented ;)

generate_badge(length=4, chars='0123456789', link=None)

Generate a robot name badge (in markdown).

paks.utils.terminal module

paks.utils.terminal.check_install(software, quiet=True, command='--version')

check_install will attempt to run the singularity command, and return True if installed. The command line utils will not run without this check.

Parameters:

• software (the software to check if installed) –
• quiet (should we be quiet? (default True)) –
• command (the command to use to check (defaults to --version)) –

paks.utils.terminal.confirm_action(question, force=False)

confirm if the user wants to perform a certain action

Parameters:

• question (the question that will be asked) –
• force (if the user wants to skip the prompt) –

paks.utils.terminal.get_installdir()

get_installdir returns the installation directory of the application

paks.utils.terminal.get_username()

paks.utils.terminal.run_command(cmd, stream=False)

run_command uses subprocess to send a command to the terminal.

Parameters:

• cmd (the command to send, should be a list for subprocess) –
• error_message (the error message to give to user if fails,) –
• failed. (if none specified, will alert that command) –

paks.utils.terminal.which(software, strip_newline=True)

get_install will return the path to where Singularity (or another executable) is installed.