"""
@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 Configuration:
""" Handles VCF Automation configuration elements
"""
def __init__(
self
) -> None:
self.__http = Http()
self.__category = Category()
def __getId(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str
) -> str | None:
""" Gets the ID of a configuration element
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@returns {string or none} ID of the configuration
"""
configurationId: str | None = None
try:
configurations = self.__http.request(
url = f"{vcoUrl}/api/configurations",
bearerToken = bearerToken
)
found: bool = False
for configuration in configurations["link"]:
configurationId = None
for attribute in configuration["attributes"]:
if attribute["name"] == "name" and \
attribute["value"] == configurationName:
found = True
if attribute["name"] == "id":
configurationId = attribute["value"]
if found is True and configurationId is not None:
categoryId = self.__http.request(
url = (
f"{vcoUrl}/api/configurations/{configurationId}"
),
bearerToken = bearerToken
)["category-id"]
categoryPath = self.__category.getPath(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
categoryId = categoryId
)
if categoryPath != configurationFolder:
found = False
if found is False:
configurationId = None
if found:
break
except Exception as err:
raise ValueError(
f"An error occurred at get configuration id - {err}"
) from err
return configurationId
def create(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str = "untitled configuration " + \
str(random.randint(10000000, 99999999)),
configurationDescription: str = "",
configurationVersion: str = "0.0.0"
) -> str | None:
""" Creates a configuration element
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@param {string} configurationDescription - Description of the
configuration
@param {string} configurationVersion - Version of the configuration
@returns {string or none} ID of the configuration
"""
returnValue: str | None = None
try:
categoryId: str = self.__category.getId(
vcoUrl,
bearerToken,
"ConfigurationElementCategory",
configurationFolder
)
if categoryId:
body: dict = {
"name": configurationName,
"description": configurationDescription,
"version": configurationVersion,
"category-id": categoryId
}
url: str = f"{vcoUrl}/api/configurations?uniqueName=false"
returnValue = self.__http.request(
url = url,
bearerToken = bearerToken,
method = "POST",
body = body
)["id"]
except Exception as err:
raise ValueError(
f"An error occurred at create configuration - {err}"
) from err
return returnValue
def read(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str
) -> dict | None:
""" Reads a configuration element
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@returns {dictionary or None}
"""
returnValue: dict | None = None
try:
configurationId: str | None = self.__getId(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
configurationFolder = configurationFolder,
configurationName = configurationName
)
if configurationId is not None:
returnValue = self.__http.request(
url = (
f"{vcoUrl}/api/configurations/{configurationId}"
),
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at read configuration - {err}"
) from err
return returnValue
def update(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str,
newConfigurationName: str,
newConfigurationDescription: str,
newConfigurationVersion: str
) -> None:
""" Updates a configuration element
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@param {string} newConfigurationName - New name of the configuration
@param {string} newConfigurationDescription - New description of the
configuration
@param {string} newConfigurationVersion - New version of the
configuration
"""
try:
configurationId: str | None = self.__getId(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
configurationFolder = configurationFolder,
configurationName = configurationName
)
if configurationId is not None:
body: dict = {
"name": newConfigurationName,
"description": newConfigurationDescription,
"version": newConfigurationVersion
}
self.__http.request(
url = (
f"{vcoUrl}/api/configurations/{configurationId}"
),
bearerToken = bearerToken,
method = "PUT",
body = body,
accept = "*/*"
)
except Exception as err:
raise ValueError(
f"An error occurred at update configuration - {err}"
) from err
def delete(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str,
force: bool = False
) -> None:
""" Deletes a configuration element
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@param {boolean} force
"""
try:
configurationId: str | None = self.__getId(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
configurationFolder = configurationFolder,
configurationName = configurationName
)
if configurationId is not None:
url: str = (
f"{vcoUrl}/api/configurations/{configurationId}"
)
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 configuration - {err}"
) from err
def createVariable(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str,
configurationVariableName: str,
configurationVariableType: str,
configurationVariableValue: bool | float | int | str
) -> None:
""" Creates a variable in a configuration element
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@param {string} configurationVariableName - Name of the variable
@param {string} configurationVariableType - Type of the variable
@param {boolean | number | string} configurationVariableValue -
Value of the variable
"""
try:
configurationId: str | None = self.__getId(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
configurationFolder = configurationFolder,
configurationName = configurationName
)
if configurationId is not None:
config: dict | None = self.__http.request(
url = (
f"{vcoUrl}/api/configurations/{configurationId}"
),
bearerToken = bearerToken
)
if config is not None:
attributes: dict = config["attributes"]
found: bool = False
for attribute in attributes:
if attribute["name"] == configurationVariableName:
found = True
break
if found is False:
attributes.append({
"value": { configurationVariableType:
{"value": configurationVariableValue}
},
"type": configurationVariableType,
"name": configurationVariableName
})
body: dict = {
"name": configurationName,
"version": config["version"],
"attributes": attributes
}
self.__http.request(
url = (
f"{vcoUrl}/api/configurations/"
f"{configurationId}"
),
bearerToken = bearerToken,
method = "PUT",
body = body,
contentType = "application/json",
accept = "*/*"
)
except Exception as err:
raise ValueError(
f"An error occurred at create variable - {err}"
) from err
def readVariable(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str,
configurationVariableName: str
) -> dict:
""" Reads a value of a variable from a configuration element
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@param {string} configurationVariableName - Name of the variable
@returns {dictionary or None}
"""
returnValue: dict | None = None
try:
config: dict | None = self.read(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
configurationFolder = configurationFolder,
configurationName = configurationName
)
if config is not None:
for attribute in config["attributes"]:
if attribute["name"] == configurationVariableName:
returnValue = attribute["value"][attribute["type"]]
break
except Exception as err:
raise ValueError(
f"An error occurred at read variable value - {err}"
) from err
return returnValue
def updateVariable(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str,
configurationVariableName: str,
configurationVariableValue: str
) -> None:
""" Updates a variable in a configuration element
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@param {string} configurationVariableName - Name of the variable
@param {string} configurationVariableValue - New value of the variable
"""
try:
configurationId: str | None = self.__getId(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
configurationFolder = configurationFolder,
configurationName = configurationName
)
if configurationId is not None:
config: dict | None = self.__http.request(
url = f"{vcoUrl}/api/configurations/{configurationId}",
bearerToken = bearerToken
)
if config is not None:
attributes: dict = config["attributes"]
for attribute in attributes:
if attribute["name"] == configurationVariableName:
attribute["value"][attribute["type"]]["value"] = \
configurationVariableValue
body: dict = {
"name": configurationName,
"version": config["version"],
"attributes": attributes
}
self.__http.request(
url = (
f"{vcoUrl}/api/configurations/"
f"{configurationId}"
),
bearerToken = bearerToken,
method = "PUT",
body = body,
contentType = "application/json",
accept = "*/*"
)
break
except Exception as err:
raise ValueError(
f"An error occurred at update variable - {err}"
) from err
def deleteVariable(
self,
vcoUrl: str,
bearerToken: str,
configurationFolder: str,
configurationName: str,
configurationVariableName: str
) -> None:
""" Deletes a variable in a configuration
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} configurationFolder - Path of the configuration
@param {string} configurationName - Name of the configuration
@param {string} configurationVariableName - Name of the variable
"""
try:
configurationId: str | None = self.__getId(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
configurationFolder = configurationFolder,
configurationName = configurationName
)
if configurationId is not None:
config: dict | None = self.__http.request(
url = (
f"{vcoUrl}/api/configurations/{configurationId}"
),
bearerToken = bearerToken
)
if config is not None:
attributes: dict = config["attributes"]
found: bool = False
i: int = 0
for attribute in attributes:
if attribute["name"] == configurationVariableName:
found = True
del attributes[i]
break
i += 1
if found is True:
body: dict = {
"name": configurationName,
"version": config["version"],
"attributes": attributes
}
self.__http.request(
url = (
f"{vcoUrl}/api/configurations/"
f"{configurationId}"
),
bearerToken = bearerToken,
method = "PUT",
body = body,
contentType = "application/json",
accept = "*/*"
)
except Exception as err:
raise ValueError(
f"An error occurred at update variable - {err}"
) from err
|