API

api

v1

clusters

Client

Source code in denvr/api/v1/clusters.py

class Client:
    def __init__(self, session: Session):
        self.session = session

    def get_all(self) -> list:
        """
        Get a list of allocated clusters ::

            client.get_all()


        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {}

        kwargs = validate_kwargs("get", "/api/v1/clusters/GetAll", parameters, {})

        return self.session.request("get", "/api/v1/clusters/GetAll", **kwargs)

get_all

Get a list of allocated clusters ::

client.get_all()

Source code in denvr/api/v1/clusters.py

def get_all(self) -> list:
    """
    Get a list of allocated clusters ::

        client.get_all()


    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {}

    kwargs = validate_kwargs("get", "/api/v1/clusters/GetAll", parameters, {})

    return self.session.request("get", "/api/v1/clusters/GetAll", **kwargs)

servers

applications

Client

Source code in denvr/api/v1/servers/applications.py

class Client:
    def __init__(self, session: Session):
        self.session = session

    def get_applications(self) -> dict:
        """
        Get a list of applications ::

            client.get_applications()


        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {}

        kwargs = validate_kwargs(
            "get", "/api/v1/servers/applications/GetApplications", parameters, {}
        )

        return self.session.request(
            "get", "/api/v1/servers/applications/GetApplications", **kwargs
        )

    def get_application_details(
        self, id: str | None = None, cluster: str | None = None
    ) -> dict:
        """
        Get detailed information about a specific application ::

            client.get_application_details(id="my-jupyter-application", cluster="Msc1")

        Keyword Arguments:
            id (str): The application name
            cluster (str): The cluster you're operating on

        Returns:
            instance_details (dict):
            application_catalog_item (dict):
            hardware_package (dict):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {
                "Id": config.getkwarg("id", id),
                "Cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "get",
            "/api/v1/servers/applications/GetApplicationDetails",
            parameters,
            {"Id", "Cluster"},
        )

        return self.session.request(
            "get", "/api/v1/servers/applications/GetApplicationDetails", **kwargs
        )

    def get_configurations(self) -> dict:
        """
        Get a list of application configurations ::

            client.get_configurations()


        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {}

        kwargs = validate_kwargs(
            "get", "/api/v1/servers/applications/GetConfigurations", parameters, {}
        )

        return self.session.request(
            "get", "/api/v1/servers/applications/GetConfigurations", **kwargs
        )

    def get_availability(
        self, cluster: str | None = None, resource_pool: str | None = None
    ) -> dict:
        """
        Get detailed information on available configurations for applications ::

            client.get_availability(cluster="Msc1", resource_pool="on-demand")

        Keyword Arguments:
            cluster (str):
            resource_pool (str):

        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {
                "cluster": config.getkwarg("cluster", cluster),
                "resourcePool": config.getkwarg("resource_pool", resource_pool),
            }
        }

        kwargs = validate_kwargs(
            "get",
            "/api/v1/servers/applications/GetAvailability",
            parameters,
            {"cluster", "resourcePool"},
        )

        return self.session.request(
            "get", "/api/v1/servers/applications/GetAvailability", **kwargs
        )

    def get_application_catalog_items(self) -> dict:
        """
        Get a list of application catalog items ::

            client.get_application_catalog_items()


        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {}

        kwargs = validate_kwargs(
            "get", "/api/v1/servers/applications/GetApplicationCatalogItems", parameters, {}
        )

        return self.session.request(
            "get", "/api/v1/servers/applications/GetApplicationCatalogItems", **kwargs
        )

    def create_catalog_application(
        self,
        name: str | None = None,
        cluster: str | None = None,
        hardware_package_name: str | None = None,
        application_catalog_item_name: str | None = None,
        application_catalog_item_version: str | None = None,
        resource_pool: str | None = None,
        ssh_keys: list | None = None,
        persist_direct_attached_storage: bool | None = None,
        personal_shared_storage: bool | None = None,
        tenant_shared_storage: bool | None = None,
        jupyter_token: str | None = None,
    ) -> dict:
        """
        Create a new application using a pre-defined configuration and application catalog item ::

            client.create_catalog_application(
                name="my-jupyter-application",
                cluster="Msc1",
                hardware_package_name="g-nvidia-1xa100-40gb-pcie-14vcpu-112gb",
                application_catalog_item_name="jupyter-notebook",
                application_catalog_item_version="python-3.11.9",
                resource_pool="on-demand",
                ssh_keys=["string"],
                persist_direct_attached_storage=False,
                personal_shared_storage=True,
                tenant_shared_storage=True,
                jupyter_token="abc123",
            )

        Keyword Arguments:
            name (str): The application name
            cluster (str): The cluster you're operating on
            hardware_package_name (str): The name or unique identifier of the application hardware configuration to use for the application.
            application_catalog_item_name (str): The name of the application catalog item.
            application_catalog_item_version (str): The version name of the application catalog item.
            resource_pool (str): The resource pool to use for the application
            ssh_keys (list): The SSH keys for accessing the application
            persist_direct_attached_storage (bool): Indicates whether to persist direct attached storage (if resource pool is reserved)
            personal_shared_storage (bool): Enable personal shared storage for the application
            tenant_shared_storage (bool): Enable tenant shared storage for the application
            jupyter_token (str): An authentication token for accessing Jupyter Notebook enabled applications

        Returns:
            id (str):
            cluster (str):
            status (str):
            tenant (str):
            created_by (str):
            private_ip (str):
            public_ip (str):
            resource_pool (str):
            dns (str):
            ssh_username (str):
            application_catalog_item_name (str):
            application_catalog_item_version_name (str):
            hardware_package_name (str):
            persisted_direct_attached_storage (bool):
            personal_shared_storage (bool):
            tenant_shared_storage (bool):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "name": config.getkwarg("name", name),
                "cluster": config.getkwarg("cluster", cluster),
                "hardwarePackageName": config.getkwarg(
                    "hardware_package_name", hardware_package_name
                ),
                "applicationCatalogItemName": config.getkwarg(
                    "application_catalog_item_name", application_catalog_item_name
                ),
                "applicationCatalogItemVersion": config.getkwarg(
                    "application_catalog_item_version", application_catalog_item_version
                ),
                "resourcePool": config.getkwarg("resource_pool", resource_pool),
                "sshKeys": config.getkwarg("ssh_keys", ssh_keys),
                "persistDirectAttachedStorage": config.getkwarg(
                    "persist_direct_attached_storage", persist_direct_attached_storage
                ),
                "personalSharedStorage": config.getkwarg(
                    "personal_shared_storage", personal_shared_storage
                ),
                "tenantSharedStorage": config.getkwarg(
                    "tenant_shared_storage", tenant_shared_storage
                ),
                "jupyterToken": config.getkwarg("jupyter_token", jupyter_token),
            }
        }

        kwargs = validate_kwargs(
            "post",
            "/api/v1/servers/applications/CreateCatalogApplication",
            parameters,
            {
                "applicationCatalogItemName",
                "applicationCatalogItemVersion",
                "cluster",
                "hardwarePackageName",
                "name",
            },
        )

        return self.session.request(
            "post", "/api/v1/servers/applications/CreateCatalogApplication", **kwargs
        )

    def create_custom_application(
        self,
        name: str | None = None,
        cluster: str | None = None,
        hardware_package_name: str | None = None,
        image_url: str | None = None,
        image_cmd_override: list | None = None,
        environment_variables: dict | None = None,
        image_repository: dict | None = None,
        resource_pool: str | None = None,
        readiness_watcher_port: int | None = None,
        proxy_port: int | None = None,
        persist_direct_attached_storage: bool | None = None,
        personal_shared_storage: bool | None = None,
        tenant_shared_storage: bool | None = None,
        user_scripts: dict | None = None,
        security_context: dict | None = None,
    ) -> dict:
        """
        Create a new custom application using a pre-defined configuration and user-defined container image. ::

            client.create_custom_application(
                name="my-custom-application",
                cluster="Msc1",
                hardware_package_name="g-nvidia-1xa100-40gb-pcie-14vcpu-112gb",
                image_url="docker.io/{namespace}/{repository}:{tag}",
                image_cmd_override=["python", "train.py"],
                environment_variables={},
                image_repository={
                    "hostname": "https://index.docker.io/v1/",
                    "username": "your-docker-username",
                    "password": "dckr_pat__xxx1234567890abcdef",
                },
                resource_pool="on-demand",
                readiness_watcher_port=443,
                proxy_port=8888,
                persist_direct_attached_storage=False,
                personal_shared_storage=True,
                tenant_shared_storage=True,
                user_scripts={},
                security_context={"runAsRoot": False},
            )

        Keyword Arguments:
            name (str): The application name
            cluster (str): The cluster you're operating on
            hardware_package_name (str): The name or unique identifier of the application hardware configuration to use for the application.
            image_url (str): Image URL for the custom application.
            image_cmd_override (list): Optional Image CMD override allows users to specify a custom command to run in the container....
            environment_variables (dict): Environment variables for the application. Names must start with a letter or underscore and...
            image_repository (dict):
            resource_pool (str): The resource pool to use for the application
            readiness_watcher_port (int): The port used for monitoring application readiness and status. Common examples:  - 443...
            proxy_port (int): The port your application uses to receive HTTPS traffic.   Port 443 is reserved for the reverse...
            persist_direct_attached_storage (bool): Indicates whether to persist direct attached storage (if resource pool is reserved)
            personal_shared_storage (bool): Enable personal shared storage for the application
            tenant_shared_storage (bool): Enable tenant shared storage for the application
            user_scripts (dict): Dictionary of script filenames to script content. Each scripts to be mounted at...
            security_context (dict):

        Returns:
            id (str):
            cluster (str):
            status (str):
            tenant (str):
            created_by (str):
            private_ip (str):
            public_ip (str):
            resource_pool (str):
            dns (str):
            ssh_username (str):
            application_catalog_item_name (str):
            application_catalog_item_version_name (str):
            hardware_package_name (str):
            persisted_direct_attached_storage (bool):
            personal_shared_storage (bool):
            tenant_shared_storage (bool):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "name": config.getkwarg("name", name),
                "cluster": config.getkwarg("cluster", cluster),
                "hardwarePackageName": config.getkwarg(
                    "hardware_package_name", hardware_package_name
                ),
                "imageUrl": config.getkwarg("image_url", image_url),
                "imageCmdOverride": config.getkwarg("image_cmd_override", image_cmd_override),
                "environmentVariables": config.getkwarg(
                    "environment_variables", environment_variables
                ),
                "imageRepository": config.getkwarg("image_repository", image_repository),
                "resourcePool": config.getkwarg("resource_pool", resource_pool),
                "readinessWatcherPort": config.getkwarg(
                    "readiness_watcher_port", readiness_watcher_port
                ),
                "proxyPort": config.getkwarg("proxy_port", proxy_port),
                "persistDirectAttachedStorage": config.getkwarg(
                    "persist_direct_attached_storage", persist_direct_attached_storage
                ),
                "personalSharedStorage": config.getkwarg(
                    "personal_shared_storage", personal_shared_storage
                ),
                "tenantSharedStorage": config.getkwarg(
                    "tenant_shared_storage", tenant_shared_storage
                ),
                "userScripts": config.getkwarg("user_scripts", user_scripts),
                "securityContext": config.getkwarg("security_context", security_context),
            }
        }

        kwargs = validate_kwargs(
            "post",
            "/api/v1/servers/applications/CreateCustomApplication",
            parameters,
            {"cluster", "hardwarePackageName", "imageRepository", "imageUrl", "name"},
        )

        return self.session.request(
            "post", "/api/v1/servers/applications/CreateCustomApplication", **kwargs
        )

    def start_application(self, id: str | None = None, cluster: str | None = None) -> dict:
        """
        Start an application that has been previously set up and provisioned, but is currently OFFLINE ::

            client.start_application(id="my-jupyter-application", cluster="Msc1")

        Keyword Arguments:
            id (str): The application name
            cluster (str): The cluster you're operating on

        Returns:
            id (str): The application name
            cluster (str): The cluster you're operating on
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "id": config.getkwarg("id", id),
                "cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "post",
            "/api/v1/servers/applications/StartApplication",
            parameters,
            {"cluster", "id"},
        )

        return self.session.request(
            "post", "/api/v1/servers/applications/StartApplication", **kwargs
        )

    def stop_application(self, id: str | None = None, cluster: str | None = None) -> dict:
        """
        Stop an application that has been previously set up and provisioned, but is currently ONLINE ::

            client.stop_application(id="my-jupyter-application", cluster="Msc1")

        Keyword Arguments:
            id (str): The application name
            cluster (str): The cluster you're operating on

        Returns:
            id (str): The application name
            cluster (str): The cluster you're operating on
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "id": config.getkwarg("id", id),
                "cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "post",
            "/api/v1/servers/applications/StopApplication",
            parameters,
            {"cluster", "id"},
        )

        return self.session.request(
            "post", "/api/v1/servers/applications/StopApplication", **kwargs
        )

    def destroy_application(self, id: str | None = None, cluster: str | None = None) -> dict:
        """
        Permanently delete a specified application, effectively wiping all its data and freeing up resources for other uses ::

            client.destroy_application(id="my-jupyter-application", cluster="Msc1")

        Keyword Arguments:
            id (str): The application name
            cluster (str): The cluster you're operating on

        Returns:
            id (str): The application name
            cluster (str): The cluster you're operating on
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {
                "Id": config.getkwarg("id", id),
                "Cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "delete",
            "/api/v1/servers/applications/DestroyApplication",
            parameters,
            {"Id", "Cluster"},
        )

        return self.session.request(
            "delete", "/api/v1/servers/applications/DestroyApplication", **kwargs
        )

