189 lines
3.5 KiB
Python
189 lines
3.5 KiB
Python
import typing as _typing
|
|
import os as _os
|
|
import json as _json
|
|
import time as _time
|
|
import requests as _requests
|
|
|
|
from helpers import *
|
|
from conf import *
|
|
from log import *
|
|
|
|
|
|
|
|
def backend_api_call_generic(
|
|
session_key : _typing.Optional[str],
|
|
http_method : str,
|
|
action_path : str,
|
|
data,
|
|
options : _typing.Optional[dict] = None
|
|
):
|
|
options = (
|
|
{
|
|
}
|
|
|
|
|
(options or {})
|
|
)
|
|
log_info(
|
|
"backend_api_call",
|
|
{
|
|
"with_session_key": (not (session_key is None)),
|
|
"http_method": http_method,
|
|
"path": action_path,
|
|
}
|
|
)
|
|
target = string_coin(
|
|
"{{scheme}}://{{host}}:{{port}}{{path_base}}{{path_action}}",
|
|
{
|
|
"scheme": conf_get()["api"]["scheme"],
|
|
"host": conf_get()["api"]["host"],
|
|
"port": ("%u" % conf_get()["api"]["port"]),
|
|
"path_base": conf_get()["api"]["path"],
|
|
"path_action": action_path,
|
|
}
|
|
)
|
|
headers_common = (
|
|
{}
|
|
if (session_key is None) else
|
|
{"X-Session-Key": session_key}
|
|
)
|
|
if (http_method == "GET"):
|
|
response_raw = _requests.get(
|
|
target,
|
|
headers = (
|
|
headers_common
|
|
)
|
|
)
|
|
elif (http_method == "POST"):
|
|
response_raw = _requests.post(
|
|
target,
|
|
headers = (
|
|
headers_common
|
|
|
|
|
{
|
|
"Content-Type": "application/json",
|
|
}
|
|
),
|
|
json = data
|
|
)
|
|
elif (http_method == "DELETE"):
|
|
response_raw = _requests.delete(
|
|
target,
|
|
headers = (
|
|
headers_common
|
|
),
|
|
json = data
|
|
)
|
|
else:
|
|
raise NotImplementedError("unhandled HTTP method: %s" % http_method)
|
|
if ((response_raw.status_code < 200) or (response_raw.status_code >= 300)):
|
|
raise ValueError("irregular HTTP response status code: %u" % response_raw.status_code)
|
|
else:
|
|
return _json.loads(response_raw.text)
|
|
|
|
|
|
def backend_api_call_session_begin(
|
|
):
|
|
return backend_api_call_generic(
|
|
None,
|
|
"POST",
|
|
"/session/begin",
|
|
{
|
|
"name": conf_get()["account"]["name"],
|
|
"password": conf_get()["account"]["password"],
|
|
}
|
|
)
|
|
|
|
|
|
def get_session_key(
|
|
):
|
|
path = conf_get()["session"]["key_path"]
|
|
if (
|
|
_os.path.exists(path)
|
|
and
|
|
(
|
|
(_time.time() - _os.path.getmtime(path))
|
|
<
|
|
conf_get()["session"]["lifetime"]
|
|
)
|
|
):
|
|
session_key = file_text_read(path)
|
|
else:
|
|
session_key = backend_api_call_session_begin()
|
|
file_text_write(path, session_key)
|
|
return session_key
|
|
|
|
|
|
def backend_api_call_wrapped(
|
|
needs_session : bool,
|
|
http_method : str,
|
|
action_path : str,
|
|
data
|
|
):
|
|
return backend_api_call_generic(
|
|
(get_session_key() if needs_session else None),
|
|
http_method,
|
|
action_path,
|
|
data
|
|
)
|
|
|
|
|
|
def backend_api_call_session_end(
|
|
):
|
|
return backend_api_call_generic(
|
|
True,
|
|
"DELETE",
|
|
"/session/end",
|
|
None
|
|
)
|
|
|
|
|
|
def backend_api_call_member_list(
|
|
):
|
|
return backend_api_call_wrapped(
|
|
True,
|
|
"GET",
|
|
"/member/list",
|
|
None
|
|
)
|
|
|
|
|
|
def backend_api_call_member_read(
|
|
member_id : int
|
|
):
|
|
return backend_api_call_wrapped(
|
|
True,
|
|
"GET",
|
|
string_coin("/member/read/{{id}}", {"id": "%u" % member_id}),
|
|
None
|
|
)
|
|
|
|
|
|
def backend_api_call_member_project(
|
|
membership_number : _typing.Optional[str],
|
|
name_real_value : str,
|
|
email_address_private : _typing.Optional[str],
|
|
notification_target_url_template : _typing.Optional[str]
|
|
):
|
|
return backend_api_call_wrapped(
|
|
True,
|
|
"POST",
|
|
"/member/project",
|
|
{
|
|
"membership_number": membership_number,
|
|
"name_real_value": name_real_value,
|
|
"email_address_private": email_address_private,
|
|
"notification_target_url_template": notification_target_url_template,
|
|
}
|
|
)
|
|
|
|
|
|
def backend_api_call_member_delete(
|
|
member_id : int
|
|
):
|
|
return backend_api_call_wrapped(
|
|
True,
|
|
"DELETE",
|
|
string_coin("/member/delete/{{id}}", {"id": ("%u" % member_id)}),
|
|
None
|
|
)
|
|
|