Source code for spython.instance

# Copyright (C) 2017-2022 Vanessa Sochat.

# This Source Code Form is subject to the terms of the
# Mozilla Public License, v. 2.0. If a copy of the MPL was not distributed
# with this file, You can obtain one at http://mozilla.org/MPL/2.0/.

from spython.utils.fileio import read_file
from spython.image import ImageBase
import os


[docs]class Instance(ImageBase): def __init__(self, image, start=True, name=None, quiet=True, **kwargs): """An instance is an image running as an instance with services. This class has functions appended under cmd/__init__ and is instantiated when the user calls Client. Parameters ========== image: the Singularity image uri to parse (required) start: boolean to start the instance (default is True) name: a name for the instance (will generate RobotName if not provided) """ super(Instance, self).__init__() self.parse_image_name(image) self.generate_name(name) # Update metadats from arguments self._update_metadata(kwargs) self.options = kwargs.get("options", []) self.args = kwargs.get("args", []) self.cmd = [] self.quiet = quiet # Start the instance if start: self.start(quiet=quiet, **kwargs) # Unique resource identifier
[docs] def generate_name(self, name=None): """generate a Robot Name for the instance to use, if the user doesn't supply one. """ # If no name provided, use robot name if name is None: name = self.RobotNamer.generate() self.name = name.replace("-", "_")
[docs] def parse_image_name(self, image): """ simply split the uri from the image. Singularity handles parsing of registry, namespace, image. Parameters ========== image: the complete image uri to load (e.g., docker://ubuntu) """ self._image = image self.protocol = "instance"
[docs] def get_uri(self): """return the image uri (instance://) along with it's name""" return self.__str__()
# Logs
[docs] def read_logs_out(self): """Read output log file, if it exists""" if hasattr(self, "log_out_path"): if os.path.exists(self.log_out_path): return read_file(self.log_out_path)
[docs] def read_logs_err(self): """Read error log file, if it exists""" if hasattr(self, "log_err_path"): if os.path.exists(self.log_err_path): return read_file(self.log_err_path)
# Metadata def _update_metadata(self, kwargs=None): """Extract any additional attributes to hold with the instance from kwargs """ # If not given metadata, use instance.list to get it for container if kwargs is None and hasattr(self, "name"): kwargs = self._list(self.name, quiet=True, return_json=True) # Add acceptable arguments for arg in ["pid", "name", "ip_address", "log_err_path", "log_out_path", "img"]: # Skip over non-iterables: if arg in kwargs: setattr(self, arg, kwargs[arg]) if "image" in kwargs: self._image = kwargs["image"] elif "container_image" in kwargs: self._image = kwargs["container_image"] def __str__(self): if hasattr(self, "name"): if self.protocol: return "%s://%s" % (self.protocol, self.name) return os.path.basename(self._image) def __repr__(self): return self.__str__()