From a4ff443d0b8f992272a31efbda1e1a8be53a7f47 Mon Sep 17 00:00:00 2001 From: yongtenglei Date: Thu, 27 Nov 2025 17:03:16 +0800 Subject: [PATCH] base apis --- api/apps/__init__.py | 4 +- api/apps/api_app.py | 8 +- api/apps/auth/github.py | 49 +++++++--- api/apps/auth/oauth.py | 53 +++++++++-- api/apps/auth/oidc.py | 13 ++- api/apps/connector_app.py | 12 +-- api/apps/conversation_app.py | 24 ++--- api/apps/dialog_app.py | 9 +- api/apps/document_app.py | 9 +- api/apps/file2document_app.py | 8 +- api/apps/file_app.py | 10 +-- api/apps/langfuse_app.py | 5 +- api/apps/llm_app.py | 13 ++- api/apps/mcp_server_app.py | 23 +++-- api/apps/sdk/agents.py | 8 +- api/apps/sdk/dify_retrieval.py | 6 +- api/apps/sdk/files.py | 25 +++--- api/apps/sdk/session.py | 46 +++++----- api/apps/search_app.py | 10 +-- api/apps/tenant_app.py | 5 +- api/apps/user_app.py | 81 +++++++++-------- api/ragflow_server.py | 5 +- api/utils/api_utils.py | 22 +++-- common/http_client.py | 157 +++++++++++++++++++++++++++++++++ 24 files changed, 419 insertions(+), 186 deletions(-) create mode 100644 common/http_client.py diff --git a/api/apps/__init__.py b/api/apps/__init__.py index a6e33c13b..5d1d67af5 100644 --- a/api/apps/__init__.py +++ b/api/apps/__init__.py @@ -13,13 +13,12 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import logging import os import sys -import logging from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path from quart import Blueprint, Quart, request, g, current_app, session -from werkzeug.wrappers.request import Request from flasgger import Swagger from itsdangerous.url_safe import URLSafeTimedSerializer as Serializer from quart_cors import cors @@ -40,7 +39,6 @@ settings.init_settings() __all__ = ["app"] -Request.json = property(lambda self: self.get_json(force=True, silent=True)) app = Quart(__name__) app = cors(app, allow_origin="*") diff --git a/api/apps/api_app.py b/api/apps/api_app.py index aa9c9fd6b..ba538a4f0 100644 --- a/api/apps/api_app.py +++ b/api/apps/api_app.py @@ -18,8 +18,7 @@ from quart import request from api.db.db_models import APIToken from api.db.services.api_service import APITokenService, API4ConversationService from api.db.services.user_service import UserTenantService -from api.utils.api_utils import server_error_response, get_data_error_result, get_json_result, validate_request, \ - generate_confirmation_token +from api.utils.api_utils import generate_confirmation_token, get_data_error_result, get_json_result, request_json, server_error_response, validate_request from common.time_utils import current_timestamp, datetime_format from api.apps import login_required, current_user @@ -27,7 +26,7 @@ from api.apps import login_required, current_user @manager.route('/new_token', methods=['POST']) # noqa: F821 @login_required async def new_token(): - req = await request.json + req = await request_json() try: tenants = UserTenantService.query(user_id=current_user.id) if not tenants: @@ -73,7 +72,7 @@ def token_list(): @validate_request("tokens", "tenant_id") @login_required async def rm(): - req = await request.json + req = await request_json() try: for token in req["tokens"]: APITokenService.filter_delete( @@ -116,4 +115,3 @@ def stats(): return get_json_result(data=res) except Exception as e: return server_error_response(e) - diff --git a/api/apps/auth/github.py b/api/apps/auth/github.py index f48d4a5fc..918ff60db 100644 --- a/api/apps/auth/github.py +++ b/api/apps/auth/github.py @@ -14,7 +14,7 @@ # limitations under the License. # -import requests +from common.http_client import async_request, sync_request from .oauth import OAuthClient, UserInfo @@ -34,24 +34,49 @@ class GithubOAuthClient(OAuthClient): def fetch_user_info(self, access_token, **kwargs): """ - Fetch GitHub user info. + Fetch GitHub user info (synchronous). """ user_info = {} try: headers = {"Authorization": f"Bearer {access_token}"} - # user info - response = requests.get(self.userinfo_url, headers=headers, timeout=self.http_request_timeout) + response = sync_request("GET", self.userinfo_url, headers=headers, timeout=self.http_request_timeout) response.raise_for_status() user_info.update(response.json()) - # email info - response = requests.get(self.userinfo_url+"/emails", headers=headers, timeout=self.http_request_timeout) - response.raise_for_status() - email_info = response.json() - user_info["email"] = next( - (email for email in email_info if email["primary"]), None - )["email"] + email_response = sync_request( + "GET", self.userinfo_url + "/emails", headers=headers, timeout=self.http_request_timeout + ) + email_response.raise_for_status() + email_info = email_response.json() + user_info["email"] = next((email for email in email_info if email["primary"]), None)["email"] return self.normalize_user_info(user_info) - except requests.exceptions.RequestException as e: + except Exception as e: + raise ValueError(f"Failed to fetch github user info: {e}") + + async def async_fetch_user_info(self, access_token, **kwargs): + """Async variant of fetch_user_info using httpx.""" + user_info = {} + headers = {"Authorization": f"Bearer {access_token}"} + try: + response = await async_request( + "GET", + self.userinfo_url, + headers=headers, + timeout=self.http_request_timeout, + ) + response.raise_for_status() + user_info.update(response.json()) + + email_response = await async_request( + "GET", + self.userinfo_url + "/emails", + headers=headers, + timeout=self.http_request_timeout, + ) + email_response.raise_for_status() + email_info = email_response.json() + user_info["email"] = next((email for email in email_info if email["primary"]), None)["email"] + return self.normalize_user_info(user_info) + except Exception as e: raise ValueError(f"Failed to fetch github user info: {e}") diff --git a/api/apps/auth/oauth.py b/api/apps/auth/oauth.py index 6f7e0e5b5..5b2afcea1 100644 --- a/api/apps/auth/oauth.py +++ b/api/apps/auth/oauth.py @@ -14,8 +14,8 @@ # limitations under the License. # -import requests import urllib.parse +from common.http_client import async_request, sync_request class UserInfo: @@ -74,15 +74,40 @@ class OAuthClient: "redirect_uri": self.redirect_uri, "grant_type": "authorization_code" } - response = requests.post( + response = sync_request( + "POST", self.token_url, data=payload, headers={"Accept": "application/json"}, - timeout=self.http_request_timeout + timeout=self.http_request_timeout, ) response.raise_for_status() return response.json() - except requests.exceptions.RequestException as e: + except Exception as e: + raise ValueError(f"Failed to exchange authorization code for token: {e}") + + async def async_exchange_code_for_token(self, code): + """ + Async variant of exchange_code_for_token using httpx. + """ + payload = { + "client_id": self.client_id, + "client_secret": self.client_secret, + "code": code, + "redirect_uri": self.redirect_uri, + "grant_type": "authorization_code", + } + try: + response = await async_request( + "POST", + self.token_url, + data=payload, + headers={"Accept": "application/json"}, + timeout=self.http_request_timeout, + ) + response.raise_for_status() + return response.json() + except Exception as e: raise ValueError(f"Failed to exchange authorization code for token: {e}") @@ -92,11 +117,27 @@ class OAuthClient: """ try: headers = {"Authorization": f"Bearer {access_token}"} - response = requests.get(self.userinfo_url, headers=headers, timeout=self.http_request_timeout) + response = sync_request("GET", self.userinfo_url, headers=headers, timeout=self.http_request_timeout) response.raise_for_status() user_info = response.json() return self.normalize_user_info(user_info) - except requests.exceptions.RequestException as e: + except Exception as e: + raise ValueError(f"Failed to fetch user info: {e}") + + async def async_fetch_user_info(self, access_token, **kwargs): + """Async variant of fetch_user_info using httpx.""" + headers = {"Authorization": f"Bearer {access_token}"} + try: + response = await async_request( + "GET", + self.userinfo_url, + headers=headers, + timeout=self.http_request_timeout, + ) + response.raise_for_status() + user_info = response.json() + return self.normalize_user_info(user_info) + except Exception as e: raise ValueError(f"Failed to fetch user info: {e}") diff --git a/api/apps/auth/oidc.py b/api/apps/auth/oidc.py index cafcaadfd..80ac79399 100644 --- a/api/apps/auth/oidc.py +++ b/api/apps/auth/oidc.py @@ -15,7 +15,7 @@ # import jwt -import requests +from common.http_client import sync_request from .oauth import OAuthClient @@ -50,10 +50,10 @@ class OIDCClient(OAuthClient): """ try: metadata_url = f"{issuer}/.well-known/openid-configuration" - response = requests.get(metadata_url, timeout=7) + response = sync_request("GET", metadata_url, timeout=7) response.raise_for_status() return response.json() - except requests.exceptions.RequestException as e: + except Exception as e: raise ValueError(f"Failed to fetch OIDC metadata: {e}") @@ -95,6 +95,13 @@ class OIDCClient(OAuthClient): user_info.update(super().fetch_user_info(access_token).to_dict()) return self.normalize_user_info(user_info) + async def async_fetch_user_info(self, access_token, id_token=None, **kwargs): + user_info = {} + if id_token: + user_info = self.parse_id_token(id_token) + user_info.update((await super().async_fetch_user_info(access_token)).to_dict()) + return self.normalize_user_info(user_info) + def normalize_user_info(self, user_info): return super().normalize_user_info(user_info) diff --git a/api/apps/connector_app.py b/api/apps/connector_app.py index 4932c3fcc..b6d1b4524 100644 --- a/api/apps/connector_app.py +++ b/api/apps/connector_app.py @@ -26,7 +26,7 @@ from google_auth_oauthlib.flow import Flow from api.db import InputType from api.db.services.connector_service import ConnectorService, SyncLogsService -from api.utils.api_utils import get_data_error_result, get_json_result, validate_request +from api.utils.api_utils import get_data_error_result, get_json_result, request_json, validate_request from common.constants import RetCode, TaskStatus from common.data_source.config import GOOGLE_DRIVE_WEB_OAUTH_REDIRECT_URI, DocumentSource from common.data_source.google_util.constant import GOOGLE_DRIVE_WEB_OAUTH_POPUP_TEMPLATE, GOOGLE_SCOPES @@ -38,7 +38,7 @@ from api.apps import login_required, current_user @manager.route("/set", methods=["POST"]) # noqa: F821 @login_required async def set_connector(): - req = await request.json + req = await request_json() if req.get("id"): conn = {fld: req[fld] for fld in ["prune_freq", "refresh_freq", "config", "timeout_secs"] if fld in req} ConnectorService.update_by_id(req["id"], conn) @@ -90,7 +90,7 @@ def list_logs(connector_id): @manager.route("//resume", methods=["PUT"]) # noqa: F821 @login_required async def resume(connector_id): - req = await request.json + req = await request_json() if req.get("resume"): ConnectorService.resume(connector_id, TaskStatus.SCHEDULE) else: @@ -102,7 +102,7 @@ async def resume(connector_id): @login_required @validate_request("kb_id") async def rebuild(connector_id): - req = await request.json + req = await request_json() err = ConnectorService.rebuild(req["kb_id"], connector_id, current_user.id) if err: return get_json_result(data=False, message=err, code=RetCode.SERVER_ERROR) @@ -179,7 +179,7 @@ async def start_google_drive_web_oauth(): message="Google Drive OAuth redirect URI is not configured on the server.", ) - req = await request.json or {} + req = await request_json() raw_credentials = req.get("credentials", "") try: credentials = _load_credentials(raw_credentials) @@ -281,7 +281,7 @@ async def google_drive_web_oauth_callback(): @login_required @validate_request("flow_id") async def poll_google_drive_web_result(): - req = await request.json or {} + req = await request_json() flow_id = req.get("flow_id") cache_raw = REDIS_CONN.get(_web_result_cache_key(flow_id)) if not cache_raw: diff --git a/api/apps/conversation_app.py b/api/apps/conversation_app.py index 77b799016..9fecda18a 100644 --- a/api/apps/conversation_app.py +++ b/api/apps/conversation_app.py @@ -26,7 +26,7 @@ from api.db.services.llm_service import LLMBundle from api.db.services.search_service import SearchService from api.db.services.tenant_llm_service import TenantLLMService from api.db.services.user_service import TenantService, UserTenantService -from api.utils.api_utils import get_data_error_result, get_json_result, server_error_response, validate_request +from api.utils.api_utils import get_data_error_result, get_json_result, request_json, server_error_response, validate_request from rag.prompts.template import load_prompt from rag.prompts.generator import chunks_format from common.constants import RetCode, LLMType @@ -35,7 +35,7 @@ from common.constants import RetCode, LLMType @manager.route("/set", methods=["POST"]) # noqa: F821 @login_required async def set_conversation(): - req = await request.json + req = await request_json() conv_id = req.get("conversation_id") is_new = req.get("is_new") name = req.get("name", "New conversation") @@ -78,7 +78,7 @@ async def set_conversation(): @manager.route("/get", methods=["GET"]) # noqa: F821 @login_required -def get(): +async def get(): conv_id = request.args["conversation_id"] try: e, conv = ConversationService.get_by_id(conv_id) @@ -129,7 +129,7 @@ def getsse(dialog_id): @manager.route("/rm", methods=["POST"]) # noqa: F821 @login_required async def rm(): - req = await request.json + req = await request_json() conv_ids = req["conversation_ids"] try: for cid in conv_ids: @@ -150,7 +150,7 @@ async def rm(): @manager.route("/list", methods=["GET"]) # noqa: F821 @login_required -def list_conversation(): +async def list_conversation(): dialog_id = request.args["dialog_id"] try: if not DialogService.query(tenant_id=current_user.id, id=dialog_id): @@ -167,7 +167,7 @@ def list_conversation(): @login_required @validate_request("conversation_id", "messages") async def completion(): - req = await request.json + req = await request_json() msg = [] for m in req["messages"]: if m["role"] == "system": @@ -252,7 +252,7 @@ async def completion(): @manager.route("/tts", methods=["POST"]) # noqa: F821 @login_required async def tts(): - req = await request.json + req = await request_json() text = req["text"] tenants = TenantService.get_info_by(current_user.id) @@ -285,7 +285,7 @@ async def tts(): @login_required @validate_request("conversation_id", "message_id") async def delete_msg(): - req = await request.json + req = await request_json() e, conv = ConversationService.get_by_id(req["conversation_id"]) if not e: return get_data_error_result(message="Conversation not found!") @@ -308,7 +308,7 @@ async def delete_msg(): @login_required @validate_request("conversation_id", "message_id") async def thumbup(): - req = await request.json + req = await request_json() e, conv = ConversationService.get_by_id(req["conversation_id"]) if not e: return get_data_error_result(message="Conversation not found!") @@ -335,7 +335,7 @@ async def thumbup(): @login_required @validate_request("question", "kb_ids") async def ask_about(): - req = await request.json + req = await request_json() uid = current_user.id search_id = req.get("search_id", "") @@ -367,7 +367,7 @@ async def ask_about(): @login_required @validate_request("question", "kb_ids") async def mindmap(): - req = await request.json + req = await request_json() search_id = req.get("search_id", "") search_app = SearchService.get_detail(search_id) if search_id else {} search_config = search_app.get("search_config", {}) if search_app else {} @@ -385,7 +385,7 @@ async def mindmap(): @login_required @validate_request("question") async def related_questions(): - req = await request.json + req = await request_json() search_id = req.get("search_id", "") search_config = {} diff --git a/api/apps/dialog_app.py b/api/apps/dialog_app.py index cbefc7752..d5868311b 100644 --- a/api/apps/dialog_app.py +++ b/api/apps/dialog_app.py @@ -21,10 +21,9 @@ from common.constants import StatusEnum from api.db.services.tenant_llm_service import TenantLLMService from api.db.services.knowledgebase_service import KnowledgebaseService from api.db.services.user_service import TenantService, UserTenantService -from api.utils.api_utils import server_error_response, get_data_error_result, validate_request +from api.utils.api_utils import get_data_error_result, get_json_result, request_json, server_error_response, validate_request from common.misc_utils import get_uuid from common.constants import RetCode -from api.utils.api_utils import get_json_result from api.apps import login_required, current_user @@ -32,7 +31,7 @@ from api.apps import login_required, current_user @validate_request("prompt_config") @login_required async def set_dialog(): - req = await request.json + req = await request_json() dialog_id = req.get("dialog_id", "") is_create = not dialog_id name = req.get("name", "New Dialog") @@ -181,7 +180,7 @@ async def list_dialogs_next(): else: desc = True - req = await request.get_json() + req = await request_json() owner_ids = req.get("owner_ids", []) try: if not owner_ids: @@ -209,7 +208,7 @@ async def list_dialogs_next(): @login_required @validate_request("dialog_ids") async def rm(): - req = await request.json + req = await request_json() dialog_list=[] tenants = UserTenantService.query(user_id=current_user.id) try: diff --git a/api/apps/document_app.py b/api/apps/document_app.py index 7ec8c1587..412cd2287 100644 --- a/api/apps/document_app.py +++ b/api/apps/document_app.py @@ -230,7 +230,7 @@ async def list_docs(): create_time_from = int(request.args.get("create_time_from", 0)) create_time_to = int(request.args.get("create_time_to", 0)) - req = await request.get_json() + req = await request_json() run_status = req.get("run_status", []) if run_status: @@ -271,7 +271,7 @@ async def list_docs(): @manager.route("/filter", methods=["POST"]) # noqa: F821 @login_required async def get_filter(): - req = await request.get_json() + req = await request_json() kb_id = req.get("kb_id") if not kb_id: @@ -341,7 +341,7 @@ def thumbnails(): @login_required @validate_request("doc_ids", "status") async def change_status(): - req = await request.get_json() + req = await request_json() doc_ids = req.get("doc_ids", []) status = str(req.get("status", "")) @@ -624,7 +624,8 @@ async def upload_and_parse(): @manager.route("/parse", methods=["POST"]) # noqa: F821 @login_required async def parse(): - url = await request.json.get("url") if await request.json else "" + req = await request_json() + url = req.get("url", "") if url: if not is_valid_url(url): return get_json_result(data=False, message="The URL format is invalid", code=RetCode.ARGUMENT_ERROR) diff --git a/api/apps/file2document_app.py b/api/apps/file2document_app.py index 1f8921e92..d5b378e51 100644 --- a/api/apps/file2document_app.py +++ b/api/apps/file2document_app.py @@ -19,22 +19,20 @@ from pathlib import Path from api.db.services.file2document_service import File2DocumentService from api.db.services.file_service import FileService -from quart import request from api.apps import login_required, current_user from api.db.services.knowledgebase_service import KnowledgebaseService -from api.utils.api_utils import server_error_response, get_data_error_result, validate_request +from api.utils.api_utils import get_data_error_result, get_json_result, request_json, server_error_response, validate_request from common.misc_utils import get_uuid from common.constants import RetCode from api.db import FileType from api.db.services.document_service import DocumentService -from api.utils.api_utils import get_json_result @manager.route('/convert', methods=['POST']) # noqa: F821 @login_required @validate_request("file_ids", "kb_ids") async def convert(): - req = await request.json + req = await request_json() kb_ids = req["kb_ids"] file_ids = req["file_ids"] file2documents = [] @@ -104,7 +102,7 @@ async def convert(): @login_required @validate_request("file_ids") async def rm(): - req = await request.json + req = await request_json() file_ids = req["file_ids"] if not file_ids: return get_json_result( diff --git a/api/apps/file_app.py b/api/apps/file_app.py index e262b3d7b..953bf6523 100644 --- a/api/apps/file_app.py +++ b/api/apps/file_app.py @@ -29,7 +29,7 @@ from common.constants import RetCode, FileSource from api.db import FileType from api.db.services import duplicate_name from api.db.services.file_service import FileService -from api.utils.api_utils import get_json_result +from api.utils.api_utils import get_json_result, request_json from api.utils.file_utils import filename_type from api.utils.web_utils import CONTENT_TYPE_MAP from common import settings @@ -124,7 +124,7 @@ async def upload(): @login_required @validate_request("name") async def create(): - req = await request.json + req = await request_json() pf_id = req.get("parent_id") input_file_type = req.get("type") if not pf_id: @@ -239,7 +239,7 @@ def get_all_parent_folders(): @login_required @validate_request("file_ids") async def rm(): - req = await request.json + req = await request_json() file_ids = req["file_ids"] def _delete_single_file(file): @@ -300,7 +300,7 @@ async def rm(): @login_required @validate_request("file_id", "name") async def rename(): - req = await request.json + req = await request_json() try: e, file = FileService.get_by_id(req["file_id"]) if not e: @@ -369,7 +369,7 @@ async def get(file_id): @login_required @validate_request("src_file_ids", "dest_file_id") async def move(): - req = await request.json + req = await request_json() try: file_ids = req["src_file_ids"] dest_parent_id = req["dest_file_id"] diff --git a/api/apps/langfuse_app.py b/api/apps/langfuse_app.py index ffdc6a5fd..a3e6fde86 100644 --- a/api/apps/langfuse_app.py +++ b/api/apps/langfuse_app.py @@ -15,20 +15,19 @@ # -from quart import request from api.apps import current_user, login_required from langfuse import Langfuse from api.db.db_models import DB from api.db.services.langfuse_service import TenantLangfuseService -from api.utils.api_utils import get_error_data_result, get_json_result, server_error_response, validate_request +from api.utils.api_utils import get_error_data_result, get_json_result, request_json, server_error_response, validate_request @manager.route("/api_key", methods=["POST", "PUT"]) # noqa: F821 @login_required @validate_request("secret_key", "public_key", "host") async def set_api_key(): - req = await request.get_json() + req = await request_json() secret_key = req.get("secret_key", "") public_key = req.get("public_key", "") host = req.get("host", "") diff --git a/api/apps/llm_app.py b/api/apps/llm_app.py index 29da88c4f..c4c80b9e1 100644 --- a/api/apps/llm_app.py +++ b/api/apps/llm_app.py @@ -21,10 +21,9 @@ from quart import request from api.apps import login_required, current_user from api.db.services.tenant_llm_service import LLMFactoriesService, TenantLLMService from api.db.services.llm_service import LLMService -from api.utils.api_utils import server_error_response, get_data_error_result, validate_request +from api.utils.api_utils import get_allowed_llm_factories, get_data_error_result, get_json_result, request_json, server_error_response, validate_request from common.constants import StatusEnum, LLMType from api.db.db_models import TenantLLM -from api.utils.api_utils import get_json_result, get_allowed_llm_factories from rag.utils.base64_image import test_image from rag.llm import EmbeddingModel, ChatModel, RerankModel, CvModel, TTSModel @@ -54,7 +53,7 @@ def factories(): @login_required @validate_request("llm_factory", "api_key") async def set_api_key(): - req = await request.json + req = await request_json() # test if api key works chat_passed, embd_passed, rerank_passed = False, False, False factory = req["llm_factory"] @@ -124,7 +123,7 @@ async def set_api_key(): @login_required @validate_request("llm_factory") async def add_llm(): - req = await request.json + req = await request_json() factory = req["llm_factory"] api_key = req.get("api_key", "x") llm_name = req.get("llm_name") @@ -269,7 +268,7 @@ async def add_llm(): @login_required @validate_request("llm_factory", "llm_name") async def delete_llm(): - req = await request.json + req = await request_json() TenantLLMService.filter_delete([TenantLLM.tenant_id == current_user.id, TenantLLM.llm_factory == req["llm_factory"], TenantLLM.llm_name == req["llm_name"]]) return get_json_result(data=True) @@ -278,7 +277,7 @@ async def delete_llm(): @login_required @validate_request("llm_factory", "llm_name") async def enable_llm(): - req = await request.json + req = await request_json() TenantLLMService.filter_update( [TenantLLM.tenant_id == current_user.id, TenantLLM.llm_factory == req["llm_factory"], TenantLLM.llm_name == req["llm_name"]], {"status": str(req.get("status", "1"))} ) @@ -289,7 +288,7 @@ async def enable_llm(): @login_required @validate_request("llm_factory") async def delete_factory(): - req = await request.json + req = await request_json() TenantLLMService.filter_delete([TenantLLM.tenant_id == current_user.id, TenantLLM.llm_factory == req["llm_factory"]]) return get_json_result(data=True) diff --git a/api/apps/mcp_server_app.py b/api/apps/mcp_server_app.py index 583f721c4..dec429a09 100644 --- a/api/apps/mcp_server_app.py +++ b/api/apps/mcp_server_app.py @@ -22,8 +22,7 @@ from api.db.services.user_service import TenantService from common.constants import RetCode, VALID_MCP_SERVER_TYPES from common.misc_utils import get_uuid -from api.utils.api_utils import get_data_error_result, get_json_result, server_error_response, validate_request, \ - get_mcp_tools +from api.utils.api_utils import get_data_error_result, get_json_result, get_mcp_tools, request_json, server_error_response, validate_request from api.utils.web_utils import get_float, safe_json_parse from common.mcp_tool_call_conn import MCPToolCallSession, close_multiple_mcp_toolcall_sessions @@ -40,7 +39,7 @@ async def list_mcp() -> Response: else: desc = True - req = await request.get_json() + req = await request_json() mcp_ids = req.get("mcp_ids", []) try: servers = MCPServerService.get_servers(current_user.id, mcp_ids, 0, 0, orderby, desc, keywords) or [] @@ -73,7 +72,7 @@ def detail() -> Response: @login_required @validate_request("name", "url", "server_type") async def create() -> Response: - req = await request.get_json() + req = await request_json() server_type = req.get("server_type", "") if server_type not in VALID_MCP_SERVER_TYPES: @@ -128,7 +127,7 @@ async def create() -> Response: @login_required @validate_request("mcp_id") async def update() -> Response: - req = await request.get_json() + req = await request_json() mcp_id = req.get("mcp_id", "") e, mcp_server = MCPServerService.get_by_id(mcp_id) @@ -184,7 +183,7 @@ async def update() -> Response: @login_required @validate_request("mcp_ids") async def rm() -> Response: - req = await request.get_json() + req = await request_json() mcp_ids = req.get("mcp_ids", []) try: @@ -202,7 +201,7 @@ async def rm() -> Response: @login_required @validate_request("mcpServers") async def import_multiple() -> Response: - req = await request.get_json() + req = await request_json() servers = req.get("mcpServers", {}) if not servers: return get_data_error_result(message="No MCP servers provided.") @@ -269,7 +268,7 @@ async def import_multiple() -> Response: @login_required @validate_request("mcp_ids") async def export_multiple() -> Response: - req = await request.get_json() + req = await request_json() mcp_ids = req.get("mcp_ids", []) if not mcp_ids: @@ -301,7 +300,7 @@ async def export_multiple() -> Response: @login_required @validate_request("mcp_ids") async def list_tools() -> Response: - req = await request.get_json() + req = await request_json() mcp_ids = req.get("mcp_ids", []) if not mcp_ids: return get_data_error_result(message="No MCP server IDs provided.") @@ -348,7 +347,7 @@ async def list_tools() -> Response: @login_required @validate_request("mcp_id", "tool_name", "arguments") async def test_tool() -> Response: - req = await request.get_json() + req = await request_json() mcp_id = req.get("mcp_id", "") if not mcp_id: return get_data_error_result(message="No MCP server ID provided.") @@ -381,7 +380,7 @@ async def test_tool() -> Response: @login_required @validate_request("mcp_id", "tools") async def cache_tool() -> Response: - req = await request.get_json() + req = await request_json() mcp_id = req.get("mcp_id", "") if not mcp_id: return get_data_error_result(message="No MCP server ID provided.") @@ -404,7 +403,7 @@ async def cache_tool() -> Response: @manager.route("/test_mcp", methods=["POST"]) # noqa: F821 @validate_request("url", "server_type") async def test_mcp() -> Response: - req = await request.get_json() + req = await request_json() url = req.get("url", "") if not url: diff --git a/api/apps/sdk/agents.py b/api/apps/sdk/agents.py index b20a22ad8..d9ce86d2f 100644 --- a/api/apps/sdk/agents.py +++ b/api/apps/sdk/agents.py @@ -25,7 +25,7 @@ from api.db.services.canvas_service import UserCanvasService from api.db.services.user_canvas_version import UserCanvasVersionService from common.constants import RetCode from common.misc_utils import get_uuid -from api.utils.api_utils import get_data_error_result, get_error_data_result, get_json_result, token_required +from api.utils.api_utils import get_data_error_result, get_error_data_result, get_json_result, request_json, token_required from api.utils.api_utils import get_result from quart import request, Response @@ -53,7 +53,7 @@ def list_agents(tenant_id): @manager.route("/agents", methods=["POST"]) # noqa: F821 @token_required async def create_agent(tenant_id: str): - req: dict[str, Any] = cast(dict[str, Any], await request.json) + req: dict[str, Any] = cast(dict[str, Any], await request_json()) req["user_id"] = tenant_id if req.get("dsl") is not None: @@ -90,7 +90,7 @@ async def create_agent(tenant_id: str): @manager.route("/agents/", methods=["PUT"]) # noqa: F821 @token_required async def update_agent(tenant_id: str, agent_id: str): - req: dict[str, Any] = {k: v for k, v in cast(dict[str, Any], (await request.json)).items() if v is not None} + req: dict[str, Any] = {k: v for k, v in cast(dict[str, Any], (await request_json())).items() if v is not None} req["user_id"] = tenant_id if req.get("dsl") is not None: @@ -136,7 +136,7 @@ def delete_agent(tenant_id: str, agent_id: str): @manager.route('/webhook/', methods=['POST']) # noqa: F821 @token_required async def webhook(tenant_id: str, agent_id: str): - req = await request.json + req = await request_json() if not UserCanvasService.accessible(req["id"], tenant_id): return get_json_result( data=False, message='Only owner of canvas authorized for this operation.', diff --git a/api/apps/sdk/dify_retrieval.py b/api/apps/sdk/dify_retrieval.py index 55ea54faf..4ddbf9a74 100644 --- a/api/apps/sdk/dify_retrieval.py +++ b/api/apps/sdk/dify_retrieval.py @@ -15,12 +15,12 @@ # import logging -from quart import request, jsonify +from quart import jsonify from api.db.services.document_service import DocumentService from api.db.services.knowledgebase_service import KnowledgebaseService from api.db.services.llm_service import LLMBundle -from api.utils.api_utils import validate_request, build_error_result, apikey_required +from api.utils.api_utils import apikey_required, build_error_result, request_json, validate_request from rag.app.tag import label_question from api.db.services.dialog_service import meta_filter, convert_conditions from common.constants import RetCode, LLMType @@ -113,7 +113,7 @@ async def retrieval(tenant_id): 404: description: Knowledge base or document not found """ - req = await request.json + req = await request_json() question = req["query"] kb_id = req["knowledge_id"] use_kg = req.get("use_kg", False) diff --git a/api/apps/sdk/files.py b/api/apps/sdk/files.py index 6377ea7c8..d67f1d072 100644 --- a/api/apps/sdk/files.py +++ b/api/apps/sdk/files.py @@ -23,12 +23,11 @@ from pathlib import Path from api.db.services.document_service import DocumentService from api.db.services.file2document_service import File2DocumentService from api.db.services.knowledgebase_service import KnowledgebaseService -from api.utils.api_utils import server_error_response, token_required +from api.utils.api_utils import get_json_result, request_json, server_error_response, token_required from common.misc_utils import get_uuid from api.db import FileType from api.db.services import duplicate_name from api.db.services.file_service import FileService -from api.utils.api_utils import get_json_result from api.utils.file_utils import filename_type from common import settings from common.constants import RetCode @@ -193,9 +192,9 @@ async def create(tenant_id): type: type: string """ - req = await request.json - pf_id = await request.json.get("parent_id") - input_file_type = await request.json.get("type") + req = await request_json() + pf_id = req.get("parent_id") + input_file_type = req.get("type") if not pf_id: root_folder = FileService.get_root_folder(tenant_id) pf_id = root_folder["id"] @@ -229,7 +228,7 @@ async def create(tenant_id): @manager.route('/file/list', methods=['GET']) # noqa: F821 @token_required -def list_files(tenant_id): +async def list_files(tenant_id): """ List files under a specific folder. --- @@ -321,7 +320,7 @@ def list_files(tenant_id): @manager.route('/file/root_folder', methods=['GET']) # noqa: F821 @token_required -def get_root_folder(tenant_id): +async def get_root_folder(tenant_id): """ Get user's root folder. --- @@ -357,7 +356,7 @@ def get_root_folder(tenant_id): @manager.route('/file/parent_folder', methods=['GET']) # noqa: F821 @token_required -def get_parent_folder(): +async def get_parent_folder(): """ Get parent folder info of a file. --- @@ -402,7 +401,7 @@ def get_parent_folder(): @manager.route('/file/all_parent_folder', methods=['GET']) # noqa: F821 @token_required -def get_all_parent_folders(tenant_id): +async def get_all_parent_folders(tenant_id): """ Get all parent folders of a file. --- @@ -481,7 +480,7 @@ async def rm(tenant_id): type: boolean example: true """ - req = await request.json + req = await request_json() file_ids = req["file_ids"] try: for file_id in file_ids: @@ -556,7 +555,7 @@ async def rename(tenant_id): type: boolean example: true """ - req = await request.json + req = await request_json() try: e, file = FileService.get_by_id(req["file_id"]) if not e: @@ -667,7 +666,7 @@ async def move(tenant_id): type: boolean example: true """ - req = await request.json + req = await request_json() try: file_ids = req["src_file_ids"] parent_id = req["dest_file_id"] @@ -694,7 +693,7 @@ async def move(tenant_id): @manager.route('/file/convert', methods=['POST']) # noqa: F821 @token_required async def convert(tenant_id): - req = await request.json + req = await request_json() kb_ids = req["kb_ids"] file_ids = req["file_ids"] file2documents = [] diff --git a/api/apps/sdk/session.py b/api/apps/sdk/session.py index 074401ede..138833936 100644 --- a/api/apps/sdk/session.py +++ b/api/apps/sdk/session.py @@ -35,7 +35,7 @@ from api.db.services.search_service import SearchService from api.db.services.user_service import UserTenantService from common.misc_utils import get_uuid from api.utils.api_utils import check_duplicate_ids, get_data_openai, get_error_data_result, get_json_result, \ - get_result, server_error_response, token_required, validate_request + get_result, request_json, server_error_response, token_required, validate_request from rag.app.tag import label_question from rag.prompts.template import load_prompt from rag.prompts.generator import cross_languages, gen_meta_filter, keyword_extraction, chunks_format @@ -45,7 +45,7 @@ from common import settings @manager.route("/chats//sessions", methods=["POST"]) # noqa: F821 @token_required async def create(tenant_id, chat_id): - req = await request.json + req = await request_json() req["dialog_id"] = chat_id dia = DialogService.query(tenant_id=tenant_id, id=req["dialog_id"], status=StatusEnum.VALID.value) if not dia: @@ -73,7 +73,7 @@ async def create(tenant_id, chat_id): @manager.route("/agents//sessions", methods=["POST"]) # noqa: F821 @token_required -def create_agent_session(tenant_id, agent_id): +async def create_agent_session(tenant_id, agent_id): user_id = request.args.get("user_id", tenant_id) e, cvs = UserCanvasService.get_by_id(agent_id) if not e: @@ -98,7 +98,7 @@ def create_agent_session(tenant_id, agent_id): @manager.route("/chats//sessions/", methods=["PUT"]) # noqa: F821 @token_required async def update(tenant_id, chat_id, session_id): - req = await request.json + req = await request_json() req["dialog_id"] = chat_id conv_id = session_id conv = ConversationService.query(id=conv_id, dialog_id=chat_id) @@ -120,7 +120,7 @@ async def update(tenant_id, chat_id, session_id): @manager.route("/chats//completions", methods=["POST"]) # noqa: F821 @token_required async def chat_completion(tenant_id, chat_id): - req = await request.json + req = await request_json() if not req: req = {"question": ""} if not req.get("session_id"): @@ -206,7 +206,7 @@ async def chat_completion_openai_like(tenant_id, chat_id): if reference: print(completion.choices[0].message.reference) """ - req = await request.get_json() + req = await request_json() need_reference = bool(req.get("reference", False)) @@ -384,7 +384,7 @@ async def chat_completion_openai_like(tenant_id, chat_id): @validate_request("model", "messages") # noqa: F821 @token_required async def agents_completion_openai_compatibility(tenant_id, agent_id): - req = await request.json + req = await request_json() tiktokenenc = tiktoken.get_encoding("cl100k_base") messages = req.get("messages", []) if not messages: @@ -442,7 +442,7 @@ async def agents_completion_openai_compatibility(tenant_id, agent_id): @manager.route("/agents//completions", methods=["POST"]) # noqa: F821 @token_required async def agent_completions(tenant_id, agent_id): - req = await request.json + req = await request_json() if req.get("stream", True): @@ -491,7 +491,7 @@ async def agent_completions(tenant_id, agent_id): @manager.route("/chats//sessions", methods=["GET"]) # noqa: F821 @token_required -def list_session(tenant_id, chat_id): +async def list_session(tenant_id, chat_id): if not DialogService.query(tenant_id=tenant_id, id=chat_id, status=StatusEnum.VALID.value): return get_error_data_result(message=f"You don't own the assistant {chat_id}.") id = request.args.get("id") @@ -545,7 +545,7 @@ def list_session(tenant_id, chat_id): @manager.route("/agents//sessions", methods=["GET"]) # noqa: F821 @token_required -def list_agent_session(tenant_id, agent_id): +async def list_agent_session(tenant_id, agent_id): if not UserCanvasService.query(user_id=tenant_id, id=agent_id): return get_error_data_result(message=f"You don't own the agent {agent_id}.") id = request.args.get("id") @@ -614,7 +614,7 @@ async def delete(tenant_id, chat_id): errors = [] success_count = 0 - req = await request.json + req = await request_json() convs = ConversationService.query(dialog_id=chat_id) if not req: ids = None @@ -662,7 +662,7 @@ async def delete(tenant_id, chat_id): async def delete_agent_session(tenant_id, agent_id): errors = [] success_count = 0 - req = await request.json + req = await request_json() cvs = UserCanvasService.query(user_id=tenant_id, id=agent_id) if not cvs: return get_error_data_result(f"You don't own the agent {agent_id}") @@ -715,7 +715,7 @@ async def delete_agent_session(tenant_id, agent_id): @manager.route("/sessions/ask", methods=["POST"]) # noqa: F821 @token_required async def ask_about(tenant_id): - req = await request.json + req = await request_json() if not req.get("question"): return get_error_data_result("`question` is required.") if not req.get("dataset_ids"): @@ -754,7 +754,7 @@ async def ask_about(tenant_id): @manager.route("/sessions/related_questions", methods=["POST"]) # noqa: F821 @token_required async def related_questions(tenant_id): - req = await request.json + req = await request_json() if not req.get("question"): return get_error_data_result("`question` is required.") question = req["question"] @@ -805,7 +805,7 @@ Related search terms: @manager.route("/chatbots//completions", methods=["POST"]) # noqa: F821 async def chatbot_completions(dialog_id): - req = await request.json + req = await request_json() token = request.headers.get("Authorization").split() if len(token) != 2: @@ -831,7 +831,7 @@ async def chatbot_completions(dialog_id): @manager.route("/chatbots//info", methods=["GET"]) # noqa: F821 -def chatbots_inputs(dialog_id): +async def chatbots_inputs(dialog_id): token = request.headers.get("Authorization").split() if len(token) != 2: return get_error_data_result(message='Authorization is not valid!"') @@ -855,7 +855,7 @@ def chatbots_inputs(dialog_id): @manager.route("/agentbots//completions", methods=["POST"]) # noqa: F821 async def agent_bot_completions(agent_id): - req = await request.json + req = await request_json() token = request.headers.get("Authorization").split() if len(token) != 2: @@ -878,7 +878,7 @@ async def agent_bot_completions(agent_id): @manager.route("/agentbots//inputs", methods=["GET"]) # noqa: F821 -def begin_inputs(agent_id): +async def begin_inputs(agent_id): token = request.headers.get("Authorization").split() if len(token) != 2: return get_error_data_result(message='Authorization is not valid!"') @@ -908,7 +908,7 @@ async def ask_about_embedded(): if not objs: return get_error_data_result(message='Authentication error: API key is invalid!"') - req = await request.json + req = await request_json() uid = objs[0].tenant_id search_id = req.get("search_id", "") @@ -947,7 +947,7 @@ async def retrieval_test_embedded(): if not objs: return get_error_data_result(message='Authentication error: API key is invalid!"') - req = await request.json + req = await request_json() page = int(req.get("page", 1)) size = int(req.get("size", 30)) question = req["question"] @@ -1046,7 +1046,7 @@ async def related_questions_embedded(): if not objs: return get_error_data_result(message='Authentication error: API key is invalid!"') - req = await request.json + req = await request_json() tenant_id = objs[0].tenant_id if not tenant_id: return get_error_data_result(message="permission denined.") @@ -1081,7 +1081,7 @@ Related search terms: @manager.route("/searchbots/detail", methods=["GET"]) # noqa: F821 -def detail_share_embedded(): +async def detail_share_embedded(): token = request.headers.get("Authorization").split() if len(token) != 2: return get_error_data_result(message='Authorization is not valid!"') @@ -1123,7 +1123,7 @@ async def mindmap(): return get_error_data_result(message='Authentication error: API key is invalid!"') tenant_id = objs[0].tenant_id - req = await request.json + req = await request_json() search_id = req.get("search_id", "") search_app = SearchService.get_detail(search_id) if search_id else {} diff --git a/api/apps/search_app.py b/api/apps/search_app.py index d350b93c3..3b0da3fe1 100644 --- a/api/apps/search_app.py +++ b/api/apps/search_app.py @@ -24,14 +24,14 @@ from api.db.services.search_service import SearchService from api.db.services.user_service import TenantService, UserTenantService from common.misc_utils import get_uuid from common.constants import RetCode, StatusEnum -from api.utils.api_utils import get_data_error_result, get_json_result, not_allowed_parameters, server_error_response, validate_request +from api.utils.api_utils import get_data_error_result, get_json_result, not_allowed_parameters, request_json, server_error_response, validate_request @manager.route("/create", methods=["post"]) # noqa: F821 @login_required @validate_request("name") async def create(): - req = await request.get_json() + req = await request_json() search_name = req["name"] description = req.get("description", "") if not isinstance(search_name, str): @@ -66,7 +66,7 @@ async def create(): @validate_request("search_id", "name", "search_config", "tenant_id") @not_allowed_parameters("id", "created_by", "create_time", "update_time", "create_date", "update_date", "created_by") async def update(): - req = await request.get_json() + req = await request_json() if not isinstance(req["name"], str): return get_data_error_result(message="Search name must be string.") if req["name"].strip() == "": @@ -150,7 +150,7 @@ async def list_search_app(): else: desc = True - req = await request.get_json() + req = await request_json() owner_ids = req.get("owner_ids", []) try: if not owner_ids: @@ -174,7 +174,7 @@ async def list_search_app(): @login_required @validate_request("search_id") async def rm(): - req = await request.get_json() + req = await request_json() search_id = req["search_id"] if not SearchService.accessible4deletion(search_id, current_user.id): return get_json_result(data=False, message="No authorization.", code=RetCode.AUTHENTICATION_ERROR) diff --git a/api/apps/tenant_app.py b/api/apps/tenant_app.py index 380838bcd..837c812c1 100644 --- a/api/apps/tenant_app.py +++ b/api/apps/tenant_app.py @@ -14,7 +14,6 @@ # limitations under the License. # -from quart import request from api.db import UserTenantRole from api.db.db_models import UserTenant from api.db.services.user_service import UserTenantService, UserService @@ -22,7 +21,7 @@ from api.db.services.user_service import UserTenantService, UserService from common.constants import RetCode, StatusEnum from common.misc_utils import get_uuid from common.time_utils import delta_seconds -from api.utils.api_utils import get_json_result, validate_request, server_error_response, get_data_error_result +from api.utils.api_utils import get_data_error_result, get_json_result, request_json, server_error_response, validate_request from api.utils.web_utils import send_invite_email from common import settings from api.apps import smtp_mail_server, login_required, current_user @@ -56,7 +55,7 @@ async def create(tenant_id): message='No authorization.', code=RetCode.AUTHENTICATION_ERROR) - req = await request.json + req = await request_json() invite_user_email = req["email"] invite_users = UserService.query(email=invite_user_email) if not invite_users: diff --git a/api/apps/user_app.py b/api/apps/user_app.py index 46e114fdf..80f94c659 100644 --- a/api/apps/user_app.py +++ b/api/apps/user_app.py @@ -39,6 +39,7 @@ from common.connection_utils import construct_response from api.utils.api_utils import ( get_data_error_result, get_json_result, + request_json, server_error_response, validate_request, ) @@ -57,6 +58,7 @@ from api.utils.web_utils import ( captcha_key, ) from common import settings +from common.http_client import async_request @manager.route("/login", methods=["POST", "GET"]) # noqa: F821 @@ -90,7 +92,7 @@ async def login(): schema: type: object """ - json_body = await request.json + json_body = await request_json() if not json_body: return get_json_result(data=False, code=RetCode.AUTHENTICATION_ERROR, message="Unauthorized!") @@ -136,7 +138,7 @@ async def login(): @manager.route("/login/channels", methods=["GET"]) # noqa: F821 -def get_login_channels(): +async def get_login_channels(): """ Get all supported authentication channels. """ @@ -157,7 +159,7 @@ def get_login_channels(): @manager.route("/login/", methods=["GET"]) # noqa: F821 -def oauth_login(channel): +async def oauth_login(channel): channel_config = settings.OAUTH_CONFIG.get(channel) if not channel_config: raise ValueError(f"Invalid channel name: {channel}") @@ -170,7 +172,7 @@ def oauth_login(channel): @manager.route("/oauth/callback/", methods=["GET"]) # noqa: F821 -def oauth_callback(channel): +async def oauth_callback(channel): """ Handle the OAuth/OIDC callback for various channels dynamically. """ @@ -192,7 +194,10 @@ def oauth_callback(channel): return redirect("/?error=missing_code") # Exchange authorization code for access token - token_info = auth_cli.exchange_code_for_token(code) + if hasattr(auth_cli, "async_exchange_code_for_token"): + token_info = await auth_cli.async_exchange_code_for_token(code) + else: + token_info = auth_cli.exchange_code_for_token(code) access_token = token_info.get("access_token") if not access_token: return redirect("/?error=token_failed") @@ -200,7 +205,10 @@ def oauth_callback(channel): id_token = token_info.get("id_token") # Fetch user info - user_info = auth_cli.fetch_user_info(access_token, id_token=id_token) + if hasattr(auth_cli, "async_fetch_user_info"): + user_info = await auth_cli.async_fetch_user_info(access_token, id_token=id_token) + else: + user_info = auth_cli.fetch_user_info(access_token, id_token=id_token) if not user_info.email: return redirect("/?error=email_missing") @@ -259,7 +267,7 @@ def oauth_callback(channel): @manager.route("/github_callback", methods=["GET"]) # noqa: F821 -def github_callback(): +async def github_callback(): """ **Deprecated**, Use `/oauth/callback/` instead. @@ -279,9 +287,8 @@ def github_callback(): schema: type: object """ - import requests - - res = requests.post( + res = await async_request( + "POST", settings.GITHUB_OAUTH.get("url"), data={ "client_id": settings.GITHUB_OAUTH.get("client_id"), @@ -299,7 +306,7 @@ def github_callback(): session["access_token"] = res["access_token"] session["access_token_from"] = "github" - user_info = user_info_from_github(session["access_token"]) + user_info = await user_info_from_github(session["access_token"]) email_address = user_info["email"] users = UserService.query(email=email_address) user_id = get_uuid() @@ -348,7 +355,7 @@ def github_callback(): @manager.route("/feishu_callback", methods=["GET"]) # noqa: F821 -def feishu_callback(): +async def feishu_callback(): """ Feishu OAuth callback endpoint. --- @@ -366,9 +373,8 @@ def feishu_callback(): schema: type: object """ - import requests - - app_access_token_res = requests.post( + app_access_token_res = await async_request( + "POST", settings.FEISHU_OAUTH.get("app_access_token_url"), data=json.dumps( { @@ -382,7 +388,8 @@ def feishu_callback(): if app_access_token_res["code"] != 0: return redirect("/?error=%s" % app_access_token_res) - res = requests.post( + res = await async_request( + "POST", settings.FEISHU_OAUTH.get("user_access_token_url"), data=json.dumps( { @@ -403,7 +410,7 @@ def feishu_callback(): return redirect("/?error=contact:user.email:readonly not in scope") session["access_token"] = res["data"]["access_token"] session["access_token_from"] = "feishu" - user_info = user_info_from_feishu(session["access_token"]) + user_info = await user_info_from_feishu(session["access_token"]) email_address = user_info["email"] users = UserService.query(email=email_address) user_id = get_uuid() @@ -451,36 +458,34 @@ def feishu_callback(): return redirect("/?auth=%s" % user.get_id()) -def user_info_from_feishu(access_token): - import requests - +async def user_info_from_feishu(access_token): headers = { "Content-Type": "application/json; charset=utf-8", "Authorization": f"Bearer {access_token}", } - res = requests.get("https://open.feishu.cn/open-apis/authen/v1/user_info", headers=headers) + res = await async_request("GET", "https://open.feishu.cn/open-apis/authen/v1/user_info", headers=headers) user_info = res.json()["data"] user_info["email"] = None if user_info.get("email") == "" else user_info["email"] return user_info -def user_info_from_github(access_token): - import requests - +async def user_info_from_github(access_token): headers = {"Accept": "application/json", "Authorization": f"token {access_token}"} - res = requests.get(f"https://api.github.com/user?access_token={access_token}", headers=headers) + res = await async_request("GET", f"https://api.github.com/user?access_token={access_token}", headers=headers) user_info = res.json() - email_info = requests.get( + email_info_response = await async_request( + "GET", f"https://api.github.com/user/emails?access_token={access_token}", headers=headers, - ).json() + ) + email_info = email_info_response.json() user_info["email"] = next((email for email in email_info if email["primary"]), None)["email"] return user_info @manager.route("/logout", methods=["GET"]) # noqa: F821 @login_required -def log_out(): +async def log_out(): """ User logout endpoint. --- @@ -531,7 +536,7 @@ async def setting_user(): type: object """ update_dict = {} - request_data = await request.json + request_data = await request_json() if request_data.get("password"): new_password = request_data.get("new_password") if not check_password_hash(current_user.password, decrypt(request_data["password"])): @@ -570,7 +575,7 @@ async def setting_user(): @manager.route("/info", methods=["GET"]) # noqa: F821 @login_required -def user_profile(): +async def user_profile(): """ Get user profile information. --- @@ -698,7 +703,7 @@ async def user_add(): code=RetCode.OPERATING_ERROR, ) - req = await request.json + req = await request_json() email_address = req["email"] # Validate the email address @@ -755,7 +760,7 @@ async def user_add(): @manager.route("/tenant_info", methods=["GET"]) # noqa: F821 @login_required -def tenant_info(): +async def tenant_info(): """ Get tenant information. --- @@ -831,14 +836,14 @@ async def set_tenant_info(): schema: type: object """ - req = await request.json + req = await request_json() try: tid = req.pop("tenant_id") TenantService.update_by_id(tid, req) return get_json_result(data=True) except Exception as e: return server_error_response(e) - + @manager.route("/forget/captcha", methods=["GET"]) # noqa: F821 async def forget_get_captcha(): @@ -875,7 +880,7 @@ async def forget_send_otp(): - Verify the image captcha stored at captcha:{email} (case-insensitive). - On success, generate an email OTP (A–Z with length = OTP_LENGTH), store hash + salt (and timestamp) in Redis with TTL, reset attempts and cooldown, and send the OTP via email. """ - req = await request.get_json() + req = await request_json() email = req.get("email") or "" captcha = (req.get("captcha") or "").strip() @@ -931,7 +936,7 @@ async def forget_send_otp(): ) except Exception: return get_json_result(data=False, code=RetCode.SERVER_ERROR, message="failed to send email") - + return get_json_result(data=True, code=RetCode.SUCCESS, message="verification passed, email sent") @@ -941,7 +946,7 @@ async def forget(): POST: Verify email + OTP and reset password, then log the user in. Request JSON: { email, otp, new_password, confirm_new_password } """ - req = await request.get_json() + req = await request_json() email = req.get("email") or "" otp = (req.get("otp") or "").strip() new_pwd = req.get("new_password") @@ -1006,4 +1011,4 @@ async def forget(): user.update_date = (datetime_format(datetime.now()),) user.save() msg = "Password reset successful. Logged in." - return construct_response(data=user.to_json(), auth=user.get_id(), message=msg) + return await construct_response(data=user.to_json(), auth=user.get_id(), message=msg) diff --git a/api/ragflow_server.py b/api/ragflow_server.py index f6cb7bc2b..59622fe68 100644 --- a/api/ragflow_server.py +++ b/api/ragflow_server.py @@ -25,7 +25,6 @@ import logging import os import signal import sys -import time import traceback import threading import uuid @@ -69,7 +68,7 @@ def signal_handler(sig, frame): logging.info("Received interrupt signal, shutting down...") shutdown_all_mcp_sessions() stop_event.set() - time.sleep(1) + stop_event.wait(1) sys.exit(0) if __name__ == '__main__': @@ -163,5 +162,5 @@ if __name__ == '__main__': except Exception: traceback.print_exc() stop_event.set() - time.sleep(1) + stop_event.wait(1) os.kill(os.getpid(), signal.SIGKILL) diff --git a/api/utils/api_utils.py b/api/utils/api_utils.py index 314211694..d3bfdb920 100644 --- a/api/utils/api_utils.py +++ b/api/utils/api_utils.py @@ -44,12 +44,22 @@ from common import settings requests.models.complexjson.dumps = functools.partial(json.dumps, cls=CustomJSONEncoder) +async def _coerce_request_data() -> dict: + """Fetch JSON body with sane defaults; fallback to form data.""" + try: + payload = await request.get_json(force=True, silent=True) + except Exception: + payload = None + if payload is None: + try: + payload = (await request.form).to_dict() + except Exception: + payload = {} + return payload or {} + async def request_json(): - try: - return await request.json - except Exception: - return {} + return await _coerce_request_data() def serialize_for_json(obj): """ @@ -137,7 +147,7 @@ def validate_request(*args, **kwargs): def wrapper(func): @wraps(func) async def decorated_function(*_args, **_kwargs): - errs = process_args(await request.json or (await request.form).to_dict()) + errs = process_args(await _coerce_request_data()) if errs: return get_json_result(code=RetCode.ARGUMENT_ERROR, message=errs) if inspect.iscoroutinefunction(func): @@ -152,7 +162,7 @@ def validate_request(*args, **kwargs): def not_allowed_parameters(*params): def decorator(func): async def wrapper(*args, **kwargs): - input_arguments = await request.json or (await request.form).to_dict() + input_arguments = await _coerce_request_data() for param in params: if param in input_arguments: return get_json_result(code=RetCode.ARGUMENT_ERROR, message=f"Parameter {param} isn't allowed") diff --git a/common/http_client.py b/common/http_client.py new file mode 100644 index 000000000..2ffbb3bce --- /dev/null +++ b/common/http_client.py @@ -0,0 +1,157 @@ +# Copyright 2025 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. +import asyncio +import logging +import os +import time +from typing import Any, Dict, Optional + +import httpx + +logger = logging.getLogger(__name__) + +# Default knobs; keep conservative to avoid unexpected behavioural changes. +DEFAULT_TIMEOUT = float(os.environ.get("HTTP_CLIENT_TIMEOUT", "15")) +# Align with requests default: follow redirects with a max of 30 unless overridden. +DEFAULT_FOLLOW_REDIRECTS = bool(int(os.environ.get("HTTP_CLIENT_FOLLOW_REDIRECTS", "1"))) +DEFAULT_MAX_REDIRECTS = int(os.environ.get("HTTP_CLIENT_MAX_REDIRECTS", "30")) +DEFAULT_MAX_RETRIES = int(os.environ.get("HTTP_CLIENT_MAX_RETRIES", "2")) +DEFAULT_BACKOFF_FACTOR = float(os.environ.get("HTTP_CLIENT_BACKOFF_FACTOR", "0.5")) +DEFAULT_PROXY = os.environ.get("HTTP_CLIENT_PROXY") +DEFAULT_USER_AGENT = os.environ.get("HTTP_CLIENT_USER_AGENT", "ragflow-http-client") + + +def _clean_headers(headers: Optional[Dict[str, str]], auth_token: Optional[str] = None) -> Optional[Dict[str, str]]: + merged_headers: Dict[str, str] = {} + if DEFAULT_USER_AGENT: + merged_headers["User-Agent"] = DEFAULT_USER_AGENT + if auth_token: + merged_headers["Authorization"] = auth_token + if headers is None: + return merged_headers or None + merged_headers.update({str(k): str(v) for k, v in headers.items() if v is not None}) + return merged_headers or None + + +def _get_delay(backoff_factor: float, attempt: int) -> float: + return backoff_factor * (2**attempt) + + +async def async_request( + method: str, + url: str, + *, + timeout: float | httpx.Timeout | None = None, + follow_redirects: bool | None = None, + max_redirects: Optional[int] = None, + headers: Optional[Dict[str, str]] = None, + auth_token: Optional[str] = None, + retries: Optional[int] = None, + backoff_factor: Optional[float] = None, + proxies: Any = None, + **kwargs: Any, +) -> httpx.Response: + """Lightweight async HTTP wrapper using httpx.AsyncClient with safe defaults.""" + timeout = timeout if timeout is not None else DEFAULT_TIMEOUT + follow_redirects = DEFAULT_FOLLOW_REDIRECTS if follow_redirects is None else follow_redirects + max_redirects = DEFAULT_MAX_REDIRECTS if max_redirects is None else max_redirects + retries = DEFAULT_MAX_RETRIES if retries is None else max(retries, 0) + backoff_factor = DEFAULT_BACKOFF_FACTOR if backoff_factor is None else backoff_factor + headers = _clean_headers(headers, auth_token=auth_token) + proxies = DEFAULT_PROXY if proxies is None else proxies + + async with httpx.AsyncClient( + timeout=timeout, + follow_redirects=follow_redirects, + max_redirects=max_redirects, + proxies=proxies, + ) as client: + last_exc: Exception | None = None + for attempt in range(retries + 1): + try: + start = time.monotonic() + response = await client.request(method=method, url=url, headers=headers, **kwargs) + duration = time.monotonic() - start + logger.debug(f"async_request {method} {url} -> {response.status_code} in {duration:.3f}s") + return response + except httpx.RequestError as exc: + last_exc = exc + if attempt >= retries: + logger.warning(f"async_request exhausted retries for {method} {url}: {exc}") + raise + delay = _get_delay(backoff_factor, attempt) + logger.warning(f"async_request attempt {attempt + 1}/{retries + 1} failed for {method} {url}: {exc}; retrying in {delay:.2f}s") + await asyncio.sleep(delay) + raise last_exc # pragma: no cover + + +def sync_request( + method: str, + url: str, + *, + timeout: float | httpx.Timeout | None = None, + follow_redirects: bool | None = None, + max_redirects: Optional[int] = None, + headers: Optional[Dict[str, str]] = None, + auth_token: Optional[str] = None, + retries: Optional[int] = None, + backoff_factor: Optional[float] = None, + proxies: Any = None, + **kwargs: Any, +) -> httpx.Response: + """Synchronous counterpart to async_request, for CLI/tests or sync contexts.""" + timeout = timeout if timeout is not None else DEFAULT_TIMEOUT + follow_redirects = DEFAULT_FOLLOW_REDIRECTS if follow_redirects is None else follow_redirects + max_redirects = DEFAULT_MAX_REDIRECTS if max_redirects is None else max_redirects + retries = DEFAULT_MAX_RETRIES if retries is None else max(retries, 0) + backoff_factor = DEFAULT_BACKOFF_FACTOR if backoff_factor is None else backoff_factor + headers = _clean_headers(headers, auth_token=auth_token) + proxies = DEFAULT_PROXY if proxies is None else proxies + + with httpx.Client( + timeout=timeout, + follow_redirects=follow_redirects, + max_redirects=max_redirects, + proxies=proxies, + ) as client: + last_exc: Exception | None = None + for attempt in range(retries + 1): + try: + start = time.monotonic() + response = client.request(method=method, url=url, headers=headers, **kwargs) + duration = time.monotonic() - start + logger.debug(f"sync_request {method} {url} -> {response.status_code} in {duration:.3f}s") + return response + except httpx.RequestError as exc: + last_exc = exc + if attempt >= retries: + logger.warning(f"sync_request exhausted retries for {method} {url}: {exc}") + raise + delay = _get_delay(backoff_factor, attempt) + logger.warning(f"sync_request attempt {attempt + 1}/{retries + 1} failed for {method} {url}: {exc}; retrying in {delay:.2f}s") + time.sleep(delay) + raise last_exc # pragma: no cover + + +__all__ = [ + "async_request", + "sync_request", + "DEFAULT_TIMEOUT", + "DEFAULT_FOLLOW_REDIRECTS", + "DEFAULT_MAX_REDIRECTS", + "DEFAULT_MAX_RETRIES", + "DEFAULT_BACKOFF_FACTOR", + "DEFAULT_PROXY", + "DEFAULT_USER_AGENT", +]