from flask import Flask from flask import request from dataclasses import dataclass from typing import Any, Callable import config import database import utils from logger import LOGGER 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(): # construct client data first clientUa = request.user_agent.string if request.headers.getlist("X-Forwarded-For"): clientIp = request.headers.getlist("X-Forwarded-For")[0] else: clientIp = request.remote_addr return SmartDbCaller(calendar_db.common_login, (FormField('username', str, False), FormField('password', str, False), FormField('clientUa', str, False), FormField('clientIp', str, False)), { 'clientUa': clientUa, 'clientIp': clientIp }) @app.route('/common/webLogin', methods=['POST']) def api_common_webLoginHandle(): # construct client data first clientUa = request.user_agent.string if request.headers.getlist("X-Forwarded-For"): clientIp = request.headers.getlist("X-Forwarded-For")[0] else: clientIp = request.remote_addr return SmartDbCaller(calendar_db.common_webLogin, (FormField('username', str, False), FormField('password', str, False), FormField('clientUa', str, False), FormField('clientIp', str, False)), { 'clientUa': clientUa, 'clientIp': clientIp }) @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: Misc Functions @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, fields: tuple[FormField, ...], padding_form: dict[str, Any] | None) -> dict[str, Any]: result = ResponseBody(False, 'Invalid parameter', None) opt_param_counter = 0 param_list: list[Any] = [] opt_param_dict: dict[str, Any] = {} real_form = request.form.to_dict() LOGGER.debug(f'Form: {real_form}') if padding_form is not None: real_form.update(padding_form) for field in fields: value = real_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: break param_list.append(value) # at least one opt param LOGGER.debug(f'All Optional Parameter: {opt_param_counter}') LOGGER.debug(f'Optional Parameter Count: {len(opt_param_dict)}') if opt_param_counter == 0 or len(opt_param_dict) != 0: result: ResponseBody = db_method(*param_list, **opt_param_dict) return ConstructResponseBody(result) @dataclass(frozen=True) class ResponseBody: success: bool """True if this operation is successful, otherwise false.""" error: str """The error message provided when operation failed.""" data: Any """The payload provided when operation successed.""" def ConstructResponseBody(body: ResponseBody) -> dict[str, Any]: return { 'success': body.success, 'error': body.error, 'data': body.data } def run(): calendar_db.open() app.run(port=config.get_config().web.port) calendar_db.close() # endregion