Source code for sightline.simon.prototype.prototype

from datetime import datetime
from pathlib import Path
from typing import List

from sightline.simon.resource import V1Resource, V1Error
from .types import PrototypeType
from ..capability import Capability, Grant

_valid_types = PrototypeType.types()

[docs]class Prototype(V1Resource): """ Represents a Prototype that can be used to create Servables in DTaaS. """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if self.type not in _valid_types: raise TypeError("Unknown Prototype type: " + self.type) @property def id(self) -> str: """ :return: A unique id that refers to a Prototype. :rtype: str """ return self._data["id"] @property def name(self) -> str: """ :return: The name of the Prototype. :rtype: str """ return self._data["name"] @property def description(self) -> str: """ :return: A description of the Prototype. :rtype: str """ return self._data["description"] @property def created(self) -> datetime: """ :return: Date and time of Prototype creation. :rtype: datetime """ return datetime.strptime(self._data["created"], "%Y-%m-%dT%H:%M:%S.%f") @property def tags(self) -> List[str]: """ :return: Tags associated with the Prototype. :rtype: List[str] """ return self._data["tags"] @property def type(self) -> str: """ The Prototype type. Can be one of: - APP - DAEMON - DATASET - MODEL - PIPELINE - REPORT - VIEW :return: The Prototype type. :rtype: str """ return self._data["type"] @property def dependencies(self) -> List["Prototype"]: """ Returns a list of :class:`Prototype` objects this prototype is dependent on. :return: The dependencies of this Prototype. :rtype: List[Prototype] """ deps = self._data.relationships["dependencies"]["data"] return [Prototype.from_id(prototype["id"]) for prototype in deps] @property def capabilities(self) -> List[Capability]: """ Returns the list of all capabilities available on this Prototype. :return: Capabilities available on the Prototype :rtype: List[:class:`Capability`] """ caps = self._data.capabilities return [getattr(Capability, c) for c in caps] @property def grants(self) -> List[Grant]: """ The list of Grants the caller has for this Prototype. The Grants are composed of the type of object through which we have been granted the capabilities, the ID of said object and the list of capabilities it has provided. Multiple grants may provide the same capabilities and the total capabilities on the Prototype may include things which are not provided by a Grant. :return: The Grants for this Prototype :rtype: List[:class:`Grant`] """ grants = self._data.grants return [Grant.from_relationship(grant) for grant in grants] @classmethod def resource_group(cls) -> str: """ Retrieves the ID representing the group of all prototypes. This ID can be used when requesting delegated capabilities to get capabilities on all prototypes without the need to request each of them individually or make new requests as new prototypes become available. :return: The ID of representing all prototypes :rtype: str """ response = cls._get("/v1/resource-groups/prototype") return response.json()["data"]
[docs] @classmethod def from_id(cls, id: str) -> "Prototype": """ Get an instance of a Prototype from its id. Using the form "name:tag" is also valid. :param id: The id of the Prototype. :type id: str :return: A Prototype. :rtype: :class:`Prototype` :raises: V1Error: This occurs if no Prototype ID is passed in """ if id is None: raise V1Error("Missing ID") response = cls._get("/v1/prototypes/" + id) response.raise_for_status() return cls(response.json())
[docs] @classmethod def list(cls) -> List["Prototype"]: """ Lists all prototypes. :return: A list of Prototypes that are available. :rtype: List[Prototype] """ return [ cls(obj) for obj in cls._paginate("/v1/prototypes?page=0&limit=100") ]
[docs] @classmethod def remove(cls, *id: str): """ Remove a number of Prototypes. If any individual prototype is unable to be removed, none of the provided prototypes will be removed. Prototypes can also be in the form of name:tag, where name and tag come from the project MANIFEST for the prototype. :args id: The id(s) of the Prototype to remove. :return: None :raises: V1Error: This occurs if no Prototype ID is passed in to remove, or if any of the Prototype IDs cannot be deleted """ if not id: raise V1Error("Missing IDs") response = cls._post("/v1/rpc/prototypes/delete", json=list(id)) response.raise_for_status()
@classmethod def cleanup_prototypes(cls, name_tag: str) -> int: """ Attempt to remove all Prototypes associated with name:tag except the latest one. If a given Prototype cannot be removed, no action will be taken on that prototype. All others however will be removed. :args name_tag: The name:tag combination of the Prototype to clean up. :return: Number of prototypes cleaned up :raises: V1Error: This occurs if no Prototype ID is passed in to remove """ response = cls._post(f"/v1/rpc/prototypes/cleanup", json=name_tag) response.raise_for_status() return response.content.decode() @classmethod def upload_project(cls, project_file: str) -> List["Prototype"]: """ Uploads a project into DTaaS. Projects define Prototypes. This should not be used directly, instead use the SDK command line interface and upload from there. :param project_file: Path to project to upload. :type project_file: str :return: The created Prototypes. :rtype: List[Prototype] """ project_path = Path(project_file) if not project_path.exists(): raise FileNotFoundError elif project_path.is_dir(): raise IsADirectoryError with open(project_file, "rb") as f: response = cls._post("/v1/prototypes/project", files={"project": f}) response.raise_for_status() objs = response.json()["data"] return [cls(obj) for obj in objs]