API reference

This file describes ssh-utilities API reference.

Warning

Beware, we are still in developement phase so API can change without warning, although most part are final and there are not expected any major changes.

Note

We often use throughout the documentation notation same as python typing. module to mark variable types as it is richer and preserves more information. e.g. List[str] obviously means list of strings. More on the matter can be read in the typing module documentation.

Note

This module should be ideally platform agnostic, but only connections from Windows and Linux(Debian, Ubuntu) to Linux(Debian, Ubuntu) have been tested so any other combinations are officially unsupported but should work.

ssh_utilities.connection

The main module with toplevel Connection class.

The connection class is the main public class that initializes local or remote connection classes as needed based on input arguments.

class ssh_utilities.connection.Connection

Bases: object

Factory for class with self-keeping SSH or local connection.

Main purpose is to have SSH connection with convenience methods which can be easily used. Connection is resiliet to errors and will reinitialize itself if for some reason it fails. It also has a local variant which is mirroring its API but uses os and shutil and subprocess modules internally.

This is a factory class so calling any of the initializer classmethods returns initialized SSHConnection or LocalConnection based on arguments.

Upon import this class automatically reads ssh configuration file in: ~/.ssh/config if it is present. The class is then indexable by keys in config file so calling:

Examples

>>> from ssh_utilities import Connection
>>> Connection[<server_name>]
>>> <ssh_utilities.ssh_utils.SSHConnection at 0x7efedff4fb38>

There is also a specific get method which is safer and with better typing support than dict-like indexing

>>> from ssh_utilities import Connection
>>> Connection(<server_name>)
>>> <ssh_utilities.ssh_utils.SSHConnection at 0x7efedff4fb38>

Class can be also used as a context manager.

>>> from ssh_utilities import Connection
>>> with Connection(<server_name>) as conn:
>>>     conn.something(...)

Connection can also be initialized from appropriately formated string. Strings are used mainly for underlying connection classes persistance to disk

>>> from ssh_utilities import Connection
>>> conn = Connection.from_str(<string>)

returns an initialized connection instance.

All these return connection with preset reasonable parameters if more customization is required, use open method, this also allows use of passwords

>>> from ssh_utilities import Connection
>>> conn = Connection.open(<ssh_username>, <ssh_server>, <ssh_key_file>,
                           <server_name>, <thread_safe>):
classmethod add_hosts(hosts: Union[_HOSTS, List[_HOSTS]])

Add or override availbale host read fron ssh config file.

You can use supplied config parser to parse some externaf ssh config file.

Parameters:hosts (Union[_HOSTS, List[_HOSTS]]) – dictionary or a list of dictionaries containing keys: user, hostname and identityfile

See also

:func:ssh_utilities.config_parser

classmethod from_dict(json: dict, quiet: bool = False) → Union[ssh_utilities.remote.remote.SSHConnection, ssh_utilities.local.local.LocalConnection]

Initializes Connection from str.

String must be formated as defined by abc.ConnectionABC._to_str method.

Parameters:
  • json (dict) – dictionary initialize connection from
  • quiet (bool) – If True suppress login messages
Returns:

initialized local or remmote connection based on parameters parsed from string

Return type:

Union[SSHConnection, LocalConnection]

classmethod from_str(string: str, quiet: bool = False) → Union[ssh_utilities.remote.remote.SSHConnection, ssh_utilities.local.local.LocalConnection]

Initializes Connection from str.

String must be formated as defined by abc.ConnectionABC._to_str method.

Parameters:
  • string (str) – json str to initialize connection from
  • quiet (bool) – If True suppress login messages
Returns:

initialized local or remmote connection based on parameters parsed from string

Return type:

Union[SSHConnection, LocalConnection]

Raises:

KeyError – if required key is missing from string

classmethod get_available_hosts() → List[str]

List all elegible hosts for connection from ~/.ssh/config.

Returns:list of available hosts
Return type:List[str]
static open(ssh_username: str, ssh_server: Optional[str] = '', ssh_key_file: Union[str, Path, None] = None, ssh_password: Optional[str] = None, server_name: Optional[str] = None, quiet: bool = False, thread_safe: bool = False)

