Connection

Connection

Connection(location)

An abstract connection wrapper to execute commands on machines.

Parameters:

Name	Type	Description	Default
location	PurePath	Target directory or file.	required

Source code in b4_backup/main/connection.py

def __init__(self, location: PurePath) -> None:
    """
    Args:
        location: Target directory or file.
    """
    self.location = location
    self.keep_open = False

    self.connected: bool = False

exec_prefix abstractmethod property

exec_prefix

Returns:

Type	Description
str	Prefix to run commands on the target using local commands.

__enter__

__enter__()

Entrypoint in a "with" statement.

Source code in b4_backup/main/connection.py

def __enter__(self) -> Connection:
    """Entrypoint in a "with" statement."""
    return self.open()

__exit__

__exit__(*args, **kwargs)

Endpoint in a "with" statement.

Source code in b4_backup/main/connection.py

def __exit__(self, *args, **kwargs) -> None:
    """Endpoint in a "with" statement."""
    if not self.keep_open:
        self.close()

close abstractmethod

close()

Close the connection.

Source code in b4_backup/main/connection.py

@abstractmethod
def close(self) -> None:
    """Close the connection."""

from_url classmethod

from_url(url)

Parse the URL and return a fitting connection instance.

Parameters:

Name	Type	Description	Default
url	str | None	URL string to parse	required

Returns:

Type	Description
Connection | nullcontext	Connection instance

Source code in b4_backup/main/connection.py

@classmethod
def from_url(cls, url: str | None) -> Connection | contextlib.nullcontext:
    """
    Parse the URL and return a fitting connection instance.

    Args:
        url: URL string to parse

    Returns:
        Connection instance
    """
    if url is None:
        return contextlib.nullcontext()

    parsed_url = URL.from_url(url)

    if parsed_url.protocol is None:
        return LocalConnection(parsed_url.location)

    if parsed_url.protocol == "ssh":
        assert parsed_url.host is not None

        return SSHConnection(
            host=parsed_url.host,
            port=parsed_url.port,
            user=parsed_url.user,
            password=parsed_url.password,
            location=parsed_url.location,
        )

    raise exceptions.UnknownProtocolError

open abstractmethod

open()

Open the connection to the target host.

Returns:

Type	Description
Connection	Itself

Source code in b4_backup/main/connection.py

@abstractmethod
def open(self) -> Connection:
    """
    Open the connection to the target host.

    Returns:
        Itself
    """

run_process abstractmethod

run_process(command)

Run a process without interaction and return the result.

Parameters:

Name	Type	Description	Default
command	list[str]	List of parameters	required

Returns: stdout of process.

Source code in b4_backup/main/connection.py

@abstractmethod
def run_process(self, command: list[str]) -> str:
    """
    Run a process without interaction and return the result.

    Args:
        command: List of parameters
    Returns:
        stdout of process.
    """

LocalConnection

LocalConnection(location)

Bases: Connection

A connection wrapper to execute commands on the local machine.

Parameters:

Name	Type	Description	Default
location	PurePath	Target directory or file.	required

Source code in b4_backup/main/connection.py

def __init__(self, location: PurePath) -> None:
    """
    Args:
        location: Target directory or file.
    """
    super().__init__(location)

    self.location: PurePath = location

exec_prefix property

exec_prefix

Returns:

Type	Description
str	Prefix to run commands on the target using local commands.

close

close()

Close the connection.

Source code in b4_backup/main/connection.py

def close(self) -> None:
    """Close the connection."""
    assert self.connected, "Connection already closed"

    self.connected = False

open

open()

Open the connection to the target host.

Returns:

Type	Description
Connection	Itself

Source code in b4_backup/main/connection.py

def open(self) -> Connection:
    """
    Open the connection to the target host.

    Returns:
        Itself
    """
    log.info("Opening local connection to %s", self.location)
    self.connected = True

    return self

run_process

run_process(command)

Run a process without interaction and return the result.

Parameters:

Name	Type	Description	Default
command	list[str]	List of parameters	required

Returns: stdout of process.

Source code in b4_backup/main/connection.py

def run_process(self, command: list[str]) -> str:
    """
    Run a process without interaction and return the result.

    Args:
        command: List of parameters
    Returns:
        stdout of process.
    """
    log.debug("Start local process:\n%s", command)
    with subprocess.Popen(  # noqa: S603
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        stdin=subprocess.PIPE,
    ) as process:
        stdout, stderr = process.communicate()
        stdout = stdout.decode()
        stderr = stderr.decode()

    if process.returncode:
        raise exceptions.FailedProcessError(command, stdout, stderr)

    return stdout

SSHConnection

SSHConnection(
    host, location, port=22, user="root", password=None
)