create_catalog_application

Create a new application using a pre-defined configuration and application catalog item ::

client.create_catalog_application(
    name="my-jupyter-application",
    cluster="Msc1",
    hardware_package_name="g-nvidia-1xa100-40gb-pcie-14vcpu-112gb",
    application_catalog_item_name="jupyter-notebook",
    application_catalog_item_version="python-3.11.9",
    resource_pool="on-demand",
    ssh_keys=["string"],
    persist_direct_attached_storage=False,
    personal_shared_storage=True,
    tenant_shared_storage=True,
    jupyter_token="abc123",
)

Other Parameters:

Name	Type	Description
name	str	The application name
cluster	str	The cluster you're operating on
hardware_package_name	str	The name or unique identifier of the application hardware configuration to use for the application.
application_catalog_item_name	str	The name of the application catalog item.
application_catalog_item_version	str	The version name of the application catalog item.
resource_pool	str	The resource pool to use for the application
ssh_keys	list	The SSH keys for accessing the application
persist_direct_attached_storage	bool	Indicates whether to persist direct attached storage (if resource pool is reserved)
personal_shared_storage	bool	Enable personal shared storage for the application
tenant_shared_storage	bool	Enable tenant shared storage for the application
jupyter_token	str	An authentication token for accessing Jupyter Notebook enabled applications

Returns:

Name	Type	Description
id	str
cluster	str
status	str
tenant	str
created_by	str
private_ip	str
public_ip	str
resource_pool	str
dns	str
ssh_username	str
application_catalog_item_name	str
application_catalog_item_version_name	str
hardware_package_name	str
persisted_direct_attached_storage	bool
personal_shared_storage	bool
tenant_shared_storage	bool

Source code in denvr/api/v1/servers/applications.py

def create_catalog_application(
    self,
    name: str | None = None,
    cluster: str | None = None,
    hardware_package_name: str | None = None,
    application_catalog_item_name: str | None = None,
    application_catalog_item_version: str | None = None,
    resource_pool: str | None = None,
    ssh_keys: list | None = None,
    persist_direct_attached_storage: bool | None = None,
    personal_shared_storage: bool | None = None,
    tenant_shared_storage: bool | None = None,
    jupyter_token: str | None = None,
) -> dict:
    """
    Create a new application using a pre-defined configuration and application catalog item ::

        client.create_catalog_application(
            name="my-jupyter-application",
            cluster="Msc1",
            hardware_package_name="g-nvidia-1xa100-40gb-pcie-14vcpu-112gb",
            application_catalog_item_name="jupyter-notebook",
            application_catalog_item_version="python-3.11.9",
            resource_pool="on-demand",
            ssh_keys=["string"],
            persist_direct_attached_storage=False,
            personal_shared_storage=True,
            tenant_shared_storage=True,
            jupyter_token="abc123",
        )

    Keyword Arguments:
        name (str): The application name
        cluster (str): The cluster you're operating on
        hardware_package_name (str): The name or unique identifier of the application hardware configuration to use for the application.
        application_catalog_item_name (str): The name of the application catalog item.
        application_catalog_item_version (str): The version name of the application catalog item.
        resource_pool (str): The resource pool to use for the application
        ssh_keys (list): The SSH keys for accessing the application
        persist_direct_attached_storage (bool): Indicates whether to persist direct attached storage (if resource pool is reserved)
        personal_shared_storage (bool): Enable personal shared storage for the application
        tenant_shared_storage (bool): Enable tenant shared storage for the application
        jupyter_token (str): An authentication token for accessing Jupyter Notebook enabled applications

    Returns:
        id (str):
        cluster (str):
        status (str):
        tenant (str):
        created_by (str):
        private_ip (str):
        public_ip (str):
        resource_pool (str):
        dns (str):
        ssh_username (str):
        application_catalog_item_name (str):
        application_catalog_item_version_name (str):
        hardware_package_name (str):
        persisted_direct_attached_storage (bool):
        personal_shared_storage (bool):
        tenant_shared_storage (bool):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "name": config.getkwarg("name", name),
            "cluster": config.getkwarg("cluster", cluster),
            "hardwarePackageName": config.getkwarg(
                "hardware_package_name", hardware_package_name
            ),
            "applicationCatalogItemName": config.getkwarg(
                "application_catalog_item_name", application_catalog_item_name
            ),
            "applicationCatalogItemVersion": config.getkwarg(
                "application_catalog_item_version", application_catalog_item_version
            ),
            "resourcePool": config.getkwarg("resource_pool", resource_pool),
            "sshKeys": config.getkwarg("ssh_keys", ssh_keys),
            "persistDirectAttachedStorage": config.getkwarg(
                "persist_direct_attached_storage", persist_direct_attached_storage
            ),
            "personalSharedStorage": config.getkwarg(
                "personal_shared_storage", personal_shared_storage
            ),
            "tenantSharedStorage": config.getkwarg(
                "tenant_shared_storage", tenant_shared_storage
            ),
            "jupyterToken": config.getkwarg("jupyter_token", jupyter_token),
        }
    }

    kwargs = validate_kwargs(
        "post",
        "/api/v1/servers/applications/CreateCatalogApplication",
        parameters,
        {
            "applicationCatalogItemName",
            "applicationCatalogItemVersion",
            "cluster",
            "hardwarePackageName",
            "name",
        },
    )

    return self.session.request(
        "post", "/api/v1/servers/applications/CreateCatalogApplication", **kwargs
    )

create_custom_application

Create a new custom application using a pre-defined configuration and user-defined container image. ::

client.create_custom_application(
    name="my-custom-application",
    cluster="Msc1",
    hardware_package_name="g-nvidia-1xa100-40gb-pcie-14vcpu-112gb",
    image_url="docker.io/{namespace}/{repository}:{tag}",
    image_cmd_override=["python", "train.py"],
    environment_variables={},
    image_repository={
        "hostname": "https://index.docker.io/v1/",
        "username": "your-docker-username",
        "password": "dckr_pat__xxx1234567890abcdef",
    },
    resource_pool="on-demand",
    readiness_watcher_port=443,
    proxy_port=8888,
    persist_direct_attached_storage=False,
    personal_shared_storage=True,
    tenant_shared_storage=True,
    user_scripts={},
    security_context={"runAsRoot": False},
)

Other Parameters:

Name	Type	Description
name	str	The application name
cluster	str	The cluster you're operating on
hardware_package_name	str	The name or unique identifier of the application hardware configuration to use for the application.
image_url	str	Image URL for the custom application.
image_cmd_override	list	Optional Image CMD override allows users to specify a custom command to run in the container....
environment_variables	dict	Environment variables for the application. Names must start with a letter or underscore and...
image_repository	dict
resource_pool	str	The resource pool to use for the application
readiness_watcher_port	int	The port used for monitoring application readiness and status. Common examples: - 443...
proxy_port	int	The port your application uses to receive HTTPS traffic. Port 443 is reserved for the reverse...
persist_direct_attached_storage	bool	Indicates whether to persist direct attached storage (if resource pool is reserved)
personal_shared_storage	bool	Enable personal shared storage for the application
tenant_shared_storage	bool	Enable tenant shared storage for the application
user_scripts	dict	Dictionary of script filenames to script content. Each scripts to be mounted at...
security_context	dict

Returns:

Name	Type	Description
id	str
cluster	str
status	str
tenant	str
created_by	str
private_ip	str
public_ip	str
resource_pool	str
dns	str
ssh_username	str
application_catalog_item_name	str
application_catalog_item_version_name	str
hardware_package_name	str
persisted_direct_attached_storage	bool
personal_shared_storage	bool
tenant_shared_storage	bool

Source code in denvr/api/v1/servers/applications.py

def create_custom_application(
    self,
    name: str | None = None,
    cluster: str | None = None,
    hardware_package_name: str | None = None,
    image_url: str | None = None,
    image_cmd_override: list | None = None,
    environment_variables: dict | None = None,
    image_repository: dict | None = None,
    resource_pool: str | None = None,
    readiness_watcher_port: int | None = None,
    proxy_port: int | None = None,
    persist_direct_attached_storage: bool | None = None,
    personal_shared_storage: bool | None = None,
    tenant_shared_storage: bool | None = None,
    user_scripts: dict | None = None,
    security_context: dict | None = None,
) -> dict:
    """
    Create a new custom application using a pre-defined configuration and user-defined container image. ::

        client.create_custom_application(
            name="my-custom-application",
            cluster="Msc1",
            hardware_package_name="g-nvidia-1xa100-40gb-pcie-14vcpu-112gb",
            image_url="docker.io/{namespace}/{repository}:{tag}",
            image_cmd_override=["python", "train.py"],
            environment_variables={},
            image_repository={
                "hostname": "https://index.docker.io/v1/",
                "username": "your-docker-username",
                "password": "dckr_pat__xxx1234567890abcdef",
            },
            resource_pool="on-demand",
            readiness_watcher_port=443,
            proxy_port=8888,
            persist_direct_attached_storage=False,
            personal_shared_storage=True,
            tenant_shared_storage=True,
            user_scripts={},
            security_context={"runAsRoot": False},
        )

    Keyword Arguments:
        name (str): The application name
        cluster (str): The cluster you're operating on
        hardware_package_name (str): The name or unique identifier of the application hardware configuration to use for the application.
        image_url (str): Image URL for the custom application.
        image_cmd_override (list): Optional Image CMD override allows users to specify a custom command to run in the container....
        environment_variables (dict): Environment variables for the application. Names must start with a letter or underscore and...
        image_repository (dict):
        resource_pool (str): The resource pool to use for the application
        readiness_watcher_port (int): The port used for monitoring application readiness and status. Common examples:  - 443...
        proxy_port (int): The port your application uses to receive HTTPS traffic.   Port 443 is reserved for the reverse...
        persist_direct_attached_storage (bool): Indicates whether to persist direct attached storage (if resource pool is reserved)
        personal_shared_storage (bool): Enable personal shared storage for the application
        tenant_shared_storage (bool): Enable tenant shared storage for the application
        user_scripts (dict): Dictionary of script filenames to script content. Each scripts to be mounted at...
        security_context (dict):

    Returns:
        id (str):
        cluster (str):
        status (str):
        tenant (str):
        created_by (str):
        private_ip (str):
        public_ip (str):
        resource_pool (str):
        dns (str):
        ssh_username (str):
        application_catalog_item_name (str):
        application_catalog_item_version_name (str):
        hardware_package_name (str):
        persisted_direct_attached_storage (bool):
        personal_shared_storage (bool):
        tenant_shared_storage (bool):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "name": config.getkwarg("name", name),
            "cluster": config.getkwarg("cluster", cluster),
            "hardwarePackageName": config.getkwarg(
                "hardware_package_name", hardware_package_name
            ),
            "imageUrl": config.getkwarg("image_url", image_url),
            "imageCmdOverride": config.getkwarg("image_cmd_override", image_cmd_override),
            "environmentVariables": config.getkwarg(
                "environment_variables", environment_variables
            ),
            "imageRepository": config.getkwarg("image_repository", image_repository),
            "resourcePool": config.getkwarg("resource_pool", resource_pool),
            "readinessWatcherPort": config.getkwarg(
                "readiness_watcher_port", readiness_watcher_port
            ),
            "proxyPort": config.getkwarg("proxy_port", proxy_port),
            "persistDirectAttachedStorage": config.getkwarg(
                "persist_direct_attached_storage", persist_direct_attached_storage
            ),
            "personalSharedStorage": config.getkwarg(
                "personal_shared_storage", personal_shared_storage
            ),
            "tenantSharedStorage": config.getkwarg(
                "tenant_shared_storage", tenant_shared_storage
            ),
            "userScripts": config.getkwarg("user_scripts", user_scripts),
            "securityContext": config.getkwarg("security_context", security_context),
        }
    }

    kwargs = validate_kwargs(
        "post",
        "/api/v1/servers/applications/CreateCustomApplication",
        parameters,
        {"cluster", "hardwarePackageName", "imageRepository", "imageUrl", "name"},
    )

    return self.session.request(
        "post", "/api/v1/servers/applications/CreateCustomApplication", **kwargs
    )