Initialize SSH or local connection.

Local connection is only a wrapper around os and shutil module methods and its purpose is to mirror API of the SSHConnection class

Parameters:
  • ssh_username (str) – login name, only used for remote connections
  • ssh_server (str) – server address, numeric address or normal address
  • ssh_key_file (Optional[Union[str, Path]]) – path to file with private rsa key. If left empty and password is None script will ask for password.
  • ssh_password (Optional[str]) – password in string form, this is mainly for testing. Using this in production is a great security risk!
  • server_name (str) – server name (default:None) only for id purposes, if it is left default than it will be replaced with address.
  • quiet (bool) – If True suppress login messages
  • thread_safe (bool) – make connection object thread safe so it can be safely accessed from any number of threads, it is disabled by default to avoid performance penalty of threading locks

Warning

Do not use plain text passwords in production, they are great security risk!

ssh_utilities.utils

Helper function and classes for ssh_utilities module.

ssh_utilities.utils.ProgressBar(total: Optional[float] = None, unit: str = 'b', unit_scale: bool = True, miniters: int = 1, ncols: int = 100, unit_divisor: int = 1024, write_up=2, quiet: bool = True, *args, **kwargs) → Union[ssh_utilities.utils._DummyTqdmWrapper, ssh_utilities.utils._TqdmWrapper]

Progress Bar factory return tqdm subclass or dummy replacement.

Parameters:
  • total (Optional[int]) – total number to transfer, units are set with unit parameter
  • unit (str) – units in which total is input, by default ‘b’
  • unit_scale (bool) – whether toscale units
  • miniters (int) – minimum number of iterations after which update takes place
  • ncols (int) – progressbar length
  • unit_divisor (int) – scale factor for units
  • write_up (int) – number of lines that are written above progressbar
  • quiet (bool) – if True progressbar is not shown
Returns:

which is returned is decided based on value of quiet argument

Return type:

Union[_DummyTqdmWrapper, _TqdmWrapper]

ssh_utilities.utils.bytes_2_human_readable(number_of_bytes: Union[int, float], unit: str = 'b') → str

Convert bytes to human readable format.

Parameters:
  • number_of_bytes (int) – number to convert
  • unit (str) – units of the passed in size, by default “b”
Returns:

filesize in best suitable format

Return type:

str

Raises:

ValueError – if number of bytes is less than 0

class ssh_utilities.utils.CompletedProcess(initial: _CompletedProcess)

Bases: typing.Generic

Completed remote process, mimics subprocess.CompletedProcess API.

Parameters:bytes_out (bool) – is true set stderr and stdout to bytes
check_returncode()

Check if remote process return code was 0, if not raise exception.

Raises:CalledProcessError – if return code was non-zero
class ssh_utilities.utils.lprint(quiet: bool = False)

Bases: object

Callable class that limits print output.

Parameters:
  • text (Any) – content to print
  • quiet (bool) – if true, do not print and return imediatelly
ssh_utilities.utils.for_all_methods(decorator: Callable, exclude: Sequence[str] = [], subclasses: bool = False)

Decorate all methods in class.

Parameters:
  • decorator (Callable) – callable to be used to decorate class methods
  • exclude (List[str]) – list of method names to exclude
  • subclasses – if true decorate also all subclasses methods

Warning

This decorator should be used on class only.

Static and class methods must be excluded or they will not work

Use subclasses=True with great care! it will decorate methods for all instances of class in your module

class ssh_utilities.utils.file_filter(include: Union[str, Sequence[str], None], exclude: Union[str, Sequence[str], None])

Bases: object

Discriminate files to copy by passed in glob patterns.

This is a callable class and works much in a same way as operator.itemgetter

Parameters:
  • include (Optional[str]) – pattern of files to include
  • exclude (Optional[str]) – patttern if files to exclude
ssh_utilities.utils.config_parser(config_path: Union[Path, str]) → paramiko.config.SSHConfig

Parses ssh config file.

