add doc (#28016)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

api/controllers/console/datasets/rag_pipeline/datasource_auth.py

@@ -3,7 +3,7 @@ from flask_restx import Resource, reqparse
 from werkzeug.exceptions import Forbidden, NotFound
 
 from configs import dify_config
-from controllers.console import console_ns
+from controllers.console import api, console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from core.model_runtime.errors.validate import CredentialsValidateFailedError
 from core.model_runtime.utils.encoders import jsonable_encoder
@@ -121,8 +121,16 @@ class DatasourceOAuthCallback(Resource):
         return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
 
 
+parser_datasource = (
+    reqparse.RequestParser()
+    .add_argument("name", type=StrLen(max_length=100), required=False, nullable=True, location="json", default=None)
+    .add_argument("credentials", type=dict, required=True, nullable=False, location="json")
+)
+
+
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>")
 class DatasourceAuth(Resource):
+    @api.expect(parser_datasource)
     @setup_required
     @login_required
     @account_initialization_required
@@ -130,14 +138,7 @@ class DatasourceAuth(Resource):
     def post(self, provider_id: str):
         _, current_tenant_id = current_account_with_tenant()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument(
-                "name", type=StrLen(max_length=100), required=False, nullable=True, location="json", default=None
-            )
-            .add_argument("credentials", type=dict, required=True, nullable=False, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_datasource.parse_args()
         datasource_provider_id = DatasourceProviderID(provider_id)
         datasource_provider_service = DatasourceProviderService()
 
@@ -168,8 +169,14 @@ class DatasourceAuth(Resource):
         return {"result": datasources}, 200
 
 
+parser_datasource_delete = reqparse.RequestParser().add_argument(
+    "credential_id", type=str, required=True, nullable=False, location="json"
+)
+
+
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/delete")
 class DatasourceAuthDeleteApi(Resource):
+    @api.expect(parser_datasource_delete)
     @setup_required
     @login_required
     @account_initialization_required
@@ -181,10 +188,7 @@ class DatasourceAuthDeleteApi(Resource):
         plugin_id = datasource_provider_id.plugin_id
         provider_name = datasource_provider_id.provider_name
 
-        parser = reqparse.RequestParser().add_argument(
-            "credential_id", type=str, required=True, nullable=False, location="json"
-        )
-        args = parser.parse_args()
+        args = parser_datasource_delete.parse_args()
         datasource_provider_service = DatasourceProviderService()
         datasource_provider_service.remove_datasource_credentials(
             tenant_id=current_tenant_id,
@@ -195,8 +199,17 @@ class DatasourceAuthDeleteApi(Resource):
         return {"result": "success"}, 200
 
 
+parser_datasource_update = (
+    reqparse.RequestParser()
+    .add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+    .add_argument("name", type=StrLen(max_length=100), required=False, nullable=True, location="json")
+    .add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+)
+
+
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/update")
 class DatasourceAuthUpdateApi(Resource):
+    @api.expect(parser_datasource_update)
     @setup_required
     @login_required
     @account_initialization_required
@@ -205,13 +218,7 @@ class DatasourceAuthUpdateApi(Resource):
         _, current_tenant_id = current_account_with_tenant()
 
         datasource_provider_id = DatasourceProviderID(provider_id)
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("credentials", type=dict, required=False, nullable=True, location="json")
-            .add_argument("name", type=StrLen(max_length=100), required=False, nullable=True, location="json")
-            .add_argument("credential_id", type=str, required=True, nullable=False, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_datasource_update.parse_args()
 
         datasource_provider_service = DatasourceProviderService()
         datasource_provider_service.update_datasource_credentials(
@@ -251,8 +258,16 @@ class DatasourceHardCodeAuthListApi(Resource):
         return {"result": jsonable_encoder(datasources)}, 200
 
 
+parser_datasource_custom = (
+    reqparse.RequestParser()
+    .add_argument("client_params", type=dict, required=False, nullable=True, location="json")
+    .add_argument("enable_oauth_custom_client", type=bool, required=False, nullable=True, location="json")
+)
+
+
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/custom-client")
 class DatasourceAuthOauthCustomClient(Resource):
+    @api.expect(parser_datasource_custom)
     @setup_required
     @login_required
     @account_initialization_required
@@ -260,12 +275,7 @@ class DatasourceAuthOauthCustomClient(Resource):
     def post(self, provider_id: str):
         _, current_tenant_id = current_account_with_tenant()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("client_params", type=dict, required=False, nullable=True, location="json")
-            .add_argument("enable_oauth_custom_client", type=bool, required=False, nullable=True, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_datasource_custom.parse_args()
         datasource_provider_id = DatasourceProviderID(provider_id)
         datasource_provider_service = DatasourceProviderService()
         datasource_provider_service.setup_oauth_custom_client_params(
@@ -291,8 +301,12 @@ class DatasourceAuthOauthCustomClient(Resource):
         return {"result": "success"}, 200
 
 
+parser_default = reqparse.RequestParser().add_argument("id", type=str, required=True, nullable=False, location="json")
+
+
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/default")
 class DatasourceAuthDefaultApi(Resource):
+    @api.expect(parser_default)
     @setup_required
     @login_required
     @account_initialization_required
@@ -300,8 +314,7 @@ class DatasourceAuthDefaultApi(Resource):
     def post(self, provider_id: str):
         _, current_tenant_id = current_account_with_tenant()
 
-        parser = reqparse.RequestParser().add_argument("id", type=str, required=True, nullable=False, location="json")
-        args = parser.parse_args()
+        args = parser_default.parse_args()
         datasource_provider_id = DatasourceProviderID(provider_id)
         datasource_provider_service = DatasourceProviderService()
         datasource_provider_service.set_default_datasource_provider(
@@ -312,8 +325,16 @@ class DatasourceAuthDefaultApi(Resource):
         return {"result": "success"}, 200
 
 
+parser_update_name = (
+    reqparse.RequestParser()
+    .add_argument("name", type=StrLen(max_length=100), required=True, nullable=False, location="json")
+    .add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+)
+
+
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/update-name")
 class DatasourceUpdateProviderNameApi(Resource):
+    @api.expect(parser_update_name)
     @setup_required
     @login_required
     @account_initialization_required
@@ -321,12 +342,7 @@ class DatasourceUpdateProviderNameApi(Resource):
     def post(self, provider_id: str):
         _, current_tenant_id = current_account_with_tenant()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("name", type=StrLen(max_length=100), required=True, nullable=False, location="json")
-            .add_argument("credential_id", type=str, required=True, nullable=False, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_update_name.parse_args()
         datasource_provider_id = DatasourceProviderID(provider_id)
         datasource_provider_service = DatasourceProviderService()
         datasource_provider_service.update_datasource_provider_name(

api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py

@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
-from controllers.console import console_ns
+from controllers.console import api, console_ns
 from controllers.console.app.error import (
     ConversationCompletedError,
     DraftWorkflowNotExist,
@@ -148,8 +148,12 @@ class DraftRagPipelineApi(Resource):
         }
 
 
+parser_run = reqparse.RequestParser().add_argument("inputs", type=dict, location="json")
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/iteration/nodes/<string:node_id>/run")
 class RagPipelineDraftRunIterationNodeApi(Resource):
+    @api.expect(parser_run)
     @setup_required
     @login_required
     @account_initialization_required
@@ -162,8 +166,7 @@ class RagPipelineDraftRunIterationNodeApi(Resource):
         # The role of the current user in the ta table must be admin, owner, or editor
         current_user, _ = current_account_with_tenant()
 
-        parser = reqparse.RequestParser().add_argument("inputs", type=dict, location="json")
-        args = parser.parse_args()
+        args = parser_run.parse_args()
 
         try:
             response = PipelineGenerateService.generate_single_iteration(
@@ -184,6 +187,7 @@ class RagPipelineDraftRunIterationNodeApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/loop/nodes/<string:node_id>/run")
 class RagPipelineDraftRunLoopNodeApi(Resource):
+    @api.expect(parser_run)
     @setup_required
     @login_required
     @account_initialization_required
@@ -197,8 +201,7 @@ class RagPipelineDraftRunLoopNodeApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = reqparse.RequestParser().add_argument("inputs", type=dict, location="json")
-        args = parser.parse_args()
+        args = parser_run.parse_args()
 
         try:
             response = PipelineGenerateService.generate_single_loop(
@@ -217,8 +220,18 @@ class RagPipelineDraftRunLoopNodeApi(Resource):
             raise InternalServerError()
 
 
+parser_draft_run = (
+    reqparse.RequestParser()
+    .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+    .add_argument("datasource_type", type=str, required=True, location="json")
+    .add_argument("datasource_info_list", type=list, required=True, location="json")
+    .add_argument("start_node_id", type=str, required=True, location="json")
+)
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/run")
 class DraftRagPipelineRunApi(Resource):
+    @api.expect(parser_draft_run)
     @setup_required
     @login_required
     @account_initialization_required
@@ -232,14 +245,7 @@ class DraftRagPipelineRunApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-            .add_argument("datasource_type", type=str, required=True, location="json")
-            .add_argument("datasource_info_list", type=list, required=True, location="json")
-            .add_argument("start_node_id", type=str, required=True, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_draft_run.parse_args()
 
         try:
             response = PipelineGenerateService.generate(
@@ -255,8 +261,21 @@ class DraftRagPipelineRunApi(Resource):
             raise InvokeRateLimitHttpError(ex.description)
 
 
+parser_published_run = (
+    reqparse.RequestParser()
+    .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+    .add_argument("datasource_type", type=str, required=True, location="json")
+    .add_argument("datasource_info_list", type=list, required=True, location="json")
+    .add_argument("start_node_id", type=str, required=True, location="json")
+    .add_argument("is_preview", type=bool, required=True, location="json", default=False)
+    .add_argument("response_mode", type=str, required=True, location="json", default="streaming")
+    .add_argument("original_document_id", type=str, required=False, location="json")
+)
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/run")
 class PublishedRagPipelineRunApi(Resource):
+    @api.expect(parser_published_run)
     @setup_required
     @login_required
     @account_initialization_required
@@ -270,17 +289,7 @@ class PublishedRagPipelineRunApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-            .add_argument("datasource_type", type=str, required=True, location="json")
-            .add_argument("datasource_info_list", type=list, required=True, location="json")
-            .add_argument("start_node_id", type=str, required=True, location="json")
-            .add_argument("is_preview", type=bool, required=True, location="json", default=False)
-            .add_argument("response_mode", type=str, required=True, location="json", default="streaming")
-            .add_argument("original_document_id", type=str, required=False, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_published_run.parse_args()
 
         streaming = args["response_mode"] == "streaming"
 
@@ -381,8 +390,17 @@ class PublishedRagPipelineRunApi(Resource):
 #
 #         return result
 #
+parser_rag_run = (
+    reqparse.RequestParser()
+    .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+    .add_argument("datasource_type", type=str, required=True, location="json")
+    .add_argument("credential_id", type=str, required=False, location="json")
+)
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/datasource/nodes/<string:node_id>/run")
 class RagPipelinePublishedDatasourceNodeRunApi(Resource):
+    @api.expect(parser_rag_run)
     @setup_required
     @login_required
     @account_initialization_required
@@ -396,13 +414,7 @@ class RagPipelinePublishedDatasourceNodeRunApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-            .add_argument("datasource_type", type=str, required=True, location="json")
-            .add_argument("credential_id", type=str, required=False, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_rag_run.parse_args()
 
         inputs = args.get("inputs")
         if inputs is None:
@@ -429,6 +441,7 @@ class RagPipelinePublishedDatasourceNodeRunApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/datasource/nodes/<string:node_id>/run")
 class RagPipelineDraftDatasourceNodeRunApi(Resource):
+    @api.expect(parser_rag_run)
     @setup_required
     @login_required
     @account_initialization_required
@@ -442,13 +455,7 @@ class RagPipelineDraftDatasourceNodeRunApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-            .add_argument("datasource_type", type=str, required=True, location="json")
-            .add_argument("credential_id", type=str, required=False, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_rag_run.parse_args()
 
         inputs = args.get("inputs")
         if inputs is None:
@@ -473,8 +480,14 @@ class RagPipelineDraftDatasourceNodeRunApi(Resource):
         )
 
 
+parser_run_api = reqparse.RequestParser().add_argument(
+    "inputs", type=dict, required=True, nullable=False, location="json"
+)
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/run")
 class RagPipelineDraftNodeRunApi(Resource):
+    @api.expect(parser_run_api)
     @setup_required
     @login_required
     @account_initialization_required
@@ -489,10 +502,7 @@ class RagPipelineDraftNodeRunApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = reqparse.RequestParser().add_argument(
-            "inputs", type=dict, required=True, nullable=False, location="json"
-        )
-        args = parser.parse_args()
+        args = parser_run_api.parse_args()
 
         inputs = args.get("inputs")
         if inputs == None:
@@ -607,8 +617,12 @@ class DefaultRagPipelineBlockConfigsApi(Resource):
         return rag_pipeline_service.get_default_block_configs()
 
 
+parser_default = reqparse.RequestParser().add_argument("q", type=str, location="args")
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/default-workflow-block-configs/<string:block_type>")
 class DefaultRagPipelineBlockConfigApi(Resource):
+    @api.expect(parser_default)
     @setup_required
     @login_required
     @account_initialization_required
@@ -622,8 +636,7 @@ class DefaultRagPipelineBlockConfigApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = reqparse.RequestParser().add_argument("q", type=str, location="args")
-        args = parser.parse_args()
+        args = parser_default.parse_args()
 
         q = args.get("q")
 
@@ -639,8 +652,18 @@ class DefaultRagPipelineBlockConfigApi(Resource):
         return rag_pipeline_service.get_default_block_config(node_type=block_type, filters=filters)
 
 
+parser_wf = (
+    reqparse.RequestParser()
+    .add_argument("page", type=inputs.int_range(1, 99999), required=False, default=1, location="args")
+    .add_argument("limit", type=inputs.int_range(1, 100), required=False, default=10, location="args")
+    .add_argument("user_id", type=str, required=False, location="args")
+    .add_argument("named_only", type=inputs.boolean, required=False, default=False, location="args")
+)
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows")
 class PublishedAllRagPipelineApi(Resource):
+    @api.expect(parser_wf)
     @setup_required
     @login_required
     @account_initialization_required
@@ -654,16 +677,9 @@ class PublishedAllRagPipelineApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("page", type=inputs.int_range(1, 99999), required=False, default=1, location="args")
-            .add_argument("limit", type=inputs.int_range(1, 100), required=False, default=20, location="args")
-            .add_argument("user_id", type=str, required=False, location="args")
-            .add_argument("named_only", type=inputs.boolean, required=False, default=False, location="args")
-        )
-        args = parser.parse_args()
-        page = int(args.get("page", 1))
-        limit = int(args.get("limit", 10))
+        args = parser_wf.parse_args()
+        page = args["page"]
+        limit = args["limit"]
         user_id = args.get("user_id")
         named_only = args.get("named_only", False)
 
@@ -691,8 +707,16 @@ class PublishedAllRagPipelineApi(Resource):
             }
 
 
+parser_wf_id = (
+    reqparse.RequestParser()
+    .add_argument("marked_name", type=str, required=False, location="json")
+    .add_argument("marked_comment", type=str, required=False, location="json")
+)
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>")
 class RagPipelineByIdApi(Resource):
+    @api.expect(parser_wf_id)
     @setup_required
     @login_required
     @account_initialization_required
@@ -707,19 +731,13 @@ class RagPipelineByIdApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("marked_name", type=str, required=False, location="json")
-            .add_argument("marked_comment", type=str, required=False, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_wf_id.parse_args()
 
         # Validate name and comment length
         if args.marked_name and len(args.marked_name) > 20:
             raise ValueError("Marked name cannot exceed 20 characters")
         if args.marked_comment and len(args.marked_comment) > 100:
             raise ValueError("Marked comment cannot exceed 100 characters")
-        args = parser.parse_args()
 
         # Prepare update data
         update_data = {}
@@ -752,8 +770,12 @@ class RagPipelineByIdApi(Resource):
         return workflow
 
 
+parser_parameters = reqparse.RequestParser().add_argument("node_id", type=str, required=True, location="args")
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/processing/parameters")
 class PublishedRagPipelineSecondStepApi(Resource):
+    @api.expect(parser_parameters)
     @setup_required
     @login_required
     @account_initialization_required
@@ -763,8 +785,7 @@ class PublishedRagPipelineSecondStepApi(Resource):
         """
         Get second step parameters of rag pipeline
         """
-        parser = reqparse.RequestParser().add_argument("node_id", type=str, required=True, location="args")
-        args = parser.parse_args()
+        args = parser_parameters.parse_args()
         node_id = args.get("node_id")
         if not node_id:
             raise ValueError("Node ID is required")
@@ -777,6 +798,7 @@ class PublishedRagPipelineSecondStepApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/pre-processing/parameters")
 class PublishedRagPipelineFirstStepApi(Resource):
+    @api.expect(parser_parameters)
     @setup_required
     @login_required
     @account_initialization_required
@@ -786,8 +808,7 @@ class PublishedRagPipelineFirstStepApi(Resource):
         """
         Get first step parameters of rag pipeline
         """
-        parser = reqparse.RequestParser().add_argument("node_id", type=str, required=True, location="args")
-        args = parser.parse_args()
+        args = parser_parameters.parse_args()
         node_id = args.get("node_id")
         if not node_id:
             raise ValueError("Node ID is required")
@@ -800,6 +821,7 @@ class PublishedRagPipelineFirstStepApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/pre-processing/parameters")
 class DraftRagPipelineFirstStepApi(Resource):
+    @api.expect(parser_parameters)
     @setup_required
     @login_required
     @account_initialization_required
@@ -809,8 +831,7 @@ class DraftRagPipelineFirstStepApi(Resource):
         """
         Get first step parameters of rag pipeline
         """
-        parser = reqparse.RequestParser().add_argument("node_id", type=str, required=True, location="args")
-        args = parser.parse_args()
+        args = parser_parameters.parse_args()
         node_id = args.get("node_id")
         if not node_id:
             raise ValueError("Node ID is required")
@@ -823,6 +844,7 @@ class DraftRagPipelineFirstStepApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/processing/parameters")
 class DraftRagPipelineSecondStepApi(Resource):
+    @api.expect(parser_parameters)
     @setup_required
     @login_required
     @account_initialization_required
@@ -832,8 +854,7 @@ class DraftRagPipelineSecondStepApi(Resource):
         """
         Get second step parameters of rag pipeline
         """
-        parser = reqparse.RequestParser().add_argument("node_id", type=str, required=True, location="args")
-        args = parser.parse_args()
+        args = parser_parameters.parse_args()
         node_id = args.get("node_id")
         if not node_id:
             raise ValueError("Node ID is required")
@@ -845,8 +866,16 @@ class DraftRagPipelineSecondStepApi(Resource):
         }
 
 
+parser_wf_run = (
+    reqparse.RequestParser()
+    .add_argument("last_id", type=uuid_value, location="args")
+    .add_argument("limit", type=int_range(1, 100), required=False, default=20, location="args")
+)
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs")
 class RagPipelineWorkflowRunListApi(Resource):
+    @api.expect(parser_wf_run)
     @setup_required
     @login_required
     @account_initialization_required
@@ -856,12 +885,7 @@ class RagPipelineWorkflowRunListApi(Resource):
         """
         Get workflow run list
         """
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("last_id", type=uuid_value, location="args")
-            .add_argument("limit", type=int_range(1, 100), required=False, default=20, location="args")
-        )
-        args = parser.parse_args()
+        args = parser_wf_run.parse_args()
 
         rag_pipeline_service = RagPipelineService()
         result = rag_pipeline_service.get_rag_pipeline_paginate_workflow_runs(pipeline=pipeline, args=args)
@@ -961,8 +985,18 @@ class RagPipelineTransformApi(Resource):
         return result
 
 
+parser_var = (
+    reqparse.RequestParser()
+    .add_argument("datasource_type", type=str, required=True, location="json")
+    .add_argument("datasource_info", type=dict, required=True, location="json")
+    .add_argument("start_node_id", type=str, required=True, location="json")
+    .add_argument("start_node_title", type=str, required=True, location="json")
+)
+
+
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/datasource/variables-inspect")
 class RagPipelineDatasourceVariableApi(Resource):
+    @api.expect(parser_var)
     @setup_required
     @login_required
     @account_initialization_required
@@ -974,14 +1008,7 @@ class RagPipelineDatasourceVariableApi(Resource):
         Set datasource variables
         """
         current_user, _ = current_account_with_tenant()
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("datasource_type", type=str, required=True, location="json")
-            .add_argument("datasource_info", type=dict, required=True, location="json")
-            .add_argument("start_node_id", type=str, required=True, location="json")
-            .add_argument("start_node_title", type=str, required=True, location="json")
-        )
-        args = parser.parse_args()
+        args = parser_var.parse_args()
 
         rag_pipeline_service = RagPipelineService()
         workflow_node_execution = rag_pipeline_service.set_datasource_variables(