destroy_application

Permanently delete a specified application, effectively wiping all its data and freeing up resources for other uses ::

client.destroy_application(id="my-jupyter-application", cluster="Msc1")

Other Parameters:

Name	Type	Description
id	str	The application name
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str	The application name
cluster	str	The cluster you're operating on

Source code in denvr/api/v1/servers/applications.py

def destroy_application(self, id: str | None = None, cluster: str | None = None) -> dict:
    """
    Permanently delete a specified application, effectively wiping all its data and freeing up resources for other uses ::

        client.destroy_application(id="my-jupyter-application", cluster="Msc1")

    Keyword Arguments:
        id (str): The application name
        cluster (str): The cluster you're operating on

    Returns:
        id (str): The application name
        cluster (str): The cluster you're operating on
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {
            "Id": config.getkwarg("id", id),
            "Cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "delete",
        "/api/v1/servers/applications/DestroyApplication",
        parameters,
        {"Id", "Cluster"},
    )

    return self.session.request(
        "delete", "/api/v1/servers/applications/DestroyApplication", **kwargs
    )

get_application_catalog_items

Get a list of application catalog items ::

client.get_application_catalog_items()

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/applications.py

def get_application_catalog_items(self) -> dict:
    """
    Get a list of application catalog items ::

        client.get_application_catalog_items()


    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {}

    kwargs = validate_kwargs(
        "get", "/api/v1/servers/applications/GetApplicationCatalogItems", parameters, {}
    )

    return self.session.request(
        "get", "/api/v1/servers/applications/GetApplicationCatalogItems", **kwargs
    )

get_application_details

Get detailed information about a specific application ::

client.get_application_details(id="my-jupyter-application", cluster="Msc1")

Other Parameters:

Name	Type	Description
id	str	The application name
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
instance_details	dict
application_catalog_item	dict
hardware_package	dict

Source code in denvr/api/v1/servers/applications.py

def get_application_details(
    self, id: str | None = None, cluster: str | None = None
) -> dict:
    """
    Get detailed information about a specific application ::

        client.get_application_details(id="my-jupyter-application", cluster="Msc1")

    Keyword Arguments:
        id (str): The application name
        cluster (str): The cluster you're operating on

    Returns:
        instance_details (dict):
        application_catalog_item (dict):
        hardware_package (dict):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {
            "Id": config.getkwarg("id", id),
            "Cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "get",
        "/api/v1/servers/applications/GetApplicationDetails",
        parameters,
        {"Id", "Cluster"},
    )

    return self.session.request(
        "get", "/api/v1/servers/applications/GetApplicationDetails", **kwargs
    )

get_applications

Get a list of applications ::

client.get_applications()

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/applications.py

def get_applications(self) -> dict:
    """
    Get a list of applications ::

        client.get_applications()


    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {}

    kwargs = validate_kwargs(
        "get", "/api/v1/servers/applications/GetApplications", parameters, {}
    )

    return self.session.request(
        "get", "/api/v1/servers/applications/GetApplications", **kwargs
    )

get_availability

Get detailed information on available configurations for applications ::

client.get_availability(cluster="Msc1", resource_pool="on-demand")

Other Parameters:

Name	Type	Description
cluster	str
resource_pool	str

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/applications.py

def get_availability(
    self, cluster: str | None = None, resource_pool: str | None = None
) -> dict:
    """
    Get detailed information on available configurations for applications ::

        client.get_availability(cluster="Msc1", resource_pool="on-demand")

    Keyword Arguments:
        cluster (str):
        resource_pool (str):

    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {
            "cluster": config.getkwarg("cluster", cluster),
            "resourcePool": config.getkwarg("resource_pool", resource_pool),
        }
    }

    kwargs = validate_kwargs(
        "get",
        "/api/v1/servers/applications/GetAvailability",
        parameters,
        {"cluster", "resourcePool"},
    )

    return self.session.request(
        "get", "/api/v1/servers/applications/GetAvailability", **kwargs
    )

get_configurations

Get a list of application configurations ::

client.get_configurations()

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/applications.py

def get_configurations(self) -> dict:
    """
    Get a list of application configurations ::

        client.get_configurations()


    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {}

    kwargs = validate_kwargs(
        "get", "/api/v1/servers/applications/GetConfigurations", parameters, {}
    )

    return self.session.request(
        "get", "/api/v1/servers/applications/GetConfigurations", **kwargs
    )

start_application

Start an application that has been previously set up and provisioned, but is currently OFFLINE ::

client.start_application(id="my-jupyter-application", cluster="Msc1")

Other Parameters:

Name	Type	Description
id	str	The application name
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str	The application name
cluster	str	The cluster you're operating on

Source code in denvr/api/v1/servers/applications.py

def start_application(self, id: str | None = None, cluster: str | None = None) -> dict:
    """
    Start an application that has been previously set up and provisioned, but is currently OFFLINE ::

        client.start_application(id="my-jupyter-application", cluster="Msc1")

    Keyword Arguments:
        id (str): The application name
        cluster (str): The cluster you're operating on

    Returns:
        id (str): The application name
        cluster (str): The cluster you're operating on
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "id": config.getkwarg("id", id),
            "cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "post",
        "/api/v1/servers/applications/StartApplication",
        parameters,
        {"cluster", "id"},
    )

    return self.session.request(
        "post", "/api/v1/servers/applications/StartApplication", **kwargs
    )

stop_application

Stop an application that has been previously set up and provisioned, but is currently ONLINE ::

client.stop_application(id="my-jupyter-application", cluster="Msc1")

Other Parameters:

Name	Type	Description
id	str	The application name
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str	The application name
cluster	str	The cluster you're operating on

Source code in denvr/api/v1/servers/applications.py

def stop_application(self, id: str | None = None, cluster: str | None = None) -> dict:
    """
    Stop an application that has been previously set up and provisioned, but is currently ONLINE ::

        client.stop_application(id="my-jupyter-application", cluster="Msc1")

    Keyword Arguments:
        id (str): The application name
        cluster (str): The cluster you're operating on

    Returns:
        id (str): The application name
        cluster (str): The cluster you're operating on
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "id": config.getkwarg("id", id),
            "cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "post",
        "/api/v1/servers/applications/StopApplication",
        parameters,
        {"cluster", "id"},
    )

    return self.session.request(
        "post", "/api/v1/servers/applications/StopApplication", **kwargs
    )

