Command Executor

Classes used for executing shell commands. Main class Executor contains a concrete implementation of various shell commands that are used throughout pubtools-quay. The logic of a command execution is not implemented in this class. The children classes (LocalExecutor, RemoteExecutor) contain the logic for generic command execution based on the environment type (remote or local). New classes may be implemented if a new execution style is required.

Base class

class pubtools._quay.command_executor.Executor

Base executor class.

Implementation of command execution should be done in descendant classes. Common pre- and post-processing should be implemented in this class.

__enter__() → Self

Use the class as context manager. Returns instance upon invocation.

__exit__(exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], exc_tb: Optional[TracebackType]) → None

Cleanup when used as context manager. No-op by default.

skopeo_login(host: str = 'quay.io', username: Optional[str] = None, password: Optional[str] = None) → None

Attempt to login to Quay if no login credentials are present.

Args:

host (str):

docker registry host (quay.io as default)

username (str):

Username for login.

password (str):

Password for login.

tag_images(source_ref: str, dest_refs: List[str], all_arch: bool = False) → None

Copy image from source to destination(s) using skopeo.

Args:

source_ref (str):

Reference of the source image.

dest_refs ([str]):

List of target references to copy the image to.

all_arch (bool):

Whether to copy all architectures (if multiarch image)

skopeo_inspect(image_ref: str, raw: bool = False) → Any

Run skopeo inspect and return the result.

NOTE: inspect command will not be run with the –raw argument. This option only returns an image manifest, which can be gathered by QuayClient. ‘raw’ argument in this function denotes if the result should be parsed or returned raw.

Args:

image_ref (str):

Image reference to inspect.

raw (bool):

Whether to parse the returned JSON, or return raw.

Returns (dict|str):

Result of skopeo inspect.

Children classes

class pubtools._quay.command_executor.LocalExecutor(params: Dict[str, Any] = {})

Run commands locally.

__init__(params: Dict[str, Any] = {}) → None

Initialize.

Args:

params (dict):

Custom parameters to be applied when running the shell commands.

_run_cmd(cmd: str, err_msg: Optional[str] = None, tolerate_err: bool = False, stdin: Optional[str] = None) → Tuple[str, str]

Run a command locally.

Args:

cmd (str):

Shell command to be executed.

error_msg (str):

Error message written when the command fails.

tolerate_err (bool):

Whether to tolerate a failed command.

stdin (str):

String to send to standard input for a command.

Returns (str, str):

Tuple of stdout and stderr generated by the command.

class pubtools._quay.command_executor.RemoteExecutor(hostname: str, username: Optional[str] = None, key_filename: Optional[str] = None, password: Optional[str] = None, port: Optional[int] = None, accept_unknown_host: bool = True)

Run commands remotely via SSH.

__init__(hostname: str, username: Optional[str] = None, key_filename: Optional[str] = None, password: Optional[str] = None, port: Optional[int] = None, accept_unknown_host: bool = True) → None

Initialize.

Args:

hostname (str):

Host to connect to.

username (str):

Username to authenticate as. Defaults to local username.

key_filename (str):

Path to a private key for authentication. Default location will be used if omitted.

password (str):

Password for ssh authentication. Has lower precedence than private key.

port (int):

Optional port of the host.

accept_unknown_host (bool):

Whether to accept an unknown host key. Defaults to True.

_run_cmd(cmd: str, err_msg: Optional[str] = None, tolerate_err: bool = False, stdin: Optional[str] = None) → Tuple[str, str]

Run a command remotely via SSH.

Args:

cmd (str):

Shell command to be executed.

error_msg (str):

Error message written when the command fails.

tolerate_err (bool):

Whether to tolerate a failed command.

stdin (str):

String to send to standard input for a command.

Returns (str, str):

Tuple of stdout and stderr generated by the command.

class pubtools._quay.command_executor.ContainerExecutor(image: str, base_url: str = 'unix://var/run/docker.sock', timeout: Optional[int] = None, verify_tls: bool = False, cert_path: Optional[str] = None, registry_username: Optional[str] = None, registry_password: Optional[str] = None)

Run commands in a Docker container.

__init__(image: str, base_url: str = 'unix://var/run/docker.sock', timeout: Optional[int] = None, verify_tls: bool = False, cert_path: Optional[str] = None, registry_username: Optional[str] = None, registry_password: Optional[str] = None) → None

Initialize.

Args:

image (str):

Path to an image which will be used for performing skopeo operations. Must be downloadable by Docker.

base_url (str):

Base URL of the Docker client.

timeout (int):

Default timeout for API calls, in seconds.

verify_tls (bool):

Whether to use TLS verification.

cert_path (str|None):

Custom path to TLS certificates. If not specified, ‘~/.docker’ is used.

registry_username (str|None):

Username to login to registry containing the specified image. If not provided, login will be assumed to not be needed.

registry_password (str|None):

Password to login to registry containing the specified image. If not provided, login will be assumed to not be needed.

__exit__(exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], exc_tb: Optional[TracebackType]) → None

Cleanup the container when used as a context manager.

_run_cmd(cmd: str, err_msg: Optional[str] = None, tolerate_err: bool = False, stdin: Optional[str] = None) → Tuple[str, str]

Run a command locally.

NOTE: Older versions of Docker API don’t support demuxing of stdout and stderr. This means that data from both streams will be mixed together. To maintain compatibility with the other classes, same output will be returned twice as a tuple. Each string will contain the same mix of stdout and stderr messages.

Args:

cmd (str):

Shell command to be executed.

error_msg (str):

Error message written when the command fails.

tolerate_err (bool):

Whether to tolerate a failed command.

stdin (None|str):

This parameter exists only for compatibility with parent class. Sending input to containers is not supported.

Returns (str, str):

Tuple of stdout and stderr generated by the command.

_add_file(data: str, file_name: str) → None

Add a text file to the running container.

The primary use-case is to store a secret which will be accessed from inside the container. File will be stored in the path /tmp/<file_name>.

Args:

data (str):

Data that should be stored in the container.

file_name (str):

Name of the file.

skopeo_login(host: str = 'quay.io', username: Optional[str] = None, password: Optional[str] = None) → None

Attempt to login to Quay if no login credentials are present.

This method is reimplemented because it uses a different approach to input the password.

Args:

host (str):

docker registry host (quay.io as default)

username (str):

Username for login.

password (str):

Password for login.