refactor: prepare backend refactor
This commit is contained in:
513
backend-legacy/server.py
Normal file
513
backend-legacy/server.py
Normal file
@@ -0,0 +1,513 @@
|
||||
from flask import Flask
|
||||
from flask import request
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable, ParamSpec, TypeVar, Generic
|
||||
|
||||
import config
|
||||
import database
|
||||
import utils
|
||||
from logger import LOGGER
|
||||
from database import ResponseBody
|
||||
|
||||
app = Flask(__name__)
|
||||
calendar_db = database.CalendarDatabase()
|
||||
|
||||
# region: API Route
|
||||
|
||||
# region: Common
|
||||
|
||||
|
||||
@app.route("/common/salt", methods=["POST"])
|
||||
def api_common_saltHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.common_salt, (FormField("username", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
@app.route("/common/login", methods=["POST"])
|
||||
def api_common_loginHandle():
|
||||
clientInfo = FetchClientNetworkInfo()
|
||||
|
||||
return SmartDbCaller(
|
||||
calendar_db.common_login,
|
||||
(
|
||||
FormField("username", str, False),
|
||||
FormField("password", str, False),
|
||||
FormField("clientUa", str, False),
|
||||
FormField("clientIp", str, False),
|
||||
),
|
||||
{"clientUa": clientInfo.user_agent, "clientIp": clientInfo.ip_addr},
|
||||
)
|
||||
|
||||
|
||||
@app.route("/common/webLogin", methods=["POST"])
|
||||
def api_common_webLoginHandle():
|
||||
clientInfo = FetchClientNetworkInfo()
|
||||
|
||||
return SmartDbCaller(
|
||||
calendar_db.common_webLogin,
|
||||
(
|
||||
FormField("username", str, False),
|
||||
FormField("password", str, False),
|
||||
FormField("clientUa", str, False),
|
||||
FormField("clientIp", str, False),
|
||||
),
|
||||
{"clientUa": clientInfo.user_agent, "clientIp": clientInfo.ip_addr},
|
||||
)
|
||||
|
||||
|
||||
@app.route("/common/logout", methods=["POST"])
|
||||
def api_common_logoutHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.common_logout, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
@app.route("/common/tokenValid", methods=["POST"])
|
||||
def api_common_tokenValidHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.common_tokenValid, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
# region: Calendar
|
||||
|
||||
|
||||
@app.route("/calendar/getFull", methods=["POST"])
|
||||
def api_calendar_getFullHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.calendar_getFull,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("startDateTime", int, False),
|
||||
FormField("endDateTime", int, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/calendar/getList", methods=["POST"])
|
||||
def api_calendar_getListHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.calendar_getList,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("startDateTime", int, False),
|
||||
FormField("endDateTime", int, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/calendar/getDetail", methods=["POST"])
|
||||
def api_calendar_getDetailHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.calendar_getDetail,
|
||||
(FormField("token", str, False), FormField("uuid", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/calendar/update", methods=["POST"])
|
||||
def api_calendar_updateHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.calendar_update,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("uuid", str, False),
|
||||
FormField("belongTo", str, True),
|
||||
FormField("title", str, True),
|
||||
FormField("description", str, True),
|
||||
FormField("eventDateTimeStart", int, True),
|
||||
FormField("eventDateTimeEnd", int, True),
|
||||
FormField("loopRules", str, True),
|
||||
FormField("timezoneOffset", int, True),
|
||||
FormField("lastChange", str, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/calendar/add", methods=["POST"])
|
||||
def api_calendar_addHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.calendar_add,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("belongTo", str, False),
|
||||
FormField("title", str, False),
|
||||
FormField("description", str, False),
|
||||
FormField("eventDateTimeStart", int, False),
|
||||
FormField("eventDateTimeEnd", int, False),
|
||||
FormField("loopRules", str, False),
|
||||
FormField("timezoneOffset", int, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/calendar/delete", methods=["POST"])
|
||||
def api_calendar_deleteHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.calendar_delete,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("uuid", str, False),
|
||||
FormField("lastChange", str, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
# region: Collection
|
||||
|
||||
|
||||
@app.route("/collection/getFullOwn", methods=["POST"])
|
||||
def api_collection_getFullOwnHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_getFullOwn, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/getListOwn", methods=["POST"])
|
||||
def api_collection_getListOwnHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_getListOwn, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/getDetailOwn", methods=["POST"])
|
||||
def api_collection_getDetailOwnHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_getDetailOwn,
|
||||
(FormField("token", str, False), FormField("uuid", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/addOwn", methods=["POST"])
|
||||
def api_collection_addOwnHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_addOwn,
|
||||
(FormField("token", str, False), FormField("name", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/updateOwn", methods=["POST"])
|
||||
def api_collection_updateOwnHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_updateOwn,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("uuid", str, False),
|
||||
FormField("name", str, False),
|
||||
FormField("lastChange", str, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/deleteOwn", methods=["POST"])
|
||||
def api_collection_deleteOwnHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_deleteOwn,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("uuid", str, False),
|
||||
FormField("lastChange", str, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/getSharing", methods=["POST"])
|
||||
def api_collection_getSharingHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_getSharing,
|
||||
(FormField("token", str, False), FormField("uuid", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/deleteSharing", methods=["POST"])
|
||||
def api_collection_deleteSharingHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_deleteSharing,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("uuid", str, False),
|
||||
FormField("target", str, False),
|
||||
FormField("lastChange", str, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/addSharing", methods=["POST"])
|
||||
def api_collection_addSharingHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_addSharing,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("uuid", str, False),
|
||||
FormField("target", str, False),
|
||||
FormField("lastChange", str, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/collection/getShared", methods=["POST"])
|
||||
def api_collection_getSharedHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.collection_getShared, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
# region: Todo
|
||||
|
||||
|
||||
@app.route("/todo/getFull", methods=["POST"])
|
||||
def api_todo_getFullHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.todo_getFull, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
@app.route("/todo/getList", methods=["POST"])
|
||||
def api_todo_getListHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.todo_getList, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
@app.route("/todo/getDetail", methods=["POST"])
|
||||
def api_todo_getDetailHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.todo_getDetail,
|
||||
(FormField("token", str, False), FormField("uuid", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/todo/add", methods=["POST"])
|
||||
def api_todo_addHandle():
|
||||
return SmartDbCaller(calendar_db.todo_add, (FormField("token", str, False),), None)
|
||||
|
||||
|
||||
@app.route("/todo/update", methods=["POST"])
|
||||
def api_todo_updateHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.todo_update,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("uuid", str, False),
|
||||
FormField("data", str, False),
|
||||
FormField("lastChange", str, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/todo/delete", methods=["POST"])
|
||||
def api_todo_deleteHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.todo_delete,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("uuid", str, False),
|
||||
FormField("lastChange", str, False),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
# region: Admin
|
||||
|
||||
|
||||
@app.route("/admin/get", methods=["POST"])
|
||||
def api_admin_getHandle():
|
||||
return SmartDbCaller(calendar_db.admin_get, (FormField("token", str, False),), None)
|
||||
|
||||
|
||||
@app.route("/admin/add", methods=["POST"])
|
||||
def api_admin_addHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.admin_add,
|
||||
(FormField("token", str, False), FormField("username", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/admin/update", methods=["POST"])
|
||||
def api_admin_updateHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.admin_update,
|
||||
(
|
||||
FormField("token", str, False),
|
||||
FormField("username", str, False),
|
||||
FormField("password", str, True),
|
||||
FormField("isAdmin", utils.Str2Bool, True),
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/admin/delete", methods=["POST"])
|
||||
def api_admin_deleteHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.admin_delete,
|
||||
(FormField("token", str, False), FormField("username", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
# region: Profile
|
||||
|
||||
|
||||
@app.route("/profile/isAdmin", methods=["POST"])
|
||||
def api_profile_isAdminHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.profile_isAdmin, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
@app.route("/profile/changePassword", methods=["POST"])
|
||||
def api_profile_changePasswordHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.profile_changePassword,
|
||||
(FormField("token", str, False), FormField("password", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/profile/getToken", methods=["POST"])
|
||||
def api_profile_getTokenHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.profile_getToken, (FormField("token", str, False),), None
|
||||
)
|
||||
|
||||
|
||||
@app.route("/profile/deleteToken", methods=["POST"])
|
||||
def api_profile_deleteTokenHandle():
|
||||
return SmartDbCaller(
|
||||
calendar_db.profile_deleteToken,
|
||||
(FormField("token", str, False), FormField("deleteToken", str, False)),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
# endregion
|
||||
|
||||
# region: Utilities
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ClientNetworkInfo:
|
||||
user_agent: str
|
||||
"""The user agent of client."""
|
||||
ip_addr: str
|
||||
"""The IP address of client."""
|
||||
|
||||
|
||||
def FetchClientNetworkInfo() -> ClientNetworkInfo:
|
||||
clientUa = request.user_agent.string
|
||||
forwardIpList = request.headers.getlist("X-Forwarded-For")
|
||||
if forwardIpList:
|
||||
clientIp = forwardIpList[0]
|
||||
else:
|
||||
directIp = request.remote_addr
|
||||
if directIp is not None:
|
||||
clientIp = directIp
|
||||
else:
|
||||
clientIp = "0.0.0.0"
|
||||
|
||||
return ClientNetworkInfo(clientUa, clientIp)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FormField:
|
||||
name: str
|
||||
"""The name of form field."""
|
||||
ty: Callable[[str], Any]
|
||||
"""The type of form field."""
|
||||
is_optional: bool
|
||||
"""True if this form field is optional, otherwise false."""
|
||||
|
||||
|
||||
def SmartDbCaller(
|
||||
db_method: Callable[..., ResponseBody[Any]],
|
||||
fields: tuple[FormField, ...],
|
||||
padding_form: dict[str, str] | None,
|
||||
) -> dict[str, Any]:
|
||||
opt_param_counter = 0
|
||||
lost_required: bool = False
|
||||
param_list: list[Any] = []
|
||||
opt_param_dict: dict[str, Any] = {}
|
||||
|
||||
# fetch user passed form
|
||||
user_form: dict[str, str] = request.form.to_dict()
|
||||
LOGGER.debug(f"User Form: {user_form}")
|
||||
# overwrite user form by our padding form
|
||||
if padding_form is not None:
|
||||
user_form.update(padding_form)
|
||||
LOGGER.debug(f"Padded User Form: {user_form}")
|
||||
|
||||
# check fields one by one
|
||||
for field in fields:
|
||||
value = user_form.get(field.name, None)
|
||||
if value is not None:
|
||||
value = field.ty(value)
|
||||
|
||||
if field.is_optional:
|
||||
# optional param
|
||||
if value is not None:
|
||||
opt_param_dict[field.name] = value
|
||||
opt_param_counter += 1
|
||||
else:
|
||||
# required param
|
||||
if value is None:
|
||||
lost_required = True
|
||||
else:
|
||||
param_list.append(value)
|
||||
|
||||
# Only execute database function if there is no lost required fields.
|
||||
# And fulfill one of following requirements:
|
||||
# 1. There are all required fields (optional parameter count is zero).
|
||||
# 1. Or, there is some optional parameter.
|
||||
LOGGER.debug(f"Has Lost Required Parameter: {lost_required}")
|
||||
LOGGER.debug(f"All Optional Parameter Count: {opt_param_counter}")
|
||||
LOGGER.debug(f"Available Optional Parameter Count: {len(opt_param_dict)}")
|
||||
result: ResponseBody[Any]
|
||||
if lost_required == False and (opt_param_counter == 0 or len(opt_param_dict) != 0):
|
||||
result = db_method(*param_list, **opt_param_dict)
|
||||
else:
|
||||
result = ResponseBody(False, "Invalid parameter", None)
|
||||
|
||||
return ConstructResponseBody(result)
|
||||
|
||||
|
||||
def ConstructResponseBody(body: ResponseBody[Any]) -> dict[str, Any]:
|
||||
return {"success": body.success, "error": body.error, "data": body.data}
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
|
||||
def run():
|
||||
calendar_db.open()
|
||||
app.run(port=config.get_config().web.port)
|
||||
calendar_db.close()
|
||||
Reference in New Issue
Block a user