images

Client

Source code in denvr/api/v1/servers/images.py

class Client:
    def __init__(self, session: Session):
        self.session = session

    def get_operating_system_images(self) -> dict:
        """
        Get a list of operating sytem images available for the tenant ::

            client.get_operating_system_images()


        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {}

        kwargs = validate_kwargs(
            "get", "/api/v1/servers/images/GetOperatingSystemImages", parameters, {}
        )

        return self.session.request(
            "get", "/api/v1/servers/images/GetOperatingSystemImages", **kwargs
        )

get_operating_system_images

Get a list of operating sytem images available for the tenant ::

client.get_operating_system_images()

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/images.py

def get_operating_system_images(self) -> dict:
    """
    Get a list of operating sytem images available for the tenant ::

        client.get_operating_system_images()


    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {}

    kwargs = validate_kwargs(
        "get", "/api/v1/servers/images/GetOperatingSystemImages", parameters, {}
    )

    return self.session.request(
        "get", "/api/v1/servers/images/GetOperatingSystemImages", **kwargs
    )

metal

Client

Source code in denvr/api/v1/servers/metal.py

class Client:
    def __init__(self, session: Session):
        self.session = session

    def get_host(self, id: str | None = None, cluster: str | None = None) -> dict:
        """
        Get detailed information about a specific metal host ::

            client.get_host(id="Id", cluster="Hou1")

        Keyword Arguments:
            id (str): Unique identifier for a resource within the cluster
            cluster (str): The cluster you're operating on

        Returns:
            id (str): The bare metal id, unique identifier
            cluster (str): The cluster where the bare metal host is allocated
            tenancy_name (str): Name of the tenant where the node has been allocated
            node_type (str): The specific host node type
            image (str): The image used to provision the host
            private_ip (str): private IP address of the host
            provisioned_hostname (str): host name provisioned by the system
            operational_status (str): operational status of the host
            powered_on (bool): true if the host is powered on
            provisioning_state (str): provisioning status of the host
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {
                "Id": config.getkwarg("id", id),
                "Cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "get", "/api/v1/servers/metal/GetHost", parameters, {"Id", "Cluster"}
        )

        return self.session.request("get", "/api/v1/servers/metal/GetHost", **kwargs)

    def get_hosts(self, cluster: str | None = None) -> dict:
        """
        Get a list of bare metal hosts in a cluster ::

            client.get_hosts(cluster="Hou1")

        Keyword Arguments:
            cluster (str):

        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {"Cluster": config.getkwarg("cluster", cluster)}
        }

        kwargs = validate_kwargs("get", "/api/v1/servers/metal/GetHosts", parameters, {})

        return self.session.request("get", "/api/v1/servers/metal/GetHosts", **kwargs)

    def reboot_host(self, id: str | None = None, cluster: str | None = None) -> dict:
        """
        Reboot the bare metal host ::

            client.reboot_host(id="string", cluster="Hou1")

        Keyword Arguments:
            id (str): Unique identifier for a resource within the cluster
            cluster (str): The cluster you're operating on

        Returns:
            id (str): The bare metal id, unique identifier
            cluster (str): The cluster where the bare metal host is allocated
            tenancy_name (str): Name of the tenant where the node has been allocated
            node_type (str): The specific host node type
            image (str): The image used to provision the host
            private_ip (str): private IP address of the host
            provisioned_hostname (str): host name provisioned by the system
            operational_status (str): operational status of the host
            powered_on (bool): true if the host is powered on
            provisioning_state (str): provisioning status of the host
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "id": config.getkwarg("id", id),
                "cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "post", "/api/v1/servers/metal/RebootHost", parameters, {"cluster", "id"}
        )

        return self.session.request("post", "/api/v1/servers/metal/RebootHost", **kwargs)

    def reprovision_host(
        self,
        image_url: str | None = None,
        image_checksum: str | None = None,
        cloud_init_base64: str | None = None,
        id: str | None = None,
        cluster: str | None = None,
    ) -> dict:
        """
        Reprovision the bare metal host ::

            client.reprovision_host(
                image_url="https://cloud-images.ubuntu.com/jammy/current/jammy-server-cloudimg-amd64.img",
                image_checksum="https://cloud-images.ubuntu.com/jammy/current/MD5SUMS",
                cloud_init_base64="SGVsbG8sIFdvcmxkIQ==",
                id="string",
                cluster="Hou1",
            )

        Keyword Arguments:
            image_url (str): The URL to the image to use for the host
            image_checksum (str): The checksum url of the image to use for the host
            cloud_init_base64 (str): Base64 encoded cloud-init data yaml file to use for the host
            id (str): Unique identifier for a resource within the cluster
            cluster (str): The cluster you're operating on

        Returns:
            id (str): The bare metal id, unique identifier
            cluster (str): The cluster where the bare metal host is allocated
            tenancy_name (str): Name of the tenant where the node has been allocated
            node_type (str): The specific host node type
            image (str): The image used to provision the host
            private_ip (str): private IP address of the host
            provisioned_hostname (str): host name provisioned by the system
            operational_status (str): operational status of the host
            powered_on (bool): true if the host is powered on
            provisioning_state (str): provisioning status of the host
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "imageUrl": config.getkwarg("image_url", image_url),
                "imageChecksum": config.getkwarg("image_checksum", image_checksum),
                "cloudInitBase64": config.getkwarg("cloud_init_base64", cloud_init_base64),
                "id": config.getkwarg("id", id),
                "cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "post", "/api/v1/servers/metal/ReprovisionHost", parameters, {"cluster", "id"}
        )

        return self.session.request("post", "/api/v1/servers/metal/ReprovisionHost", **kwargs)

get_host

Get detailed information about a specific metal host ::

client.get_host(id="Id", cluster="Hou1")

Other Parameters:

Name	Type	Description
id	str	Unique identifier for a resource within the cluster
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str	The bare metal id, unique identifier
cluster	str	The cluster where the bare metal host is allocated
tenancy_name	str	Name of the tenant where the node has been allocated
node_type	str	The specific host node type
image	str	The image used to provision the host
private_ip	str	private IP address of the host
provisioned_hostname	str	host name provisioned by the system
operational_status	str	operational status of the host
powered_on	bool	true if the host is powered on
provisioning_state	str	provisioning status of the host

Source code in denvr/api/v1/servers/metal.py

def get_host(self, id: str | None = None, cluster: str | None = None) -> dict:
    """
    Get detailed information about a specific metal host ::

        client.get_host(id="Id", cluster="Hou1")

    Keyword Arguments:
        id (str): Unique identifier for a resource within the cluster
        cluster (str): The cluster you're operating on

    Returns:
        id (str): The bare metal id, unique identifier
        cluster (str): The cluster where the bare metal host is allocated
        tenancy_name (str): Name of the tenant where the node has been allocated
        node_type (str): The specific host node type
        image (str): The image used to provision the host
        private_ip (str): private IP address of the host
        provisioned_hostname (str): host name provisioned by the system
        operational_status (str): operational status of the host
        powered_on (bool): true if the host is powered on
        provisioning_state (str): provisioning status of the host
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {
            "Id": config.getkwarg("id", id),
            "Cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "get", "/api/v1/servers/metal/GetHost", parameters, {"Id", "Cluster"}
    )

    return self.session.request("get", "/api/v1/servers/metal/GetHost", **kwargs)

get_hosts

Get a list of bare metal hosts in a cluster ::

client.get_hosts(cluster="Hou1")

Other Parameters:

Name	Type	Description
cluster	str

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/metal.py

def get_hosts(self, cluster: str | None = None) -> dict:
    """
    Get a list of bare metal hosts in a cluster ::

        client.get_hosts(cluster="Hou1")

    Keyword Arguments:
        cluster (str):

    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {"Cluster": config.getkwarg("cluster", cluster)}
    }

    kwargs = validate_kwargs("get", "/api/v1/servers/metal/GetHosts", parameters, {})

    return self.session.request("get", "/api/v1/servers/metal/GetHosts", **kwargs)

reboot_host

Reboot the bare metal host ::

client.reboot_host(id="string", cluster="Hou1")

Other Parameters:

Name	Type	Description
id	str	Unique identifier for a resource within the cluster
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str	The bare metal id, unique identifier
cluster	str	The cluster where the bare metal host is allocated
tenancy_name	str	Name of the tenant where the node has been allocated
node_type	str	The specific host node type
image	str	The image used to provision the host
private_ip	str	private IP address of the host
provisioned_hostname	str	host name provisioned by the system
operational_status	str	operational status of the host
powered_on	bool	true if the host is powered on
provisioning_state	str	provisioning status of the host

Source code in denvr/api/v1/servers/metal.py

def reboot_host(self, id: str | None = None, cluster: str | None = None) -> dict:
    """
    Reboot the bare metal host ::

        client.reboot_host(id="string", cluster="Hou1")

    Keyword Arguments:
        id (str): Unique identifier for a resource within the cluster
        cluster (str): The cluster you're operating on

    Returns:
        id (str): The bare metal id, unique identifier
        cluster (str): The cluster where the bare metal host is allocated
        tenancy_name (str): Name of the tenant where the node has been allocated
        node_type (str): The specific host node type
        image (str): The image used to provision the host
        private_ip (str): private IP address of the host
        provisioned_hostname (str): host name provisioned by the system
        operational_status (str): operational status of the host
        powered_on (bool): true if the host is powered on
        provisioning_state (str): provisioning status of the host
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "id": config.getkwarg("id", id),
            "cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "post", "/api/v1/servers/metal/RebootHost", parameters, {"cluster", "id"}
    )

    return self.session.request("post", "/api/v1/servers/metal/RebootHost", **kwargs)

reprovision_host

Reprovision the bare metal host ::

client.reprovision_host(
    image_url="https://cloud-images.ubuntu.com/jammy/current/jammy-server-cloudimg-amd64.img",
    image_checksum="https://cloud-images.ubuntu.com/jammy/current/MD5SUMS",
    cloud_init_base64="SGVsbG8sIFdvcmxkIQ==",
    id="string",
    cluster="Hou1",
)

Other Parameters:

Name	Type	Description
image_url	str	The URL to the image to use for the host
image_checksum	str	The checksum url of the image to use for the host
cloud_init_base64	str	Base64 encoded cloud-init data yaml file to use for the host
id	str	Unique identifier for a resource within the cluster
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str	The bare metal id, unique identifier
cluster	str	The cluster where the bare metal host is allocated
tenancy_name	str	Name of the tenant where the node has been allocated
node_type	str	The specific host node type
image	str	The image used to provision the host
private_ip	str	private IP address of the host
provisioned_hostname	str	host name provisioned by the system
operational_status	str	operational status of the host
powered_on	bool	true if the host is powered on
provisioning_state	str	provisioning status of the host

Source code in denvr/api/v1/servers/metal.py

def reprovision_host(
    self,
    image_url: str | None = None,
    image_checksum: str | None = None,
    cloud_init_base64: str | None = None,
    id: str | None = None,
    cluster: str | None = None,
) -> dict:
    """
    Reprovision the bare metal host ::

        client.reprovision_host(
            image_url="https://cloud-images.ubuntu.com/jammy/current/jammy-server-cloudimg-amd64.img",
            image_checksum="https://cloud-images.ubuntu.com/jammy/current/MD5SUMS",
            cloud_init_base64="SGVsbG8sIFdvcmxkIQ==",
            id="string",
            cluster="Hou1",
        )

    Keyword Arguments:
        image_url (str): The URL to the image to use for the host
        image_checksum (str): The checksum url of the image to use for the host
        cloud_init_base64 (str): Base64 encoded cloud-init data yaml file to use for the host
        id (str): Unique identifier for a resource within the cluster
        cluster (str): The cluster you're operating on

    Returns:
        id (str): The bare metal id, unique identifier
        cluster (str): The cluster where the bare metal host is allocated
        tenancy_name (str): Name of the tenant where the node has been allocated
        node_type (str): The specific host node type
        image (str): The image used to provision the host
        private_ip (str): private IP address of the host
        provisioned_hostname (str): host name provisioned by the system
        operational_status (str): operational status of the host
        powered_on (bool): true if the host is powered on
        provisioning_state (str): provisioning status of the host
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "imageUrl": config.getkwarg("image_url", image_url),
            "imageChecksum": config.getkwarg("image_checksum", image_checksum),
            "cloudInitBase64": config.getkwarg("cloud_init_base64", cloud_init_base64),
            "id": config.getkwarg("id", id),
            "cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "post", "/api/v1/servers/metal/ReprovisionHost", parameters, {"cluster", "id"}
    )

    return self.session.request("post", "/api/v1/servers/metal/ReprovisionHost", **kwargs)

