feature: add system setting service (#12408)

### What problem does this PR solve?

#12409 

### Type of change

- [x] New Feature (non-breaking change which adds functionality)

---------

Signed-off-by: Jin Hai <haijin.chn@gmail.com>
This commit is contained in:
Jin Hai
2026-01-04 14:21:39 +08:00
committed by GitHub
parent 11779697de
commit ac9113b0ef
8 changed files with 301 additions and 5 deletions

View File

@ -21,7 +21,7 @@ from flask_login import current_user, login_required, logout_user
from auth import login_verify, login_admin, check_admin_auth
from responses import success_response, error_response
from services import UserMgr, ServiceMgr, UserServiceMgr
from services import UserMgr, ServiceMgr, UserServiceMgr, SettingsMgr
from roles import RoleMgr
from api.common.exceptions import AdminException
from common.versions import get_ragflow_version
@ -406,6 +406,49 @@ def get_user_permission(user_name: str):
except Exception as e:
return error_response(str(e), 500)
@admin_bp.route('/variables', methods=['PUT'])
@login_required
@check_admin_auth
def set_variable():
try:
data = request.get_json()
if not data and 'var_name' not in data:
return error_response("Var name is required", 400)
if 'var_value' not in data:
return error_response("Var value is required", 400)
var_name: str = data['var_name']
var_value: str = data['var_value']
SettingsMgr.update_by_name(var_name, var_value)
return success_response(None, "Set variable successfully")
except AdminException as e:
return error_response(str(e), 400)
except Exception as e:
return error_response(str(e), 500)
@admin_bp.route('/variables', methods=['GET'])
@login_required
@check_admin_auth
def get_variable():
try:
if request.content_length is None or request.content_length == 0:
# list variables
res = list(SettingsMgr.get_all())
return success_response(res)
# get var
data = request.get_json()
if not data and 'var_name' not in data:
return error_response("Var name is required", 400)
var_name: str = data['var_name']
res = SettingsMgr.get_by_name(var_name)
return success_response(res)
except AdminException as e:
return error_response(str(e), 400)
except Exception as e:
return error_response(str(e), 500)
@admin_bp.route('/version', methods=['GET'])
@login_required
@check_admin_auth

View File

@ -24,6 +24,7 @@ from api.db.joint_services.user_account_service import create_new_user, delete_u
from api.db.services.canvas_service import UserCanvasService
from api.db.services.user_service import TenantService
from api.db.services.knowledgebase_service import KnowledgebaseService
from api.db.services.system_settings_service import SystemSettingsService
from api.utils.crypt import decrypt
from api.utils import health_utils
@ -263,3 +264,47 @@ class ServiceMgr:
@staticmethod
def restart_service(service_id: int):
raise AdminException("restart_service: not implemented")
class SettingsMgr:
@staticmethod
def get_all():
settings = SystemSettingsService.get_all()
result = []
for setting in settings:
result.append({
'name': setting.name,
'setting_type': setting.setting_type,
'data_type': setting.data_type,
'value': setting.value,
})
return result
@staticmethod
def get_by_name(name: str):
settings = SystemSettingsService.get_by_name(name)
if len(settings) == 0:
raise AdminException(f"Can't get setting: {name}")
result = []
for setting in settings:
result.append({
'name': setting.name,
'setting_type': setting.setting_type,
'data_type': setting.data_type,
'value': setting.value,
})
return result
@staticmethod
def update_by_name(name: str, value: str):
settings = SystemSettingsService.get_by_name(name)
if len(settings) == 1:
setting = settings[0]
setting.value = value
setting_dict = setting.to_dict()
SystemSettingsService.update_by_name(name, setting_dict)
elif len(settings) > 1:
raise AdminException(f"Can't update more than 1 setting: {name}")
else:
raise AdminException(f"No sett"
f"ing: {name}")