coconut-leaf/backend/server.py

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()