virtual

Client

Source code in denvr/api/v1/servers/virtual.py

class Client:
    def __init__(self, session: Session):
        self.session = session

    def get_servers(self, cluster: str | None = None) -> dict:
        """
        Get a list of virtual machines ::

            client.get_servers(cluster="Cluster")

        Keyword Arguments:
            cluster (str):

        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {"Cluster": config.getkwarg("cluster", cluster)}
        }

        kwargs = validate_kwargs("get", "/api/v1/servers/virtual/GetServers", parameters, {})

        return self.session.request("get", "/api/v1/servers/virtual/GetServers", **kwargs)

    def get_server(
        self, id: str | None = None, namespace: str | None = None, cluster: str | None = None
    ) -> dict:
        """
        Get detailed information about a specific virtual machine ::

            client.get_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

        Keyword Arguments:
            id (str): The virtual machine id
            namespace (str): The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
            cluster (str): The cluster you're operating on

        Returns:
            username (str): The user that creatd the vm
            tenancy_name (str): Name of the tenant where the VM has been created
            rpool (str): Resource pool where the VM has been created
            direct_attached_storage_persisted (bool):
            id (str): The name of the virtual machine
            namespace (str):
            configuration (str): A VM configuration ID
            storage (int): The amount of storage attached to the VM in GB
            gpu_type (str): The specific host GPU type
            gpus (int): Number of GPUs attached to the VM
            vcpus (int): Number of vCPUs available to the VM
            memory (int): Amount of system memory available in GB
            ip (str): The public IP address of the VM
            private_ip (str): The private IP address of the VM
            image (str): Name of the VM image used
            cluster (str): The cluster where the VM is allocated
            status (str): The status of the VM (e.g. 'PLANNED', 'PENDING' 'PENDING_RESOURCES', 'PENDING_READINESS',...
            storage_type (str):
            root_disk_size (str):
            last_updated (str):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {
                "Id": config.getkwarg("id", id),
                "Namespace": config.getkwarg("namespace", namespace),
                "Cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "get",
            "/api/v1/servers/virtual/GetServer",
            parameters,
            {"Id", "Namespace", "Cluster"},
        )

        return self.session.request("get", "/api/v1/servers/virtual/GetServer", **kwargs)

    def create_server(
        self,
        name: str | None = None,
        rpool: str | None = None,
        vpc: str | None = None,
        configuration: str | None = None,
        cluster: str | None = None,
        ssh_keys: list | None = None,
        snapshot_name: str | None = None,
        operating_system_image: str | None = None,
        personal_storage_mount_path: str | None = None,
        tenant_shared_additional_storage: str | None = None,
        persist_storage: bool | None = None,
        direct_storage_mount_path: str | None = None,
        root_disk_size: int | None = None,
    ) -> dict:
        """
        Create a new virtual machine using a pre-defined configuration ::

            client.create_server(
                name="my-denvr-vm",
                rpool="reserved-denvr",
                vpc="denvr-vpc",
                configuration="A100_40GB_PCIe_1x",
                cluster="Hou1",
                ssh_keys=["string"],
                snapshot_name="string",
                operating_system_image="Ubuntu 22.04.4 LTS",
                personal_storage_mount_path="/home/ubuntu/personal",
                tenant_shared_additional_storage="/home/ubuntu/tenant-shared",
                persist_storage=False,
                direct_storage_mount_path="/home/ubuntu/direct-attached",
                root_disk_size=500,
            )

        Keyword Arguments:
            name (str): Name of virtual server to be created. If not provided, name will be auto-generated.
            rpool (str): Name of the pool to be used. If not provided, first pool assigned to a tenant will be used. In...
            vpc (str): Name of the VPC to be used. Usually this will match the tenant name.
            configuration (str): Name of the configuration to be used. For possible values, refer to the otput of...
            cluster (str): Cluster to be used. For possible values, refer to the otput of api/v1/clusters/GetAll"/>
            ssh_keys (list):
            snapshot_name (str): Snapshot name.
            operating_system_image (str): Name of the Operating System image to be used.
            personal_storage_mount_path (str): Personal storage file system mount path.
            tenant_shared_additional_storage (str): Tenant shared storage file system mount path.
            persist_storage (bool): Whether direct attached storage should be persistant or ephemeral.
            direct_storage_mount_path (str): Direct attached storage mount path.
            root_disk_size (int): Size of root disk to be created (Gi).

        Returns:
            username (str): The user that creatd the vm
            tenancy_name (str): Name of the tenant where the VM has been created
            rpool (str): Resource pool where the VM has been created
            direct_attached_storage_persisted (bool):
            id (str): The name of the virtual machine
            namespace (str):
            configuration (str): A VM configuration ID
            storage (int): The amount of storage attached to the VM in GB
            gpu_type (str): The specific host GPU type
            gpus (int): Number of GPUs attached to the VM
            vcpus (int): Number of vCPUs available to the VM
            memory (int): Amount of system memory available in GB
            ip (str): The public IP address of the VM
            private_ip (str): The private IP address of the VM
            image (str): Name of the VM image used
            cluster (str): The cluster where the VM is allocated
            status (str): The status of the VM (e.g. 'PLANNED', 'PENDING' 'PENDING_RESOURCES', 'PENDING_READINESS',...
            storage_type (str):
            root_disk_size (str):
            last_updated (str):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "name": config.getkwarg("name", name),
                "rpool": config.getkwarg("rpool", rpool),
                "vpc": config.getkwarg("vpc", vpc),
                "configuration": config.getkwarg("configuration", configuration),
                "cluster": config.getkwarg("cluster", cluster),
                "ssh_keys": config.getkwarg("ssh_keys", ssh_keys),
                "snapshotName": config.getkwarg("snapshot_name", snapshot_name),
                "operatingSystemImage": config.getkwarg(
                    "operating_system_image", operating_system_image
                ),
                "personalStorageMountPath": config.getkwarg(
                    "personal_storage_mount_path", personal_storage_mount_path
                ),
                "tenantSharedAdditionalStorage": config.getkwarg(
                    "tenant_shared_additional_storage", tenant_shared_additional_storage
                ),
                "persistStorage": config.getkwarg("persist_storage", persist_storage),
                "directStorageMountPath": config.getkwarg(
                    "direct_storage_mount_path", direct_storage_mount_path
                ),
                "rootDiskSize": config.getkwarg("root_disk_size", root_disk_size),
            }
        }

        kwargs = validate_kwargs(
            "post",
            "/api/v1/servers/virtual/CreateServer",
            parameters,
            {"cluster", "configuration", "ssh_keys", "vpc"},
        )

        return self.session.request("post", "/api/v1/servers/virtual/CreateServer", **kwargs)

    def start_server(
        self, id: str | None = None, namespace: str | None = None, cluster: str | None = None
    ) -> dict:
        """
        Start a virtual machine that has been previously set up and provisioned, but is currently OFFLINE ::

            client.start_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

        Keyword Arguments:
            id (str): The virtual machine id
            namespace (str): The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
            cluster (str): The cluster you're operating on

        Returns:
            id (str):
            cluster (str):
            status (str):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "id": config.getkwarg("id", id),
                "namespace": config.getkwarg("namespace", namespace),
                "cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "post",
            "/api/v1/servers/virtual/StartServer",
            parameters,
            {"cluster", "id", "namespace"},
        )

        return self.session.request("post", "/api/v1/servers/virtual/StartServer", **kwargs)

    def stop_server(
        self, id: str | None = None, namespace: str | None = None, cluster: str | None = None
    ) -> dict:
        """
        Stop a virtual machine, ensuring a secure and orderly shutdown of its operations within the cloud environment ::

            client.stop_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

        Keyword Arguments:
            id (str): The virtual machine id
            namespace (str): The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
            cluster (str): The cluster you're operating on

        Returns:
            id (str):
            cluster (str):
            status (str):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "json": {
                "id": config.getkwarg("id", id),
                "namespace": config.getkwarg("namespace", namespace),
                "cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "post",
            "/api/v1/servers/virtual/StopServer",
            parameters,
            {"cluster", "id", "namespace"},
        )

        return self.session.request("post", "/api/v1/servers/virtual/StopServer", **kwargs)

    def destroy_server(
        self,
        delete_snapshots: bool | None = None,
        id: str | None = None,
        namespace: str | None = None,
        cluster: str | None = None,
    ) -> dict:
        """
        Permanently delete a specified virtual machine, effectively wiping all its data and freeing up resources for other uses ::

            client.destroy_server(
                delete_snapshots=True, id="vm-2024093009357617", namespace="denvr", cluster="Hou1"
            )

        Keyword Arguments:
            delete_snapshots (bool): Should also delete snapshots with virtual machine.
            id (str): The virtual machine id
            namespace (str): The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
            cluster (str): The cluster you're operating on

        Returns:
            id (str):
            cluster (str):
            status (str):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {
                "DeleteSnapshots": config.getkwarg("delete_snapshots", delete_snapshots),
                "Id": config.getkwarg("id", id),
                "Namespace": config.getkwarg("namespace", namespace),
                "Cluster": config.getkwarg("cluster", cluster),
            }
        }

        kwargs = validate_kwargs(
            "delete",
            "/api/v1/servers/virtual/DestroyServer",
            parameters,
            {"Id", "Namespace", "Cluster"},
        )

        return self.session.request("delete", "/api/v1/servers/virtual/DestroyServer", **kwargs)

    def get_configurations(self) -> dict:
        """
        Get detailed information on available configurations for virtual machines ::

            client.get_configurations()


        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {}

        kwargs = validate_kwargs(
            "get", "/api/v1/servers/virtual/GetConfigurations", parameters, {}
        )

        return self.session.request(
            "get", "/api/v1/servers/virtual/GetConfigurations", **kwargs
        )

    def get_availability(
        self,
        cluster: str | None = None,
        resource_pool: str | None = None,
        report_nodes: bool | None = None,
    ) -> dict:
        """
        Get information about the current availability of different virtual machine configurations ::

            client.get_availability(cluster="Hou1", resource_pool="reserved-denvr", report_nodes=True)

        Keyword Arguments:
            cluster (str):
            resource_pool (str):
            report_nodes (bool): controls if Count and MaxCount is calculated and returned in the response. If they are not...

        Returns:
            items (list):
        """
        config = self.session.config  # noqa: F841

        parameters: dict[str, dict] = {
            "params": {
                "cluster": config.getkwarg("cluster", cluster),
                "resourcePool": config.getkwarg("resource_pool", resource_pool),
                "reportNodes": config.getkwarg("report_nodes", report_nodes),
            }
        }

        kwargs = validate_kwargs(
            "get", "/api/v1/servers/virtual/GetAvailability", parameters, {"cluster"}
        )

        return self.session.request("get", "/api/v1/servers/virtual/GetAvailability", **kwargs)

create_server

Create a new virtual machine using a pre-defined configuration ::

client.create_server(
    name="my-denvr-vm",
    rpool="reserved-denvr",
    vpc="denvr-vpc",
    configuration="A100_40GB_PCIe_1x",
    cluster="Hou1",
    ssh_keys=["string"],
    snapshot_name="string",
    operating_system_image="Ubuntu 22.04.4 LTS",
    personal_storage_mount_path="/home/ubuntu/personal",
    tenant_shared_additional_storage="/home/ubuntu/tenant-shared",
    persist_storage=False,
    direct_storage_mount_path="/home/ubuntu/direct-attached",
    root_disk_size=500,
)

Other Parameters:

Name	Type	Description
name	str	Name of virtual server to be created. If not provided, name will be auto-generated.
rpool	str	Name of the pool to be used. If not provided, first pool assigned to a tenant will be used. In...
vpc	str	Name of the VPC to be used. Usually this will match the tenant name.
configuration	str	Name of the configuration to be used. For possible values, refer to the otput of...
cluster	str	Cluster to be used. For possible values, refer to the otput of api/v1/clusters/GetAll"/>
ssh_keys	list
snapshot_name	str	Snapshot name.
operating_system_image	str	Name of the Operating System image to be used.
personal_storage_mount_path	str	Personal storage file system mount path.
tenant_shared_additional_storage	str	Tenant shared storage file system mount path.
persist_storage	bool	Whether direct attached storage should be persistant or ephemeral.
direct_storage_mount_path	str	Direct attached storage mount path.
root_disk_size	int	Size of root disk to be created (Gi).

Returns:

Name	Type	Description
username	str	The user that creatd the vm
tenancy_name	str	Name of the tenant where the VM has been created
rpool	str	Resource pool where the VM has been created
direct_attached_storage_persisted	bool
id	str	The name of the virtual machine
namespace	str
configuration	str	A VM configuration ID
storage	int	The amount of storage attached to the VM in GB
gpu_type	str	The specific host GPU type
gpus	int	Number of GPUs attached to the VM
vcpus	int	Number of vCPUs available to the VM
memory	int	Amount of system memory available in GB
ip	str	The public IP address of the VM
private_ip	str	The private IP address of the VM
image	str	Name of the VM image used
cluster	str	The cluster where the VM is allocated
status	str	The status of the VM (e.g. 'PLANNED', 'PENDING' 'PENDING_RESOURCES', 'PENDING_READINESS',...
storage_type	str
root_disk_size	str
last_updated	str

Source code in denvr/api/v1/servers/virtual.py

def create_server(
    self,
    name: str | None = None,
    rpool: str | None = None,
    vpc: str | None = None,
    configuration: str | None = None,
    cluster: str | None = None,
    ssh_keys: list | None = None,
    snapshot_name: str | None = None,
    operating_system_image: str | None = None,
    personal_storage_mount_path: str | None = None,
    tenant_shared_additional_storage: str | None = None,
    persist_storage: bool | None = None,
    direct_storage_mount_path: str | None = None,
    root_disk_size: int | None = None,
) -> dict:
    """
    Create a new virtual machine using a pre-defined configuration ::

        client.create_server(
            name="my-denvr-vm",
            rpool="reserved-denvr",
            vpc="denvr-vpc",
            configuration="A100_40GB_PCIe_1x",
            cluster="Hou1",
            ssh_keys=["string"],
            snapshot_name="string",
            operating_system_image="Ubuntu 22.04.4 LTS",
            personal_storage_mount_path="/home/ubuntu/personal",
            tenant_shared_additional_storage="/home/ubuntu/tenant-shared",
            persist_storage=False,
            direct_storage_mount_path="/home/ubuntu/direct-attached",
            root_disk_size=500,
        )

    Keyword Arguments:
        name (str): Name of virtual server to be created. If not provided, name will be auto-generated.
        rpool (str): Name of the pool to be used. If not provided, first pool assigned to a tenant will be used. In...
        vpc (str): Name of the VPC to be used. Usually this will match the tenant name.
        configuration (str): Name of the configuration to be used. For possible values, refer to the otput of...
        cluster (str): Cluster to be used. For possible values, refer to the otput of api/v1/clusters/GetAll"/>
        ssh_keys (list):
        snapshot_name (str): Snapshot name.
        operating_system_image (str): Name of the Operating System image to be used.
        personal_storage_mount_path (str): Personal storage file system mount path.
        tenant_shared_additional_storage (str): Tenant shared storage file system mount path.
        persist_storage (bool): Whether direct attached storage should be persistant or ephemeral.
        direct_storage_mount_path (str): Direct attached storage mount path.
        root_disk_size (int): Size of root disk to be created (Gi).

    Returns:
        username (str): The user that creatd the vm
        tenancy_name (str): Name of the tenant where the VM has been created
        rpool (str): Resource pool where the VM has been created
        direct_attached_storage_persisted (bool):
        id (str): The name of the virtual machine
        namespace (str):
        configuration (str): A VM configuration ID
        storage (int): The amount of storage attached to the VM in GB
        gpu_type (str): The specific host GPU type
        gpus (int): Number of GPUs attached to the VM
        vcpus (int): Number of vCPUs available to the VM
        memory (int): Amount of system memory available in GB
        ip (str): The public IP address of the VM
        private_ip (str): The private IP address of the VM
        image (str): Name of the VM image used
        cluster (str): The cluster where the VM is allocated
        status (str): The status of the VM (e.g. 'PLANNED', 'PENDING' 'PENDING_RESOURCES', 'PENDING_READINESS',...
        storage_type (str):
        root_disk_size (str):
        last_updated (str):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "name": config.getkwarg("name", name),
            "rpool": config.getkwarg("rpool", rpool),
            "vpc": config.getkwarg("vpc", vpc),
            "configuration": config.getkwarg("configuration", configuration),
            "cluster": config.getkwarg("cluster", cluster),
            "ssh_keys": config.getkwarg("ssh_keys", ssh_keys),
            "snapshotName": config.getkwarg("snapshot_name", snapshot_name),
            "operatingSystemImage": config.getkwarg(
                "operating_system_image", operating_system_image
            ),
            "personalStorageMountPath": config.getkwarg(
                "personal_storage_mount_path", personal_storage_mount_path
            ),
            "tenantSharedAdditionalStorage": config.getkwarg(
                "tenant_shared_additional_storage", tenant_shared_additional_storage
            ),
            "persistStorage": config.getkwarg("persist_storage", persist_storage),
            "directStorageMountPath": config.getkwarg(
                "direct_storage_mount_path", direct_storage_mount_path
            ),
            "rootDiskSize": config.getkwarg("root_disk_size", root_disk_size),
        }
    }

    kwargs = validate_kwargs(
        "post",
        "/api/v1/servers/virtual/CreateServer",
        parameters,
        {"cluster", "configuration", "ssh_keys", "vpc"},
    )

    return self.session.request("post", "/api/v1/servers/virtual/CreateServer", **kwargs)

