Source code for miranda.remote.connect
"""Remote Connection Operations module."""
from __future__ import annotations
import logging.config
import warnings
from getpass import getpass
from pathlib import Path
from miranda.scripting import LOGGING_CONFIG
try:
import fabric # noqa
from paramiko import SSHClient # noqa
from scp import SCPClient # noqa
except ImportError:
warnings.warn(
f"{__name__} functions require additional dependencies."
f"Please install them with `pip install miranda[remote]`."
)
logging.config.dictConfig(LOGGING_CONFIG)
__all__ = ["Connection"]
[docs]
class Connection:
"""Connection contextualise class."""
def __init__(
self,
username: str | Path = None,
host: str | Path = None,
protocol: str = "sftp",
*args,
**kwargs,
):
self.user = username or input("Enter username: ")
self.host = host or input("Enter host URL: ")
self._args = list(*args)
self._kwargs = {**kwargs}
self.__c = None
if protocol.lower() in ["sftp", "scp"]:
self.protocol = protocol.lower()
else:
raise ValueError('Protocol must be "sftp" or "scp".')
[docs]
def update(self, **kwargs):
"""Update connection keyword arguments.
Warnings
--------
Credentials are not encrypted.
"""
self._kwargs = kwargs
def __call__(self, **kwargs):
"""Update keyword arguments on call."""
self.update(**kwargs)
return self
def __str__(self):
return f"Connection to {self.host} as {self.user}"
def __repr__(self):
return f"<{self.__class__.__module__}.{self.__class__.__name__} object at {hex(id(self))}>"
[docs]
def connect(self, **kwargs):
"""Connect to a remote server with credential prompts."""
try:
keywords = (
dict(**kwargs)
or dict(**self._kwargs)
or dict(password=getpass("Enter password: "))
)
if self.protocol == "sftp":
c = fabric.Connection(
host=self.host, user=self.user, connect_kwargs=keywords
)
self.__c = c
else:
c = SSHClient()
c.connect(
self.host,
username=self.user,
password=self._kwargs["password"] or getpass("Enter password: "),
)
self.__c = SCPClient(c.get_transport())
return self.__c
except Exception as e:
raise e
def __enter__(self, **kwargs):
return self.connect()
def __exit__(self, exc_type, exc_val, exc_tb):
self.__c.close()