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
(ssh_server: str, local: bool = False, quiet: bool = False, thread_safe: bool = False)¶ 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.
All methods belong to class so this object should not be instantiated.
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.get(<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
(ssh_server: str, local: bool = False, quiet: bool = False, thread_safe: bool = False)¶ Get Connection based on one of names defined in .ssh/config file.
If name of local PC is passed initilize LocalConnection
Parameters: - ssh_server (str) – server name to connect to defined in ~/.ssh/config file
- local (bool) – if True return emulated connection to loacl host
- 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
Raises: KeyError
– if server name is not in config fileReturns: Instance of SSHConnection for selected server
Return type:
-
classmethod
get_available_hosts
() → List[str]¶ List all elegible hosts for connection from ~/.ssh/config.
Returns: list of available hosts Return type: List[str]
-
classmethod
get_connection
(ssh_server: str, local: bool = False, quiet: bool = False, thread_safe: bool = False)¶ Get Connection based on one of names defined in .ssh/config file.
If name of local PC is passed initilize LocalConnection
Parameters: - ssh_server (str) – server name to connect to defined in ~/.ssh/config file
- local (bool) – if True return emulated connection to loacl host
- 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
Raises: KeyError
– if server name is not in config fileReturns: Instance of SSHConnection for selected server
Return type:
-
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!
-
classmethod
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
-
ssh_utilities.utils.
glob2re
(patern)¶ Translate a shell PATTERN to a regular expression.
There is no way to quote meta-characters.
Parameters: patern (str) – shell glob pattern References
-
class
ssh_utilities.utils.
file_filter
(include: Optional[str], exclude: Optional[str])¶ 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']¶ aliac 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