Bases: Connection

A connection wrapper to execute commands on remote machines via SSH.

Parameters:

Name	Type	Description	Default
host	str	Hostname	required
location	PurePath	Target directory or file	required
port	int	Port	22
user	str	Username	'root'
password	str | None	Optional password. SSH key recommended.	None

Source code in b4_backup/main/connection.py

def __init__(
    self,
    host: str,
    location: PurePath,
    port: int = 22,
    user: str = "root",
    password: str | None = None,
) -> None:
    """
    Args:
        host: Hostname
        location: Target directory or file
        port: Port
        user: Username
        password: Optional password. SSH key recommended.
    """
    super().__init__(location)

    self.host = host
    self.port = port
    self.user = user
    self.password = password
    self._ssh_client: paramiko.SSHClient | None

exec_prefix property

exec_prefix

Returns:

Type	Description
str	Prefix to run commands on the target using local commands.

close

close()

Close the connection.

Source code in b4_backup/main/connection.py

def close(self) -> None:
    """Close the connection."""
    assert self.connected, "Connection already closed"
    assert self._ssh_client

    log.info("Closing ssh connection to %s %s", self.host, self.location)
    self._ssh_client.close()
    del SSHConnection.ssh_client_pool[(self.host, self.port, self.user)]
    self.connected = False
    self._ssh_client = None

open

open()

Open the connection to the target host.

Returns:

Type	Description
SSHConnection	Itself

Source code in b4_backup/main/connection.py

def open(self) -> SSHConnection:
    """
    Open the connection to the target host.

    Returns:
        Itself
    """
    ssh_client = SSHConnection.ssh_client_pool.get((self.host, self.port, self.user), None)
    if not ssh_client:
        ssh_client = paramiko.SSHClient()
        ssh_client.load_system_host_keys()
        ssh_client.set_missing_host_key_policy(paramiko.RejectPolicy())

        log.info("Opening ssh connection to %s@%s:%s", self.user, self.host, self.port)
        ssh_client.connect(
            self.host,
            username=self.user,
            password=self.password,
            port=self.port,
        )
        SSHConnection.ssh_client_pool[(self.host, self.port, self.user)] = ssh_client

    self.connected = True
    self._ssh_client = ssh_client

    return self

run_process

run_process(command)

Run a process without interaction and return the result.

Parameters:

Name	Type	Description	Default
command	list[str]	List of parameters	required

Returns: stdout of process.

Source code in b4_backup/main/connection.py

def run_process(self, command: list[str]) -> str:
    """
    Run a process without interaction and return the result.

    Args:
        command: List of parameters
    Returns:
        stdout of process.
    """
    assert self._ssh_client, "Not connected"

    log.debug("Start SSH process:\n%s", command)

    _stdin, stdout, stderr = self._ssh_client.exec_command(shlex.join(command))
    stdout_str = stdout.read().decode()
    stderr_str = stderr.read().decode()

    if stdout.channel.recv_exit_status():
        raise exceptions.FailedProcessError(command, stdout_str, stderr_str)

    return stdout_str

URL dataclass

URL(
    protocol=None,
    user="root",
    password=None,
    host=None,
    port=0,
    location=PurePath("/"),
)

Contains an URL.

Parameters:

Name	Type	Description	Default
protocol	str | None	protocol used. eg. ssh	None
user	str	Username	'root'
password	str | None	Password	None
host	str | None	Hostname	None
port	int	Port	0
location	PurePath	Protocol specific location	PurePath('/')

from_url classmethod

from_url(source)

Create an instance by providing an URL string.

Parameters:

Name	Type	Description	Default
source	str	URL string	required

Returns:

Type	Description
URL	ParsedURL instance

Source code in b4_backup/main/connection.py

@classmethod
def from_url(cls, source: str) -> URL:
    """
    Create an instance by providing an URL string.

    Args:
        source: URL string

    Returns:
        ParsedURL instance
    """
    result = cls._url_pattern.match(source)

    if not result:
        result = cls._local_dir_pattern.match(source)

    if not result:
        raise exceptions.InvalidConnectionUrlError(
            f"The connection url {source} got an invalid format."
        )

    result_dict = asdict(URL())
    result_dict.update(result.groupdict())

    if result_dict["protocol"] is not None:
        result_dict["protocol"] = result_dict["protocol"].lower()

    return URL(
        protocol=result_dict["protocol"],
        user=result_dict["user"] or URL.user,
        password=result_dict["password"],
        host=result_dict["host"],
        port=int(result_dict["port"] or cls._protocol_mapping.get(result_dict["protocol"], 0)),
        location=PurePath(result_dict["location"] or "/"),
    )