"""
@module de.stschnell
@version 0.1.0
@runtime python:3.10
@outputType Properties
"""
import base64
import json
import ssl
import http.client
import urllib.request
def httpRequest(
url: str,
user: str = None,
password: str = None,
bearerToken: str = None,
method: str = "GET",
body: dict = {},
contentType: str = "application/json;charset=utf-8",
accept: str = "application/json"
):
""" Executes a REST request
@param {string} url - URL to execute the request
@param {string} user
@param {string} password
@param {string} bearerToken
@param {string} method - Method of request, e.g. GET, POST, etc
@param {dictionary} body - Body of request
@param {string} contentType - MIME type of the body for the request
@param {string} accept - MIME type of response content
@returns {dictionary | bytes}
"""
returnValue: dict | bytes = {}
try:
request = urllib.request.Request(
url = url,
method = method,
data = bytes(json.dumps(body).encode("utf-8"))
)
if user and password:
authorization = base64.b64encode(
bytes(user + ":" + password, "UTF-8")
).decode("UTF-8")
request.add_header(
"Authorization", "Basic " + authorization
)
if bearerToken:
request.add_header(
"Authorization", "Bearer " + bearerToken
)
request.add_header(
"Content-Type", contentType
)
request.add_header(
"Accept", accept
)
response = urllib.request.urlopen(
request,
context = ssl._create_unverified_context()
)
responseCode: int = response.status
responseRead: bytes = response.read()
if responseCode == 200:
if len(responseRead) > 0:
if "json" in accept:
returnValue = json.loads(responseRead)
else:
returnValue = responseRead
except Exception as err:
raise Exception(f"An error occurred at request - {err}") \
from err
finally:
if response:
response.close()
return returnValue
def getResource(
vcoUrl: str,
bearerToken: str,
resourceFolder: str,
resourceName: str,
mimeType: str
) -> dict | bytes | None:
""" Gets a resource
@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 {dictionary | bytes | None}
"""
returnValue: dict | bytes | None = None
try:
resources: dict = httpRequest(
url = vcoUrl + "/api/resources",
bearerToken = bearerToken
)
found: bool = False
resource: dict
for resource in resources["link"]:
id: str | None = None
attribute: dict
for attribute in resource["attributes"]:
if attribute["name"] == "name" and \
attribute["value"] == resourceName:
found = True
if attribute["name"] == "id":
id = attribute["value"]
if found == True and id != None:
categoryId: str = httpRequest(
url = vcoUrl + "/api/resources/" + id,
bearerToken = bearerToken,
accept = (
"application/vnd.o11n.resource.metadata+json;"
"charset=UTF-8"
)
)["category-id"]
categoryPath: str = httpRequest(
url = vcoUrl + "/api/categories/" + categoryId,
bearerToken = bearerToken
)["path"]
if categoryPath != resourceFolder:
found = False
if found:
break
if id != None:
returnValue = httpRequest(
url = vcoUrl + "/api/resources/" + id,
bearerToken = bearerToken,
accept = mimeType
)
except Exception as err:
raise ValueError(
f"An error occurred at get resource id - {err}"
) from err
return returnValue
def handler(context: dict, inputs: dict) -> dict:
outputs: dict | None = None
vcoUrl: str = context["vcoUrl"]
bearerToken: str = context["getToken"]()
resourceFolder: str = "Library/Dynamic Types/images"
resourceName: str = "folder-16x16.png"
mimeType: str = "image/png"
# resourceFolder: str = "Library/VC/Configuration"
# resourceName: str = "b56969bb-eef7-459c-aad1-13d23ece2f97"
# mimeType: str = "text/xml"
# resourceFolder: str = "Library/VC/Configuration"
# resourceName: str = "properties.json"
# mimeType: str = "application/json"
resource: dict | bytes | None = None
output: str = ""
try:
resource = getResource(
vcoUrl,
bearerToken,
resourceFolder,
resourceName,
mimeType
)
if resource is not None:
if isinstance(resource, bytes):
try:
output = resource.decode("utf-8")
except UnicodeDecodeError:
output = base64.b64encode(resource).decode("utf-8")
elif isinstance(resource, dict):
output = json.dumps(resource)
else:
raise FileNotFoundError("Resource not found")
outputs = {
"status": "done",
"error": None,
"result": output
}
except Exception as err:
outputs = {
"status": "incomplete",
"error": repr(err),
"result": output
}
return outputs
|
