improve: change db schema for settings
This commit is contained in:
122
api/setting.py
Normal file
122
api/setting.py
Normal file
@@ -0,0 +1,122 @@
|
||||
from flask import Blueprint, jsonify, request
|
||||
from api import limiter, etag_response, require_auth
|
||||
from db import get_db
|
||||
from db.models.MarkdownSetting import MarkdownSetting
|
||||
from db.models.PathSetting import PathSetting
|
||||
import api
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
setting_bp = Blueprint('setting', __name__, url_prefix='/api/setting')
|
||||
|
||||
@setting_bp.route('/path/<int:setting_id>', methods=['GET'])
|
||||
@limiter.limit(api.get_rate_limit)
|
||||
@etag_response
|
||||
def get_path_setting(setting_id):
|
||||
with get_db() as session:
|
||||
setting = session.query(PathSetting).get(setting_id)
|
||||
if setting is None:
|
||||
return jsonify({}), 204
|
||||
return jsonify(setting.to_dict()), 200
|
||||
|
||||
@setting_bp.route('/path/', methods=['POST'])
|
||||
@require_auth(roles=['admin'])
|
||||
def create_path_setting():
|
||||
data = request.json
|
||||
webhook_setting_id = data.get('webhook_setting_id')
|
||||
template_setting_id = data.get('template_setting_id')
|
||||
setting = PathSetting(
|
||||
webhook_setting_id=webhook_setting_id,
|
||||
template_setting_id=template_setting_id
|
||||
)
|
||||
with get_db() as session:
|
||||
try:
|
||||
session.add(setting)
|
||||
session.commit()
|
||||
return jsonify(setting.to_dict()), 200
|
||||
except Exception:
|
||||
logger.error(f"failed to create path setting")
|
||||
return jsonify({"error": "failed to create setting"}), 500
|
||||
|
||||
|
||||
@setting_bp.route('/path/<int:setting_id>', methods=['PUT', 'PATCH'])
|
||||
@require_auth(roles=['admin'])
|
||||
def update_path_setting(setting_id):
|
||||
data = request.json
|
||||
with get_db() as session:
|
||||
try:
|
||||
setting = session.query(PathSetting).get(setting_id)
|
||||
if setting is None:
|
||||
return jsonify({"error": "setting not exists"}), 400
|
||||
webhook_setting_id = data.get('webhook_setting_id', setting.webhook_setting_id)
|
||||
template_setting_id = data.get('template_setting_id', setting.template_setting_id)
|
||||
setting.webhook_setting_id = webhook_setting_id
|
||||
setting.template_setting_id = template_setting_id
|
||||
session.commit()
|
||||
return jsonify(setting.to_dict()), 200
|
||||
except Exception:
|
||||
return jsonify({"error": "failed to update path setting"}), 500
|
||||
|
||||
@setting_bp.route('/path/<int:setting_id>', methods=['DELETE'])
|
||||
@require_auth(roles=['admin'])
|
||||
def delete_path_setting(setting_id):
|
||||
with get_db() as session:
|
||||
setting = session.query(PathSetting).get(setting_id)
|
||||
if setting is None:
|
||||
return jsonify({"error": "setting not exists"}), 400
|
||||
session.delete(setting)
|
||||
return jsonify({"message": "deleted"}), 200
|
||||
|
||||
|
||||
@setting_bp.route('/markdown/<int:setting_id>', methods=['GET'])
|
||||
@require_auth(roles=['admin'])
|
||||
@etag_response
|
||||
def get_markdown_path(setting_id):
|
||||
with get_db() as session:
|
||||
setting = session.query(MarkdownSetting).get(setting_id)
|
||||
if setting is None:
|
||||
return jsonify({}), 204
|
||||
return jsonify(setting.to_dict()), 200
|
||||
|
||||
|
||||
@setting_bp.route('/markdown/', methods=['POST'])
|
||||
@require_auth(roles=['admin'])
|
||||
def create_markdown_setting():
|
||||
data = request.json
|
||||
template_setting_id = data.get('template_setting_id')
|
||||
setting = MarkdownSetting(template_setting_id=template_setting_id)
|
||||
try:
|
||||
with get_db() as session:
|
||||
session.add(setting)
|
||||
session.commit()
|
||||
return jsonify(setting.to_dict()), 200
|
||||
except Exception:
|
||||
return jsonify({"error": "failed to create setting"}), 500
|
||||
|
||||
@setting_bp.route('/markdown/<int:setting_id>', methods=['PUT', 'PATCH'])
|
||||
@require_auth(roles=['admin'])
|
||||
def update_markdown_setting(setting_id):
|
||||
data = request.json
|
||||
try:
|
||||
with get_db() as session:
|
||||
setting = session.query(MarkdownSetting).get(setting_id)
|
||||
if setting is None:
|
||||
return jsonify({"error": "setting not exists"}), 400
|
||||
template_setting_id = data.get('template_setting_id', setting.template_setting_id)
|
||||
setting.template_setting_id = template_setting_id
|
||||
session.commit()
|
||||
return jsonify(setting.to_dict()), 200
|
||||
except Exception:
|
||||
return jsonify({"error": "failed to update setting"}), 500
|
||||
|
||||
|
||||
@setting_bp.route('/markdown/<int:setting_id>', methods=['DELETE'])
|
||||
@require_auth(roles=['admin'])
|
||||
def delete_markdown_setting(setting_id):
|
||||
with get_db() as session:
|
||||
setting = session.query(MarkdownSetting).get(setting_id)
|
||||
if setting is None:
|
||||
return jsonify({"error": "setting not exists"}), 400
|
||||
session.delete(setting)
|
||||
session.commit()
|
||||
return jsonify({"message": "deleted"}), 200
|
||||
Reference in New Issue
Block a user