# # Copyright 2026 The InfiniFlow Authors. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # from quart import jsonify from api.apps import login_required, current_user from api.utils.api_utils import get_json_result, get_data_error_result, server_error_response, generate_confirmation_token from api.utils.health_utils import run_health_checks from common.versions import get_ragflow_version from datetime import datetime from common.time_utils import current_timestamp, datetime_format from api.db.db_models import APIToken from api.db.services.api_service import APITokenService from api.db.services.user_service import UserTenantService from common.log_utils import get_log_levels, set_log_level @manager.route("/system/ping", methods=["GET"]) # noqa: F821 async def ping(): return "pong", 200 @manager.route("/system/version", methods=["GET"]) # noqa: F821 @login_required def version(): """ Get the current version of the application. --- tags: - System security: - ApiKeyAuth: [] responses: 200: description: Version retrieved successfully. schema: type: object properties: version: type: string description: Version number. """ return get_json_result(data=get_ragflow_version()) @manager.route("/system/healthz", methods=["GET"]) # noqa: F821 def healthz(): result, all_ok = run_health_checks() return jsonify(result), (200 if all_ok else 500) @manager.route("/system/tokens", methods=["GET"]) # noqa: F821 @login_required def token_list(): """ List all API tokens for the current user. --- tags: - API Tokens security: - ApiKeyAuth: [] responses: 200: description: List of API tokens. schema: type: object properties: tokens: type: array items: type: object properties: token: type: string description: The API token. name: type: string description: Name of the token. create_time: type: string description: Token creation time. """ try: tenants = UserTenantService.query(user_id=current_user.id) if not tenants: return get_data_error_result(message="Tenant not found!") tenant_id = [tenant for tenant in tenants if tenant.role == "owner"][0].tenant_id objs = APITokenService.query(tenant_id=tenant_id) objs = [o.to_dict() for o in objs] for o in objs: if not o["beta"]: o["beta"] = generate_confirmation_token().replace("ragflow-", "")[:32] APITokenService.filter_update([APIToken.tenant_id == tenant_id, APIToken.token == o["token"]], o) return get_json_result(data=objs) except Exception as e: return server_error_response(e) @manager.route("/system/tokens", methods=["POST"]) # noqa: F821 @login_required def new_token(): """ Generate a new API token. --- tags: - API Tokens security: - ApiKeyAuth: [] parameters: - in: query name: name type: string required: false description: Name of the token. responses: 200: description: Token generated successfully. schema: type: object properties: token: type: string description: The generated API token. """ try: tenants = UserTenantService.query(user_id=current_user.id) if not tenants: return get_data_error_result(message="Tenant not found!") tenant_id = [tenant for tenant in tenants if tenant.role == "owner"][0].tenant_id obj = { "tenant_id": tenant_id, "token": generate_confirmation_token(), "beta": generate_confirmation_token().replace("ragflow-", "")[:32], "create_time": current_timestamp(), "create_date": datetime_format(datetime.now()), "update_time": None, "update_date": None, } if not APITokenService.save(**obj): return get_data_error_result(message="Fail to new a dialog!") return get_json_result(data=obj) except Exception as e: return server_error_response(e) @manager.route("/system/tokens/", methods=["DELETE"]) # noqa: F821 @login_required def rm(token): """ Remove an API token. --- tags: - API Tokens security: - ApiKeyAuth: [] parameters: - in: path name: token type: string required: true description: The API token to remove. responses: 200: description: Token removed successfully. schema: type: object properties: success: type: boolean description: Deletion status. """ try: tenants = UserTenantService.query(user_id=current_user.id) if not tenants: return get_data_error_result(message="Tenant not found!") tenant_id = tenants[0].tenant_id APITokenService.filter_delete([APIToken.tenant_id == tenant_id, APIToken.token == token]) return get_json_result(data=True) except Exception as e: return server_error_response(e) @manager.route("/system/config/log", methods=["GET"]) # noqa: F821 @login_required async def get_logger_levels(): """ Get current log levels for all packages. --- tags: - System responses: 200: description: Return current log levels """ return get_json_result(data=get_log_levels()) @manager.route("/system/config/log", methods=["PUT"]) # noqa: F821 @login_required async def set_logger_level(): """ Set log level for a package. --- tags: - System parameters: - in: body name: body required: true schema: type: object properties: pkg_name: type: string description: Package name (e.g., "rag.utils.es_conn") level: type: string description: Log level (DEBUG, INFO, WARNING, ERROR) responses: 200: description: Log level updated successfully """ from quart import request data = await request.get_json() if not data or "pkg_name" not in data or "level" not in data: return get_data_error_result(message="pkg_name and level are required") pkg_name = data["pkg_name"] level = data["level"] success = set_log_level(pkg_name, level) if success: return get_json_result(data={"pkg_name": pkg_name, "level": level}) else: return get_data_error_result(message=f"Invalid log level: {level}")