refactor: port reqparse to Pydantic model (#28949)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

api/controllers/console/auth/activate.py

@@ -1,28 +1,53 @@
 from flask import request
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, fields
+from pydantic import BaseModel, Field, field_validator
 
 from constants.languages import supported_language
 from controllers.console import console_ns
 from controllers.console.error import AlreadyActivateError
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
-from libs.helper import StrLen, email, extract_remote_ip, timezone
+from libs.helper import EmailStr, extract_remote_ip, timezone
 from models import AccountStatus
 from services.account_service import AccountService, RegisterService
 
-active_check_parser = (
-    reqparse.RequestParser()
-    .add_argument("workspace_id", type=str, required=False, nullable=True, location="args", help="Workspace ID")
-    .add_argument("email", type=email, required=False, nullable=True, location="args", help="Email address")
-    .add_argument("token", type=str, required=True, nullable=False, location="args", help="Activation token")
-)
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
+
+class ActivateCheckQuery(BaseModel):
+    workspace_id: str | None = Field(default=None)
+    email: EmailStr | None = Field(default=None)
+    token: str
+
+
+class ActivatePayload(BaseModel):
+    workspace_id: str | None = Field(default=None)
+    email: EmailStr | None = Field(default=None)
+    token: str
+    name: str = Field(..., max_length=30)
+    interface_language: str = Field(...)
+    timezone: str = Field(...)
+
+    @field_validator("interface_language")
+    @classmethod
+    def validate_lang(cls, value: str) -> str:
+        return supported_language(value)
+
+    @field_validator("timezone")
+    @classmethod
+    def validate_tz(cls, value: str) -> str:
+        return timezone(value)
+
+
+for model in (ActivateCheckQuery, ActivatePayload):
+    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
 
 @console_ns.route("/activate/check")
 class ActivateCheckApi(Resource):
     @console_ns.doc("check_activation_token")
     @console_ns.doc(description="Check if activation token is valid")
-    @console_ns.expect(active_check_parser)
+    @console_ns.expect(console_ns.models[ActivateCheckQuery.__name__])
     @console_ns.response(
         200,
         "Success",
@@ -35,11 +60,11 @@ class ActivateCheckApi(Resource):
         ),
     )
     def get(self):
-        args = active_check_parser.parse_args()
+        args = ActivateCheckQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
-        workspaceId = args["workspace_id"]
-        reg_email = args["email"]
-        token = args["token"]
+        workspaceId = args.workspace_id
+        reg_email = args.email
+        token = args.token
 
         invitation = RegisterService.get_invitation_if_token_valid(workspaceId, reg_email, token)
         if invitation:
@@ -56,22 +81,11 @@ class ActivateCheckApi(Resource):
             return {"is_valid": False}
 
 
-active_parser = (
-    reqparse.RequestParser()
-    .add_argument("workspace_id", type=str, required=False, nullable=True, location="json")
-    .add_argument("email", type=email, required=False, nullable=True, location="json")
-    .add_argument("token", type=str, required=True, nullable=False, location="json")
-    .add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
-    .add_argument("interface_language", type=supported_language, required=True, nullable=False, location="json")
-    .add_argument("timezone", type=timezone, required=True, nullable=False, location="json")
-)
-
-
 @console_ns.route("/activate")
 class ActivateApi(Resource):
     @console_ns.doc("activate_account")
     @console_ns.doc(description="Activate account with invitation token")
-    @console_ns.expect(active_parser)
+    @console_ns.expect(console_ns.models[ActivatePayload.__name__])
     @console_ns.response(
         200,
         "Account activated successfully",
@@ -85,19 +99,19 @@ class ActivateApi(Resource):
     )
     @console_ns.response(400, "Already activated or invalid token")
     def post(self):
-        args = active_parser.parse_args()
+        args = ActivatePayload.model_validate(console_ns.payload)
 
-        invitation = RegisterService.get_invitation_if_token_valid(args["workspace_id"], args["email"], args["token"])
+        invitation = RegisterService.get_invitation_if_token_valid(args.workspace_id, args.email, args.token)
         if invitation is None:
             raise AlreadyActivateError()
 
-        RegisterService.revoke_token(args["workspace_id"], args["email"], args["token"])
+        RegisterService.revoke_token(args.workspace_id, args.email, args.token)
 
         account = invitation["account"]
-        account.name = args["name"]
+        account.name = args.name
 
-        account.interface_language = args["interface_language"]
-        account.timezone = args["timezone"]
+        account.interface_language = args.interface_language
+        account.timezone = args.timezone
         account.interface_theme = "light"
         account.status = AccountStatus.ACTIVE
         account.initialized_at = naive_utc_now()