Source code for hostedpi.pi

from ipaddress import IPv6Address, IPv6Network
from time import sleep
from typing import Optional, Union, List, Set, TYPE_CHECKING

if TYPE_CHECKING:
    from .picloud import PiCloud

from requests import Session, HTTPError

from .utils import parse_ssh_keys
from .exc import HostedPiException


NOT_AUTHORISED = "Not authorised to access server or server does not exist"
NOT_PROVISIONED = "Server is not fully provisioned"


[docs]class Pi: """ The ``Pi`` class represents a single Raspberry Pi service in the Mythic Beasts Pi cloud. Initialising a ``Pi`` object does not provision a new Pi, rather initialisation is for internal construction only. There are two ways to get access to a ``Pi`` object: retrieval from the :attr:`~hostedpi.picloud.PiCloud.pis` dictionary; and the return value of :meth:`~hostedpi.picloud.PiCloud.create_pi` method. With a ``Pi`` object, you can access data about that particular Pi service, add SSH keys, reboot it, cancel it and more. .. note:: The ``Pi`` class should not be initialised by the user, only internally within the module. """ def __init__(self, *, cloud: "PiCloud", name: str, model: int): self._cloud = cloud self._name = name self._model = model self._cancelled = False self._boot_progress = None self._disk_size = None self._initialised_keys = None self._ipv6_address = None self._ipv6_network = None self._is_booting = None self._location = None self._model_full = None self._power = None self._ipv4_ssh_port = None self._provision_status = None def __repr__(self): if self._cancelled: return f"<Pi {self.name} cancelled>" else: model = self.model_full if self.model_full else self.model return f"<Pi model {model} {self.name}>"
[docs] def __str__(self): """ A multi-line string of the information about the Pi """ self._get_data() if self._provision_status == "live": if self._is_booting: boot_progress = "booting: {self._boot_progress}" elif self._boot_progress: boot_progress = self._boot_progress else: boot_progress = "live" num_keys = len(self.ssh_keys) power = "on" if self._power else "off" initialised_keys = "yes" if self._initialised_keys else "no" return f""" Name: {self.name} Provision status: {boot_progress} Model: Raspberry Pi {self._model_full} Disk size: {self._disk_size}GB Power: {power} IPv6 address: {self._ipv6_address} IPv6 network: {self._ipv6_network} Initialised keys: {initialised_keys} SSH keys: {num_keys} IPv4 SSH port: {self.ipv4_ssh_port} Location: {self.location} URLs: {self.url} {self.url_ssl} SSH commands: {self.ipv4_ssh_command} # IPv4 {self.ipv6_ssh_command} # IPv6 """[ 1:-1 ] else: return f""" Name: {self.name} Provision status: {self._provision_status} Model: Raspberry Pi {self._model} Disk size: {self._disk_size}GB IPv6 address: {self._ipv6_address} IPv6 network: {self._ipv6_network} SSH port: {self.ipv4_ssh_port} Location: {self.location} URLs: {self.url} {self.url_ssl} SSH commands: {self.ipv4_ssh_command} # IPv4 {self.ipv6_ssh_command} # IPv6 """[ 1:-1 ]
def _get_data(self): # https://www.mythic-beasts.com/support/api/raspberry-pi#ep-get-piserversidentifier url = f"{self._API_URL}/{self.name}" r = self.session.get(url) try: r.raise_for_status() except HTTPError as e: if r.status_code == 403: raise HostedPiException(NOT_AUTHORISED) from e if r.status_code == 409: raise HostedPiException(NOT_PROVISIONED) from e raise HostedPiException(e) from e data = r.json() self._boot_progress = data["boot_progress"] disk_size = data.get("disk_size") if disk_size: self._disk_size = int(float(disk_size)) self._initialised_keys = data["initialised_keys"] self._ipv4_ssh_port = int(data["ssh_port"]) self._ipv6_address = IPv6Address(data["ip"]) self._ipv6_network = IPv6Network(data["ip_routed"]) self._is_booting = bool(data["is_booting"]) self._location = data["location"] self._model = int(data["model"]) self._model_full = data["model_full"] self._power = bool(data["power"]) self._provision_status = data["status"] @property def _API_URL(self) -> str: return self._cloud._API_URL + "/servers" @property def session(self) -> Session: return self._cloud.session @property def name(self) -> str: """ The name of the Pi service """ return self._name @property def boot_progress(self) -> str: """ A string representing the Pi's boot progress. Can be ``booted``, ``powered off`` or a particular stage of the boot process if currently booting. """ self._get_data() if self._boot_progress: return self._boot_progress return "booted" if self.power else "powered off" @property def disk_size(self) -> int: """ The Pi's disk size in GB """ if self._disk_size is None: self._get_data() return self._disk_size @property def initialised_keys(self) -> bool: """ A boolean representing whether or not the Pi has been initialised with SSH keys """ self._get_data() return self._initialised_keys @property def ipv4_ssh_port(self) -> int: """ The SSH port to use when connecting via the IPv4 proxy """ if self._ipv4_ssh_port is None: self._get_data() return self._ipv4_ssh_port @property def ipv6_address(self) -> IPv6Address: """ The Pi's IPv6 address as an :class:`~ipaddress.IPv6Address` object """ if self._ipv6_address is None: self._get_data() return self._ipv6_address @property def ipv6_network(self) -> IPv6Network: """ The Pi's IPv6 network as an :class:`~ipaddress.IPv6Network` object """ if self._ipv6_network is None: self._get_data() return self._ipv6_network @property def is_booting(self) -> bool: """ A boolean representing whether or not the Pi is currently booting """ self._get_data() return self._is_booting @property def location(self) -> str: """ The Pi's physical location (data centre) """ if self._location is None: self._get_data() return self._location @property def model(self) -> int: """ The Pi's model (3 or 4) """ return self._model @property def model_full(self) -> str: """ The Pi's model name (3B, 3B+ or 4B) """ return self._model_full @property def power(self) -> bool: """ A boolean representing whether or not the Pi is currently powered on """ self._get_data() return self._power @property def provision_status(self) -> str: """ A string representing the provision status of the Pi. Can be "provisioning", "initialising" or "live". """ self._get_data() return self._provision_status @property def ipv4_ssh_command(self) -> str: """ The SSH command required to connect to the Pi over IPv4 """ return f"ssh -p {self.ipv4_ssh_port} root@ssh.{self.name}.hostedpi.com" @property def ipv6_ssh_command(self) -> str: """ The SSH command required to connect to the Pi over IPv6 """ return f"ssh root@[{self.ipv6_address}]" @property def ipv4_ssh_config(self) -> str: """ A string containing the IPv4 SSH config for the Pi. The contents could be added to an SSH config file for easy access to the Pi. """ return f"""Host {self.name} user root port {self.ipv4_ssh_port} hostname ssh.{self.name}.hostedpi.com """.strip() @property def ipv6_ssh_config(self) -> str: """ A string containing the IPv6 SSH config for the Pi. The contents could be added to an SSH config file for easy access to the Pi. """ return f"""Host {self.name} user root hostname {self.ipv6_address} """.strip() @property def ssh_keys(self) -> Set[str]: """ Retrieve the SSH keys on the Pi, or use assignment to update them. Property value is a set of strings. Assigned value should also be a set of strings. """ # https://www.mythic-beasts.com/support/api/raspberry-pi#ep-get-piserversidentifierssh-key url = f"{self._API_URL}/{self.name}/ssh-key" r = self.session.get(url) try: r.raise_for_status() except HTTPError as e: if r.status_code == 403: raise HostedPiException(NOT_AUTHORISED) from e raise HostedPiException(e) from e body = r.json() keys = body["ssh_key"] return {key.strip() for key in keys.split("\n") if key.strip()} @ssh_keys.setter def ssh_keys(self, ssh_keys: Union[Set[str], List[str]]): # https://www.mythic-beasts.com/support/api/raspberry-pi#ep-put-piserversidentifierssh-key url = f"{self._API_URL}/{self.name}/ssh-key" if ssh_keys: ssh_keys_str = "\r\n".join(set(ssh_keys)) else: ssh_keys_str = "\r\n" # server doesn't allow empty string data = { "ssh_key": ssh_keys_str, } r = self.session.put(url, json=data) try: r.raise_for_status() except HTTPError as e: if r.status_code == 403: raise HostedPiException(NOT_AUTHORISED) from e raise HostedPiException(e) from e @property def url(self) -> str: """ The http version of the hostedpi.com URL of the Pi. .. note:: Note that a web server must be installed on the Pi for the URL to resolve in a web browser. """ return f"http://www.{self.name}.hostedpi.com" @property def url_ssl(self) -> str: """ The https version of the hostedpi.com URL of the Pi. .. note:: Note that a web server must be installed on the Pi for the URL to resolve in a web browser, and an SSL certificate must be created. See https://letsencrypt.org/ """ return f"https://www.{self.name}.hostedpi.com" def _power_on_off(self, *, on=False): # https://www.mythic-beasts.com/support/api/raspberry-pi#ep-put-piserversidentifierpower url = f"{self._API_URL}/{self.name}/power" data = { "power": on, } r = self.session.put(url, json=data) try: r.raise_for_status() except HTTPError as e: if r.status_code == 400: msg = "The server is already being rebooted" raise HostedPiException(NOT_AUTHORISED) from e if r.status_code == 403: raise HostedPiException(NOT_AUTHORISED) from e raise HostedPiException(e) from e
[docs] def on(self, *, wait: bool = False) -> Optional[bool]: """ Power the Pi on. If *wait* is ``False`` (the default), return immediately. If *wait* is ``True``, wait until the power on request is completed, and return ``True`` on success, and ``False`` on failure. """ self._power_on_off(on=True) if wait: while self.is_booting: sleep(2) return self.power
[docs] def off(self): """ Power the Pi off and return immediately """ self._power_on_off(on=False)
[docs] def reboot(self, *, wait: bool = False): """ Reboot the Pi. If *wait* is ``False`` (the default), return ``None`` immediately. If *wait* is ``True``, wait until the reboot request is completed, and return ``True`` on success, and ``False`` on failure. .. note:: Note that if *wait* is ``False``, you can poll for the boot status while rebooting by inspecting the properties :attr:`~hostedpi.pi.Pi.is_booting` and :attr:`~hostedpi.pi.Pi.boot_progress`. """ # https://www.mythic-beasts.com/support/api/raspberry-pi#ep-post-piserversidentifierreboot url = f"{self._API_URL}/{self.name}/reboot" r = self.session.post(url) try: r.raise_for_status() except HTTPError as e: if r.status_code == 403: raise HostedPiException(NOT_AUTHORISED) from e if r.status_code == 409: # The server is already being rebooted pass raise HostedPiException(e) from e if wait: while self.is_booting: sleep(2) return self.power
[docs] def cancel(self): """ Cancel the Pi service """ # https://www.mythic-beasts.com/support/api/raspberry-pi#ep-delete-piserversidentifier url = f"{self._API_URL}/{self.name}" r = self.session.delete(url) try: r.raise_for_status() except HTTPError as e: if r.status_code == 403: raise HostedPiException(NOT_AUTHORISED) from e raise HostedPiException(e) from e self._cancelled = True
[docs] def ssh_import_id( self, *, github: Optional[Union[Set[str], List[str]]] = None, launchpad: Optional[Union[Set[str], List[str]]] = None, ) -> Set[str]: """ Import SSH keys from GitHub or Launchpad, and add them to the Pi. Return the set of keys added. :type ssh_import_github: list or set or None :param ssh_import_github: A list/set of GitHub usernames to import SSH keys from (keyword-only argument) :type ssh_import_launchpad: list or set or None :param ssh_import_launchpad: A list/set of Launchpad usernames to import SSH keys from (keyword-only argument) """ ssh_keys_set = parse_ssh_keys( ssh_import_github=github, ssh_import_launchpad=launchpad, ) self.ssh_keys |= ssh_keys_set return ssh_keys_set