"""
@author Stefan Schnell <mail@stefan-schnell.de>
@license MIT
@version 1.0
"""
import base64
import http.client
import json
import ssl
import urllib.error
import urllib.parse
import urllib.request
class Http:
""" Handles HTTP requests
"""
def __init__(self):
pass
def __validateJson(jsonData: str) -> bool:
try:
json.loads(jsonData)
except ValueError as err:
return False
return True
def request(
self,
url: str,
user: str | None = None,
password: str | None = None,
bearerToken: str | None = None,
method: str = "GET",
header: list[list] = [],
body: dict | bytes | None = None,
contentType: str = "application/json;charset=utf-8",
accept: str = "application/json"
) -> dict | bytes | http.client.HTTPResponse | None:
""" Executes a REST request
@param {string} url - URL to execute the request
@param {string or None} user
@param {string or None} password
@param {string or None} bearerToken
@param {string} method - Method of request, e.g. GET, POST, etc
@param {list.<list>} header - Additional header entries
@param {dictionary or bytes or None} body - Body of request
@param {string} contentType - MIME type of the body for the request
@param {string} accept - MIME type of content expect/prefer response
@returns {dictionary or bytes or HTTPResponse or None}
"""
returnValue: dict | bytes | None = None
data: dict | bytes | None = None
if isinstance(body, dict):
data = bytes(json.dumps(body).encode("utf-8"))
elif isinstance(body, bytes):
data = body
else:
data = bytes(json.dumps({}).encode("utf-8"))
try:
request: urllib.request.Request = urllib.request.Request(
url = url,
method = method,
data = data
)
if user and password:
authorization: str = 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
)
if len(header) > 0:
for key, value in header:
request.add_header(
key, value
)
response: http.client.HTTPResponse = urllib.request.urlopen(
request,
context = ssl._create_unverified_context()
)
responseCode: int = response.status
responseRead: bytes = response.read()
if (
responseCode == 200 or
responseCode == 201 or
responseCode == 202 or
responseCode == 204
):
if len(responseRead) > 0:
if "json" in accept:
returnValue = json.loads(responseRead)
else:
returnValue = responseRead
else:
returnValue = response
except urllib.error.HTTPError as err:
if __validateJson(err.read().decode()):
error = json.loads(err.read().decode())
status = error["status"]
message = error["message"]
raise Exception(
f"An HTTP error occurred at request - {status}: {message}"
) from err
else:
raise Exception(
f"An error occurred at request - {err}"
) from err
except Exception as err:
raise Exception(
f"An error occurred at request - {err}"
) from err
return returnValue
|