destroy_server

Permanently delete a specified virtual machine, effectively wiping all its data and freeing up resources for other uses ::

client.destroy_server(
    delete_snapshots=True, id="vm-2024093009357617", namespace="denvr", cluster="Hou1"
)

Other Parameters:

Name	Type	Description
delete_snapshots	bool	Should also delete snapshots with virtual machine.
id	str	The virtual machine id
namespace	str	The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str
cluster	str
status	str

Source code in denvr/api/v1/servers/virtual.py

def destroy_server(
    self,
    delete_snapshots: bool | None = None,
    id: str | None = None,
    namespace: str | None = None,
    cluster: str | None = None,
) -> dict:
    """
    Permanently delete a specified virtual machine, effectively wiping all its data and freeing up resources for other uses ::

        client.destroy_server(
            delete_snapshots=True, id="vm-2024093009357617", namespace="denvr", cluster="Hou1"
        )

    Keyword Arguments:
        delete_snapshots (bool): Should also delete snapshots with virtual machine.
        id (str): The virtual machine id
        namespace (str): The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
        cluster (str): The cluster you're operating on

    Returns:
        id (str):
        cluster (str):
        status (str):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {
            "DeleteSnapshots": config.getkwarg("delete_snapshots", delete_snapshots),
            "Id": config.getkwarg("id", id),
            "Namespace": config.getkwarg("namespace", namespace),
            "Cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "delete",
        "/api/v1/servers/virtual/DestroyServer",
        parameters,
        {"Id", "Namespace", "Cluster"},
    )

    return self.session.request("delete", "/api/v1/servers/virtual/DestroyServer", **kwargs)

get_availability

Get information about the current availability of different virtual machine configurations ::

client.get_availability(cluster="Hou1", resource_pool="reserved-denvr", report_nodes=True)

Other Parameters:

Name	Type	Description
cluster	str
resource_pool	str
report_nodes	bool	controls if Count and MaxCount is calculated and returned in the response. If they are not...

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/virtual.py

def get_availability(
    self,
    cluster: str | None = None,
    resource_pool: str | None = None,
    report_nodes: bool | None = None,
) -> dict:
    """
    Get information about the current availability of different virtual machine configurations ::

        client.get_availability(cluster="Hou1", resource_pool="reserved-denvr", report_nodes=True)

    Keyword Arguments:
        cluster (str):
        resource_pool (str):
        report_nodes (bool): controls if Count and MaxCount is calculated and returned in the response. If they are not...

    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {
            "cluster": config.getkwarg("cluster", cluster),
            "resourcePool": config.getkwarg("resource_pool", resource_pool),
            "reportNodes": config.getkwarg("report_nodes", report_nodes),
        }
    }

    kwargs = validate_kwargs(
        "get", "/api/v1/servers/virtual/GetAvailability", parameters, {"cluster"}
    )

    return self.session.request("get", "/api/v1/servers/virtual/GetAvailability", **kwargs)

