Automation Blog

from Stefan Schnell

A resource element is a central storage location for files, such as documents, templates or binary data, that can be accessed across the entire Orchestrator server. Instead of embedding files directly into individual actions or workflows, it is possible to store them centrally as a resource element. In this way, if a file changes, it is only necessary to update it in this central location. This post presents a class that simplifies the handling of resource elements in Python.

Resource Element Class for Python Runtime Environment

This class contains several methods, including the basic operations for creating, reading, updating and deleting (CRUD) of resource elements. The update method is split up into two methods, one to update the metadata and one to update the content of a resource element.

Hint: In everyday language, the resource element is often called simply a "resource". This can lead to confusion, because there are several other types of resources within the Orchestrator environment, such as cloud resources, infrastructure resources or deployment resources. 

"""
@author Stefan Schnell <mail@stefan-schnell.de>
@license MIT
@version 1.0
"""

import json
import random

from .http import Http
from .category import Category

class Resource:
    """ Handles VCF Automation resource elements
    """

    def __init__(
        self
    ) -> None:
        self.__http = Http()
        self.__category = Category()

    def __getId(
        self,
        vcoUrl: str,
        bearerToken: str,
        resourceFolder: str,
        resourceName: str
    ) -> str | None:
        """ Gets the ID of a resource element

        @param {string} vcoUrl - URL of Aria orchestrator
        @param {string} bearerToken
        @param {string} resourceFolder - Path of the resource
        @param {string} resourceName - Name of the resource
        @returns {string or none} ID of the resource
        """

        resourceId: str | None = None

        try:

            resources = self.__http.request(
                url = vcoUrl + "/api/resources",
                bearerToken = bearerToken
            )

            found: bool = False
            for resource in resources["link"]:
                resourceId = None
                for attribute in resource["attributes"]:
                    if attribute["name"] == "name" and \
                    attribute["value"] == resourceName:
                        found = True
                    if attribute["name"] == "id":
                        resourceId = attribute["value"]

                if found is True and resourceId is not None:

                    categoryId = self.__http.request(
                        url = f"{vcoUrl}/api/resources/{resourceId}",
                        bearerToken = bearerToken,
                        accept = (
                            "application/vnd.o11n.resource.metadata+json;"
                            "charset=UTF-8"
                        )
                    )["category-id"]

                    categoryPath = self.__category.getPath(
                        vcoUrl = vcoUrl,
                        bearerToken = bearerToken,
                        categoryId = categoryId
                    )

                    if categoryPath != resourceFolder:
                        found = False

                if found is False:
                    resourceId = None

                if found:
                    break

        except Exception as err:
            raise ValueError(
                f"An error occurred at get resource id - {err}"
            ) from err

        return resourceId

    def create(
        self,
        vcoUrl: str,
        bearerToken: str,
        resourceFolder: str,
        resourceName: str,
        mimeType: str,
        resourceContent: str
    ) -> str | None:
        """ Creates a resource element

        @param {string} vcoUrl - URL of Aria orchestrator
        @param {string} bearerToken
        @param {string} resourceFolder - Path of the resource
        @param {string} resourceName - Name of the resource
        @param {string} mimeType - MIME type of the resource
        @param {string} resourceContent - Content of the resource
        @returns {string or none} ID of the resource
        """

        returnValue: str | None = None

        try:

            categoryId: str = self.__category.getId(
                vcoUrl,
                bearerToken,
                "ResourceElementCategory",
                resourceFolder
            )

            if categoryId:

                boundary: str = "-----" + str(random.randint(10000, 99999))

                body = []
                body.append(b"--" + str.encode(boundary))
                body.append(
                    b"Content-Disposition: form-data; name=\"categoryId\""
                )
                body.append(b"")
                body.append(str.encode(categoryId))
                body.append(b"--" + str.encode(boundary))
                body.append(
                    (
                        f"Content-Disposition: form-data; "
                        f"name=\"file\"; filename=\"{resourceName}\""
                    ).encode("utf-8")
                )
                body.append(b"Content-Type: " + str.encode(mimeType))
                body.append(b"")
                lines = resourceContent.split('\n')
                for line in lines:
                    body.append(str.encode(line))
                body.append(b"--" + str.encode(boundary) + b"--")
                body.append(b"")
                payload = b"\r\n".join(body)

                response = self.__http.request(
                    url = f"{vcoUrl}/api/resources",
                    bearerToken = bearerToken,
                    method = "POST",
                    body = payload,
                    header = [
                        [ "Content-Length", str(len(payload)) ]
                    ],
                    contentType = (
                        f"multipart/form-data; boundary={boundary}"
                    ),
                    accept = "*/*"
                )

                returnValue = response.headers["Location"].split("/")[-1]

        except Exception as err:
            raise ValueError(
                f"An error occurred at create resource - {err}"
            ) from err

        return returnValue

    def read(
        self,
        vcoUrl: str,
        bearerToken: str,
        resourceFolder: str,
        resourceName: str,
        mimeType: str = "application/vnd.o11n.resource.metadata+json"
    ) -> str | dict | None:
        """ Reads a resource element

        @param {string} vcoUrl - URL of Aria orchestrator
        @param {string} bearerToken
        @param {string} resourceFolder - Path of the resource
        @param {string} resourceName - Name of the resource
        @param {string} mimeType - MIME type of the resource
        @returns {string or dictionary or none} Return type depends from
                                                mimeType
        """

        returnValue: str | dict | None = None

        try:

            resourceId: str | None = self.__getId(
              vcoUrl = vcoUrl,
              bearerToken = bearerToken,
              resourceFolder = resourceFolder,
              resourceName = resourceName
            )

            if resourceId is not None:

                returnValue = self.__http.request(
                    url = f"{vcoUrl}/api/resources/{resourceId}",
                    bearerToken = bearerToken,
                    accept = mimeType
                )

        except Exception as err:
            raise ValueError(
                f"An error occurred at read resource - {err}"
            ) from err

        return returnValue

    def updateMetadata(
        self,
        vcoUrl: str,
        bearerToken: str,
        resourceFolder: str,
        resourceName: str,
        newResourceName: str,
        newResourceDescription: str,
        newResourceVersion: str
    ) -> None:
        """ Updates the meta data of a resource element

        @param {string} vcoUrl - URL of Aria orchestrator
        @param {string} bearerToken
        @param {string} resourceFolder - Path of the resource
        @param {string} resourceName - Name of the resource
        @param {string} newResourceName - New name of the resource
        @param {string} newResourceDescription - New description of the
                                                 resource
        @param {string} newResourceVersion - New version of the resource
        """

        try:

            resourceId: str | None = self.__getId(
              vcoUrl = vcoUrl,
              bearerToken = bearerToken,
              resourceFolder = resourceFolder,
              resourceName = resourceName
            )

            categoryId: str = self.__category.getId(
                vcoUrl,
                bearerToken,
                "ResourceElementCategory",
                resourceFolder
            )

            if categoryId and resourceId is not None:

                body: dict = {
                    "name": newResourceName,
                    "description": newResourceDescription,
                    "version": newResourceVersion,
                    "category-id": categoryId
                }

                self.__http.request(
                    url = f"{vcoUrl}/api/resources/{resourceId}",
                    bearerToken = bearerToken,
                    method = "PUT",
                    body = body,
                    accept = "*/*"
                )

        except Exception as err:
            raise ValueError(
                f"An error occurred at update resource metadata - {err}"
            ) from err

    def updateContent(
        self,
        vcoUrl: str,
        bearerToken: str,
        resourceFolder: str,
        resourceName: str,
        newResourceMimeType: str,
        newResourceContent: str
    ) -> None:
        """ Updates the content of a resource element

        @param {string} vcoUrl - URL of Aria orchestrator
        @param {string} bearerToken
        @param {string} resourceFolder - Path of the resource
        @param {string} resourceName - Name of the resource
        @param {string} newResourceMimeType - New mime type of the resource
        @param {string} newResourceContent - New content of the resource
        """

        try:

            resourceId: str | None = self.__getId(
              vcoUrl = vcoUrl,
              bearerToken = bearerToken,
              resourceFolder = resourceFolder,
              resourceName = resourceName
            )

            categoryId: str = self.__category.getId(
                vcoUrl,
                bearerToken,
                "ResourceElementCategory",
                resourceFolder
            )

            if categoryId and resourceId is not None:

                boundary: str = "-----" + str(random.randint(10000, 99999))

                body = []
                body.append(b"--" + str.encode(boundary))
                body.append(
                    b"Content-Disposition: form-data; name=\"file\"; "
                    b"filename=\"" + str.encode(resourceName) + b"\""
                )
                body.append(
                    b"Content-Type: " + str.encode(newResourceMimeType)
                )
                body.append(b"")
                lines = newResourceContent.split('\n')
                for line in lines:
                    body.append(str.encode(line))
                body.append(b"--" + str.encode(boundary) + b"--")
                body.append(b"")
                payload = b"\r\n".join(body)

                response = self.__http.request(
                    url = f"{vcoUrl}/api/resources/{resourceId}",
                    bearerToken = bearerToken,
                    method = "POST",
                    body = payload,
                    header = [
                        [ "Content-Length", str(len(payload)) ]
                    ],
                    contentType = (
                        f"multipart/form-data; boundary={boundary}"
                    ),
                    accept = "*/*"
                )

        except Exception as err:
            raise ValueError(
                f"An error occurred at update resource content - {err}"
            ) from err

    def delete(
        self,
        vcoUrl: str,
        bearerToken: str,
        resourceFolder: str,
        resourceName: str,
        force: bool = False
    ) -> None:
        """ Deletes a resource element

        @param {string} vcoUrl - URL of Aria orchestrator
        @param {string} bearerToken
        @param {string} resourceFolder - Path of the resource
        @param {string} resourceName - Name of the resource
        @param {boolean} force
        """

        try:

            resourceId: str | None = self.__getId(
              vcoUrl = vcoUrl,
              bearerToken = bearerToken,
              resourceFolder = resourceFolder,
              resourceName = resourceName
            )

            if resourceId is not None:

                url: str = f"{vcoUrl}/api/resources/{resourceId}"

                if force is True:
                    url += "?force=true"
                elif force is False:
                    url += "?force=false"

                self.__http.request(
                    url = url,
                    bearerToken = bearerToken,
                    method = "DELETE",
                    accept = "*/*"
                )

        except Exception as err:
            raise ValueError(
                f"An error occurred at delete resource - {err}"
            ) from err

Conclusion

This Python class greatly simplifies the handling of resource elements and it can be used as an integration basis in VCF Automation. It can also be used in other environments that support Python.

References