Parameters:config_path (Path) – path to config file
Returns:paramiko SSHConfig object that parses config file
Return type:SSHConfig
ssh_utilities.utils.context_timeit(quiet: bool = False)

Context manager which timme the code executed in its scope.

Simple stats about CPU time are printed on context exit.

Parameters:quiet (bool, optional) – If true no statistics are printed on context exit, by default False
class ssh_utilities.utils.NullContext

Bases: object

Replacement for contextlib.nullcontext for python 3.6.

ssh_utilities.exceptions

Module defining exceptions for ssh-utilities.

exception ssh_utilities.exceptions.CalledProcessError(returncode, cmd, output=None, stderr=None)

Bases: subprocess.SubprocessError

Raised when run() is called with check=True and the process returns a non-zero exit status.

cmd, returncode, stdout, stderr, output
stdout

Alias for output attribute, to match stderr

exception ssh_utilities.exceptions.SFTPOpenError

Bases: Exception

Raised when sftp channel could not be opened.

exception ssh_utilities.exceptions.ConnectionError

Bases: Exception

Raised when connection to remote cannot be established.

exception ssh_utilities.exceptions.TimeoutExpired(cmd, timeout, output=None, stderr=None)

Bases: subprocess.SubprocessError

This exception is raised when the timeout expires while waiting for a child process.

cmd, output, stdout, stderr, timeout

ssh_utilities.constants

Module housing ssh-utilities constants.

ssh_utilities.constants.G = '\x1b[32m'

used to higlight important messages in CLI mode

ssh_utilities.constants.LG = '\x1b[92m'

used to highlight important messages, when copying files

ssh_utilities.constants.R = '\x1b[39m'

resets the foreground color to default

ssh_utilities.constants.RED = '\x1b[31m'

used to highlight errors

ssh_utilities.constants.C = '\x1b[96m'

used to highlight important messages, on command execution

ssh_utilities.constants.Y = '\x1b[33m'

used to highlight important messages, on command execution

ssh_utilities.constants.CONFIG_PATH = PosixPath('/home/docs/.ssh/config')

default path to ssh configuration file

ssh_utilities.constants.GET = 'get'

used to specify copy direction sever -> local

ssh_utilities.constants.PUT = 'put'

used to specify copy direction local -> sever

ssh_utilities.typeshed

Module containing typing aliases for ssh-utilities.

ssh_utilities.typeshed._FILE = typing.Union[NoneType, int, typing.IO[typing.Any]]

opened file object or int file descriptor

ssh_utilities.typeshed._CMD = typing.Union[bytes, str, typing.Sequence[ForwardRef('AnyPath')]]

command to exectute for subprocess run, can be str, bytes or sefuence of these

ssh_utilities.typeshed._ENV = typing.Union[typing.Mapping[bytes, typing.Union[bytes, str]], typing.Mapping[str, typing.Union[bytes, str]], NoneType]

mapping of environment varibles names

ssh_utilities.typeshed._GLOBPAT = typing.Union[str, NoneType]

srting glob pattern

ssh_utilities.typeshed._SPATH = typing.Union[str, ForwardRef('Path'), ForwardRef('SSHPath')]

accepted path types by ssh_utilities - str, Path or SSHPath

ssh_utilities.typeshed._DIRECTION = typing_extensions.Literal['get', 'put']

alias for file send direction - put or get

ssh_utilities.typeshed._CALLBACK = typing.Union[typing.Callable[[float, float], typing.Any], NoneType]

copy callback function - callable that accepts two floats first reperesents done part and second total amount

ssh_utilities.typeshed._WALK = typing.Iterator[typing.Tuple[str, typing.List[str], typing.List[str]]]

walk iterator yield typle exactly same as os.walk

ssh_utilities.typeshed._EXCTYPE = typing.Union[typing.Type[Exception], typing.Tuple[typing.Type[Exception], ...]]

alias for exception of tuple of exceptions

ssh_utilities.typeshed._ONERROR = typing.Union[typing.Callable[[Exception], typing.Any], NoneType]

callble that accept one argument which is of exception type