get_configurations

Get detailed information on available configurations for virtual machines ::

client.get_configurations()

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/virtual.py

def get_configurations(self) -> dict:
    """
    Get detailed information on available configurations for virtual machines ::

        client.get_configurations()


    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {}

    kwargs = validate_kwargs(
        "get", "/api/v1/servers/virtual/GetConfigurations", parameters, {}
    )

    return self.session.request(
        "get", "/api/v1/servers/virtual/GetConfigurations", **kwargs
    )

get_server

Get detailed information about a specific virtual machine ::

client.get_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

Other Parameters:

Name	Type	Description
id	str	The virtual machine id
namespace	str	The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
username	str	The user that creatd the vm
tenancy_name	str	Name of the tenant where the VM has been created
rpool	str	Resource pool where the VM has been created
direct_attached_storage_persisted	bool
id	str	The name of the virtual machine
namespace	str
configuration	str	A VM configuration ID
storage	int	The amount of storage attached to the VM in GB
gpu_type	str	The specific host GPU type
gpus	int	Number of GPUs attached to the VM
vcpus	int	Number of vCPUs available to the VM
memory	int	Amount of system memory available in GB
ip	str	The public IP address of the VM
private_ip	str	The private IP address of the VM
image	str	Name of the VM image used
cluster	str	The cluster where the VM is allocated
status	str	The status of the VM (e.g. 'PLANNED', 'PENDING' 'PENDING_RESOURCES', 'PENDING_READINESS',...
storage_type	str
root_disk_size	str
last_updated	str

Source code in denvr/api/v1/servers/virtual.py

def get_server(
    self, id: str | None = None, namespace: str | None = None, cluster: str | None = None
) -> dict:
    """
    Get detailed information about a specific virtual machine ::

        client.get_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

    Keyword Arguments:
        id (str): The virtual machine id
        namespace (str): The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
        cluster (str): The cluster you're operating on

    Returns:
        username (str): The user that creatd the vm
        tenancy_name (str): Name of the tenant where the VM has been created
        rpool (str): Resource pool where the VM has been created
        direct_attached_storage_persisted (bool):
        id (str): The name of the virtual machine
        namespace (str):
        configuration (str): A VM configuration ID
        storage (int): The amount of storage attached to the VM in GB
        gpu_type (str): The specific host GPU type
        gpus (int): Number of GPUs attached to the VM
        vcpus (int): Number of vCPUs available to the VM
        memory (int): Amount of system memory available in GB
        ip (str): The public IP address of the VM
        private_ip (str): The private IP address of the VM
        image (str): Name of the VM image used
        cluster (str): The cluster where the VM is allocated
        status (str): The status of the VM (e.g. 'PLANNED', 'PENDING' 'PENDING_RESOURCES', 'PENDING_READINESS',...
        storage_type (str):
        root_disk_size (str):
        last_updated (str):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {
            "Id": config.getkwarg("id", id),
            "Namespace": config.getkwarg("namespace", namespace),
            "Cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "get",
        "/api/v1/servers/virtual/GetServer",
        parameters,
        {"Id", "Namespace", "Cluster"},
    )

    return self.session.request("get", "/api/v1/servers/virtual/GetServer", **kwargs)

get_servers

Get a list of virtual machines ::

client.get_servers(cluster="Cluster")

Other Parameters:

Name	Type	Description
cluster	str

Returns:

Name	Type	Description
items	list

Source code in denvr/api/v1/servers/virtual.py

def get_servers(self, cluster: str | None = None) -> dict:
    """
    Get a list of virtual machines ::

        client.get_servers(cluster="Cluster")

    Keyword Arguments:
        cluster (str):

    Returns:
        items (list):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "params": {"Cluster": config.getkwarg("cluster", cluster)}
    }

    kwargs = validate_kwargs("get", "/api/v1/servers/virtual/GetServers", parameters, {})

    return self.session.request("get", "/api/v1/servers/virtual/GetServers", **kwargs)

start_server

Start a virtual machine that has been previously set up and provisioned, but is currently OFFLINE ::

client.start_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

Other Parameters:

Name	Type	Description
id	str	The virtual machine id
namespace	str	The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str
cluster	str
status	str

Source code in denvr/api/v1/servers/virtual.py

def start_server(
    self, id: str | None = None, namespace: str | None = None, cluster: str | None = None
) -> dict:
    """
    Start a virtual machine that has been previously set up and provisioned, but is currently OFFLINE ::

        client.start_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

    Keyword Arguments:
        id (str): The virtual machine id
        namespace (str): The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
        cluster (str): The cluster you're operating on

    Returns:
        id (str):
        cluster (str):
        status (str):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "id": config.getkwarg("id", id),
            "namespace": config.getkwarg("namespace", namespace),
            "cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "post",
        "/api/v1/servers/virtual/StartServer",
        parameters,
        {"cluster", "id", "namespace"},
    )

    return self.session.request("post", "/api/v1/servers/virtual/StartServer", **kwargs)

stop_server

Stop a virtual machine, ensuring a secure and orderly shutdown of its operations within the cloud environment ::

client.stop_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

Other Parameters:

Name	Type	Description
id	str	The virtual machine id
namespace	str	The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
cluster	str	The cluster you're operating on

Returns:

Name	Type	Description
id	str
cluster	str
status	str

Source code in denvr/api/v1/servers/virtual.py

def stop_server(
    self, id: str | None = None, namespace: str | None = None, cluster: str | None = None
) -> dict:
    """
    Stop a virtual machine, ensuring a secure and orderly shutdown of its operations within the cloud environment ::

        client.stop_server(id="vm-2024093009357617", namespace="denvr", cluster="Hou1")

    Keyword Arguments:
        id (str): The virtual machine id
        namespace (str): The namespace/vpc where the virtual machine lives. Default one is same as tenant name.
        cluster (str): The cluster you're operating on

    Returns:
        id (str):
        cluster (str):
        status (str):
    """
    config = self.session.config  # noqa: F841

    parameters: dict[str, dict] = {
        "json": {
            "id": config.getkwarg("id", id),
            "namespace": config.getkwarg("namespace", namespace),
            "cluster": config.getkwarg("cluster", cluster),
        }
    }

    kwargs = validate_kwargs(
        "post",
        "/api/v1/servers/virtual/StopServer",
        parameters,
        {"cluster", "id", "namespace"},
    )

    return self.session.request("post", "/api/v1/servers/virtual/StopServer", **kwargs)

auth

Auth

Bases: AuthBase

Auth(server, username, password)

Handles authorization, renewal and logouts given a username and password.

Source code in denvr/auth.py

class Auth(AuthBase):
    """
    Auth(server, username, password)

    Handles authorization, renewal and logouts given a
    username and password.
    """

    def __init__(self, server, username, password, retries=3):
        self._server = server
        self._session = requests.Session()
        self._session.headers.update({"Content-type": "application/json"})
        if retries:
            self._session.mount(
                self._server,
                HTTPAdapter(max_retries=retry(retries=retries, idempotent_only=False)),
            )

        # Requests an initial authorization token
        # storing the username, password, token / refresh tokens and when they expire
        resp = self._session.post(
            f"{self._server}/api/TokenAuth/Authenticate",
            json={"userNameOrEmailAddress": username, "password": password},
        )
        resp.raise_for_status()
        content = resp.json()["result"]
        self._access_token = content["accessToken"]
        self._refresh_token = content["refreshToken"]
        self._access_expires = time.time() + content["expireInSeconds"]
        self._refresh_expires = time.time() + content["refreshTokenExpireInSeconds"]

    @property
    def token(self):
        if time.time() > self._refresh_expires:
            raise Exception("Auth refresh token has expired. Unable to refresh access token.")

        if time.time() > self._access_expires:
            resp = self._session.get(
                f"{self._server}/api/TokenAuth/RefreshToken",
                params={"refreshToken": self._refresh_token},
            )
            resp.raise_for_status()
            content = resp.json()["result"]
            self._access_token = content["accessToken"]
            self._access_expires = time.time() + content["expireInSeconds"]

        return self._access_token

    def __call__(self, request):
        request.headers["Authorization"] = "Bearer " + self.token
        return request

    def __del__(self):
        # TODO: Add a logout request on auth object deletion
        pass

client

client

client("servers/virtual", config=None)

A shorthand for loading a specific client with a default session/config. Optionally, a Config object can be supplied as a keyword.

Source code in denvr/client.py

def client(name: str, conf: Config | None = None):
    """
    client("servers/virtual", config=None)

    A shorthand for loading a specific client with a default session/config.
    Optionally, a Config object can be supplied as a keyword.
    """
    _config = conf if conf else config()

    # TODO: Better vetting of `name` for cross-platform paths
    mod = importlib.import_module(
        "denvr.api.{}.{}".format(_config.api, ".".join(name.split("/")))
    )

    return mod.Client(Session(_config))

config

Config

Stores the auth and defaults.

Source code in denvr/config.py

class Config:
    """
    Stores the auth and defaults.
    """

    def __init__(self, defaults: dict, auth: Auth | None):
        self.defaults = defaults
        self.auth = auth

    @property
    def server(self):
        return self.defaults.get("server", "https://api.cloud.denvrdata.com")

    @property
    def api(self):
        return self.defaults.get("api", "v1")

    @property
    def cluster(self):
        return self.defaults.get("cluster", "Msc1")

    @property
    def tenant(self):
        return self.defaults.get("tenant", None)

    @property
    def vpcid(self):
        return self.defaults.get("vpcid", self.tenant)

    @property
    def rpool(self):
        return self.defaults.get("rpool", "on-demand")

    @property
    def retries(self):
        return self.defaults.get("retries", 3)

    def getkwarg(self, name, val):
        """
        Uses default value for the provided `name` if `val` is `None`.
        """
        if val is None:
            return getattr(self, name, None)

        return val

