"""
Includes dependencies of hosts to be used when calling
:func:`ourobori.apps.http.run_service`.
"""
from typing import List
from typing import NoReturn
from typing import Optional
import attr
import asyncssh
from ourobori.services.errors import HostCmdException
from ourobori.services.errors import HostSCPFromException
from ourobori.services.errors import HostSCPToException
from ourobori.services.errors import HostConnectionException
from ourobori.services.errors import prepare_exception_ctx
[docs]@attr.s(auto_attribs=True)
class Host:
"""
This class represents a host used as dependency.
It should be used as a dependency on service-startup in the ``app.py``
passing to the :func:`ourobori.apps.http.run_service`.
Attributes:
name: The name of the host.
host: Host either as ip or as hostname.
user: The user to use for ssh-connection.
known_hosts: The path to the known_hosts-file to use.
pwd: The password to use for the user.
client_keys: The paths of the client-keys to use.
"""
name: str
host: str
user: str
known_hosts: Optional[str] = False
pwd: Optional[str] = False
client_keys: Optional[List[str]] = False
connection_args: dict = attr.ib()
[docs] @connection_args.default
def set_connection_args(self) -> dict:
"""
Sets the attribute connection_args based on the content of the
attributes ``self.known_hosts``, ``self.password`` and
``self.client_keys``.
Returns:
connection_args: The combined connection arguments.
"""
connection_args = dict(username=self.user)
if self.known_hosts:
connection_args['known_hosts'] = self.known_hosts
if self.pwd:
connection_args['password'] = self.pwd
if self.client_keys:
connection_args['client_keys'] = self.client_keys
return connection_args
[docs] async def run_cmd(self, cmd: str) -> str:
"""
Runs the passed command on the host.
Parameters:
cmd: the command to run.
Returns:
The result of the command.
Raises: HostCmdException
"""
async with asyncssh.connect(self.host,
**self.connection_args) as conn:
cmd_result = await conn.run(cmd, check=True)
if cmd_result.exit_status == 0:
return cmd_result.stdout
else:
exc_params = dict(name=self.name,
cmd=cmd,
err=cmd_result.stderr)
exc = prepare_exception_ctx(exception_class=HostCmdException,
exception_params=exc_params)
raise exc
[docs] async def scp_from(self, *, src: str, dst: str) -> NoReturn:
"""
Copies the passed src from the host to the passed dst.
Parameters:
src: The source on the host to copy.
dst: The destination (local) to copy the source to.
Returns:
The result of the command.
Raises: HostSCPFromException
"""
try:
async with asyncssh.connect(self.host,
**self.connection_args) as conn:
await asyncssh.scp(srcpaths=(conn, src),
dstpath=dst,
preserve=True,
recurse=True)
except Exception as err:
exc_params = dict(name=self.name,
err=err)
exc = prepare_exception_ctx(exception_class=HostSCPFromException,
exception_params=exc_params)
raise exc
[docs] async def scp_to(self, *, src: str, dst: str) -> NoReturn:
"""
Copies the passed local src to the passed dst on the host.
Parameters:
src: The source (local) to copy.
dst: The destination on the host to copy the source to.
Returns:
The result of the command.
Raises: HostSCPToException
"""
try:
async with asyncssh.connect(self.host,
**self.connection_args) as conn:
await asyncssh.scp(srcpaths=src,
dstpath=(conn, dst),
preserve=True,
recurse=True)
except Exception as err:
exc_params = dict(name=self.name,
err=err)
exc = prepare_exception_ctx(exception_class=HostSCPToException,
exception_params=exc_params)
raise exc
[docs] async def is_alive(self) -> NoReturn:
"""
Checks if the host can be accessed.
Raises: HostConnectionException
"""
try:
await self.run_cmd(cmd='echo "test"')
except Exception as err:
exc_params = dict(name=self.name,
err=err)
exc = prepare_exception_ctx(exception_class=HostConnectionException,
exception_params=exc_params)
raise exc