frontend-mondvogel/source/backend.py

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],
	groups : list[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,
			"groups": groups,
			"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
	)