getkwarg

Uses default value for the provided name if val is None.

Source code in denvr/config.py

def getkwarg(self, name, val):
    """
    Uses default value for the provided `name` if `val` is `None`.
    """
    if val is None:
        return getattr(self, name, None)

    return val

config

Construct a Config object from the provide config file path.

Source code in denvr/config.py

def config(path=None):
    """
    Construct a Config object from the provide config file path.
    """
    config_path = path if path else os.getenv("DENVR_CONFIG", DEFAULT_CONFIG_PATH)
    config = toml.load(config_path) if os.path.exists(config_path) else {}
    defaults = config.get("defaults", {})
    credentials = config.get("credentials", {})
    server = defaults.get("server", "https://api.cloud.denvrdata.com")

    username = os.getenv("DENVR_USERNAME", credentials.get("username", ""))
    if not username:
        raise Exception('Could not find username in "DENVR_USERNAME" or ' + config_path)

    password = os.getenv("DENVR_PASSWORD", credentials.get("password", ""))
    if not password:
        raise Exception('Could not find password in "DENVR_PASSWORD" or ' + config_path)

    # NOTE: We're intentionally letting the loaded username/password go out of scope for security reasons.
    # The auth object should be able to handle everything from here onward.
    return Config(
        defaults=defaults, auth=Auth(server, username, password, defaults.get("retries", 3))
    )

session

Session

Session(config: Config)

Handles authentication and HTTP requests to Denvr's API.

Source code in denvr/session.py

class Session:
    """
    Session(config: Config)

    Handles authentication and HTTP requests to Denvr's API.
    """

    def __init__(self, config: Config):
        self.config = config
        self.session = requests.Session()

        # Set the auth, header and retry strategy for the session object
        self.session.auth = self.config.auth
        self.session.headers.update({"Content-Type": "application/json"})
        if self.config.retries:
            self.session.mount(
                self.config.server, HTTPAdapter(max_retries=retry(retries=self.config.retries))
            )

    def request(self, method, path, **kwargs):
        url = "/".join([self.config.server, *filter(None, path.split("/"))])
        logger.debug("Request: self.session.request(%s, %s, **%s", method, url, kwargs)
        resp = self.session.request(method, url, **kwargs)
        raise_for_status(resp)
        result = resp.json()
        logger.debug("Response: resp.json() -> %s", result)

        # According to the spec we should just be return result and not {"result": result }?
        # For mock-server testing purposes we'll support both.
        result = result.get("result", result) if isinstance(result, dict) else result

        # Standardize the response keys to snakecase if it's a dict'
        if isinstance(result, dict):
            return {snakecase(k): v for k, v in result.items()}

        return result

utils

raise_for_status

Given a response object return either resp.json() or resp.json()["error"]. This is basically just a modified version of https://requests.readthedocs.io/en/latest/_modules/requests/models/#Response.raise_for_status

Parameters:

Name	Type	Description	Default
resp	Response	The request response object.	required

Returns:

Type	Description
	The request response error.

Source code in denvr/utils.py

@typing.no_type_check
def raise_for_status(resp: Response):
    """
    Given a response object return either resp.json() or resp.json()["error"].
    This is basically just a modified version of
    https://requests.readthedocs.io/en/latest/_modules/requests/models/#Response.raise_for_status

    Args:
        resp (Response): The request response object.

    Returns:
        The request response error.
    """
    # Early exit if everything is fine.
    if resp.status_code < 400:
        return None

    # Start building the error message that we'll raise
    if isinstance(resp.reason, bytes):
        # We attempt to decode utf-8 first because some servers
        # choose to localize their reason strings. If the string
        # isn't utf-8, we fall back to iso-8859-1 for all other
        # encodings. (See PR #3538)
        try:
            reason = resp.reason.decode("utf-8")
        except UnicodeDecodeError:
            reason = resp.reason.decode("iso-8859-1")
    else:
        reason = resp.reason

    details = ""
    try:
        details = " - {}".format(resp.json()["error"]["message"])
    except JSONDecodeError:
        logger.debug("Failed to decode JSON response")
    except KeyError:
        logger.debug("Failed to extract error message from response")

    msg = ""
    if 400 <= resp.status_code < 500:
        msg = f"{resp.status_code} Client Error: {reason} for url: {resp.url}{details}"
    elif 500 <= resp.status_code < 600:
        msg = f"{resp.status_code} Server Error: {reason} for url: {resp.url}{details}"

    if msg:
        raise HTTPError(msg, response=resp)

retry

Generates a reasonable default Retry object for use with the requests library given a total number of retries.

NOTES

• by default only retry on idempotent requests (including DELETE and PUT)
• only retry for 5xx or 429 errors
• Allow redirects, but remove Authorization headers
• Leave error handling to the caller

Source code in denvr/utils.py

def retry(retries: int = 3, idempotent_only: bool = True):
    """
    Generates a reasonable default Retry object for use with the requests library
    given a total number of retries.

    NOTES:
        - by default only retry on idempotent requests (including DELETE and PUT)
        - only retry for 5xx or 429 errors
        - Allow redirects, but remove Authorization headers
        - Leave error handling to the caller
    """
    allowed_methods = ["GET", "HEAD", "PUT", "DELETE", "OPTIONS", "TRACE"]

    if not idempotent_only:
        allowed_methods.extend(["POST", "PATCH"])

    return Retry(
        total=retries,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=allowed_methods,
        respect_retry_after_header=True,
        remove_headers_on_redirect=["Authorization"],
        raise_on_redirect=False,
        raise_on_status=False,
    )

snakecase

Convert camelcase and titlecase strings to snakecase.

Parameters:

Name	Type	Description	Default
str	str	The string to convert.	required

Returns:

Name	Type	Description
str	str	The converted string.

Source code in denvr/utils.py

def snakecase(text: str) -> str:
    """
    Convert camelcase and titlecase strings to snakecase.

    Args:
        str (str): The string to convert.

    Returns:
        str: The converted string.
    """
    return "".join(["_" + i.lower() if i.isupper() else i for i in text]).lstrip("_")

validate

validate_kwargs

For None values in kwargs error if they are in required or drop them.

Source code in denvr/validate.py

def validate_kwargs(method, path, kwargs, required):
    """
    For `None` values in `kwargs` error if they are in `required` or drop them.
    """
    result: Dict[str, Dict] = {}
    for kw, args in kwargs.items():
        result[kw] = {}
        for k, v in args.items():
            if v is None:
                if k in required:
                    raise TypeError(
                        f"Required {kw} parameter {k} is missing for {method} request to {path}"
                    )

                logger.debug("Dropping missing %s argument %s", kw, k)
            elif kw == "params" and isinstance(v, bool):
                # Handle converting True to 'true' for boolean params
                result[kw][k] = json.dumps(v)
            else:
                result[kw][k] = v

    return result

waiters

Waiter

A utility class which waits on a check function to return True after executing an action function. For example, waiting for get_server to return status "ONLINE" after calling create_server.

Parameters:

Name	Type	Description	Default
action	callable	Function which takes kwargs, runs an operation and returns a response.	required
check	callable	Function which takes the action response and returns (bool, result) representing whether a check has passed and any results to return.	required
cleanup	callable	An optional function to run in failure conditions.	None

Source code in denvr/waiters.py

class Waiter:
    """
    A utility class which waits on a check function to return True after executing an action function.
    For example, waiting for `get_server` to return status "ONLINE" after calling `create_server`.

    Args:
        action (callable): Function which takes kwargs, runs an operation and returns a response.
        check (callable): Function which takes the action response and returns (bool, result) representing
            whether a check has passed and any results to return.
        cleanup (callable): An optional function to run in failure conditions.
    """

    def __init__(
        self, action: Callable, check: Callable, cleanup: Union[Callable, None] = None
    ):
        self.action = action
        self.check = check
        self.cleanup = cleanup

    def __call__(self, interval=30, timeout=600, **kwargs):
        resp = self.action(**kwargs)
        try:
            return self.wait(resp, interval, timeout)
        except Exception as e:
            if self.cleanup:
                self.cleanup(resp)
            raise e

    def wait(self, resp, interval=30, timeout=600):
        start_time = time.time()

        # Loop until check succeeds or timeout occurs
        while True:
            passes, result = self.check(resp)
            if passes:
                return result

            if time.time() - start_time > timeout:
                raise TimeoutError("Wait operation timed out")

            time.sleep(interval)

waiter

A waiter factory function that creates a Waiter instance for a given operation.

Example:

create_server = waiter(virtual.create_server)
create_server(name="my-test-vm", rpool="on-demand", vpc="denvr", ...)

Parameters:

Name	Type	Description	Default
operation	Callable	The operation to wait for.	required

Returns:

Type	Description
Waiter	A Waiter instance.

Raises:

Type	Description
ValueError	If the operation is not supported.

Source code in denvr/waiters.py

def waiter(operation: Callable) -> Waiter:
    """
    A waiter factory function that creates a Waiter instance for a given operation.

    Example:

        create_server = waiter(virtual.create_server)
        create_server(name="my-test-vm", rpool="on-demand", vpc="denvr", ...)

    Args:
        operation: The operation to wait for.

    Returns:
        A Waiter instance.

    Raises:
        ValueError: If the operation is not supported.
    """
    # NOTE: This function is a bit of a hack that could use a more declarative approach for describing waiter rules.
    # Arguably each service client should be responsible for this, but that would complicate the existing code generation.
    client = getattr(operation, "__self__", None)
    if client is None:
        raise ValueError(f"Operation must be a method of a client. Not {operation}.")

    class_name = getattr(client.__class__, "__name__", "")
    module_name = getattr(client, "__module__", "")
    method_name = getattr(operation, "__name__", "")

    if client and class_name == "Client" and module_name.startswith("denvr"):
        if module_name.endswith("virtual"):
            if method_name in ["create_server", "start_server"]:
                return Waiter(
                    action=operation, check=lambda resp: _vm_online_check(client, resp)
                )
            elif method_name == "stop_server":
                return Waiter(
                    action=operation, check=lambda resp: _vm_offline_check(client, resp)
                )
        elif module_name.endswith("applications"):
            if method_name in [
                "create_catalog_application",
                "create_custom_application",
                "start_application",
            ]:
                return Waiter(
                    action=operation, check=lambda resp: _app_online_check(client, resp)
                )
            elif method_name == "stop_application":
                return Waiter(
                    action=operation, check=lambda resp: _app_offline_check(client, resp)
                )

    # If we don't find a waiter configuration then raise a ValueError
    raise ValueError(f"Unsupported operation: {module_name}.{class_name}/{method_name}")