refactor:user pydantic validator

Author: sunlin92

app/api/cms/admin.py

@@ -23,16 +23,16 @@
 from sqlalchemy import func
 
 from app.api import AuthorizationBearerSecurity, api
+from app.api.cms.schema import ResetPasswordSchema
 from app.api.cms.schema.admin import (
     AdminGroupListSchema,
     AdminGroupPermissionSchema,
     AdminUserPageSchema,
-    GroupQuerySearchSchema,
-    ResetPasswordSchema,
-    UpdateUserInfoSchema,
     CreateGroupSchema,
     GroupBaseSchema,
     GroupIdWithPermissionIdListSchema,
+    GroupQuerySearchSchema,
+    UpdateUserInfoSchema,
 )
 from app.util.page import get_page_from_query
 

app/api/cms/schema/__init__.py

@@ -0,0 +1,33 @@
+from typing import List, Optional
+
+from lin import BaseModel, ParameterError
+from pydantic import EmailStr, Field, validator
+
+
+class EmailSchema(BaseModel):
+    email: Optional[str] = Field(description="用户邮箱")
+
+    @validator("email")
+    def check_email(cls, v, values, **kwargs):
+        return EmailStr.validate(v) if v else ""
+
+
+class ResetPasswordSchema(BaseModel):
+    new_password: str = Field(description="新密码", min_length=6, max_length=22)
+    confirm_password: str = Field(description="确认密码", min_length=6, max_length=22)
+
+    @validator("confirm_password")
+    def passwords_match(cls, v, values, **kwargs):
+        if v != values["new_password"]:
+            raise ParameterError("两次输入的密码不一致，请输入相同的密码")
+        return v
+
+
+class GroupIdListSchema(BaseModel):
+    group_ids: List[int] = Field(description="用户组ID列表")
+
+    @validator("group_ids", each_item=True)
+    def check_group_id(cls, v, values, **kwargs):
+        if v <= 0:
+            raise ParameterError("用户组ID必须大于0")
+        return v

app/api/cms/schema/admin.py

@@ -1,41 +1,11 @@
-import re
 from typing import List, Optional
 
 from lin import BaseModel, ParameterError
-from pydantic import EmailStr, Field, validator
+from pydantic import Field, validator
 
 from app.schema import BasePageSchema, QueryPageSchema
 
-
-class ResetPasswordSchema(BaseModel):
-    new_password: str = Field(description="新密码", min_length=6, max_length=22)
-    confirm_password: str = Field(description="确认密码", min_length=6, max_length=22)
-
-    @validator("confirm_password", allow_reuse=True)
-    def passwords_match(cls, v, values, **kwargs):
-        if v != values["new_password"]:
-            raise ParameterError("两次输入的密码不一致，请输入相同的密码")
-        return v
-
-
-class GroupIdListSchema(BaseModel):
-    group_ids: List[int] = Field(description="用户组ID列表")
-
-    @validator("group_ids", each_item=True)
-    def check_group_id(cls, v, values, **kwargs):
-        if v <= 0:
-            raise ParameterError("用户组ID必须大于0")
-        return v
-
-
-class RegisterSchema(ResetPasswordSchema, GroupIdListSchema):
-    username: str = Field(description="用户名", min_length=2, max_length=10)
-
-    @validator("username")
-    def check_username(cls, v, values, **kwargs):
-        if not re.match(r"^[a-zA-Z0-9_]{2,10}$", v):
-            raise ParameterError("用户名只能由字母、数字、下划线组成，且长度为2-10位")
-        return v
+from . import EmailSchema, GroupIdListSchema
 
 
 class AdminGroupSchema(BaseModel):
@@ -48,10 +18,9 @@ class AdminGroupListSchema(BaseModel):
     __root__: List[AdminGroupSchema]
 
 
-class AdminUserSchema(BaseModel):
+class AdminUserSchema(EmailSchema):
     id: int = Field(description="用户ID")
     username: str = Field(description="用户名")
-    email: Optional[EmailStr] = Field(description="邮箱")
     groups: List[AdminGroupSchema] = Field(description="用户组列表")
 
 
@@ -63,8 +32,8 @@ class GroupQuerySearchSchema(QueryPageSchema):
     group_id: Optional[int] = Field(gt=0, description="用户组ID")
 
 
-class UpdateUserInfoSchema(GroupIdListSchema):
-    email: EmailStr = Field(description="电子邮件")
+class UpdateUserInfoSchema(GroupIdListSchema, EmailSchema):
+    pass
 
 
 class PermissionSchema(BaseModel):

app/api/cms/schema/user.py

@@ -1,7 +1,10 @@
-from typing import Optional
+import re
+from typing import List, Optional
 
-from lin import BaseModel
-from pydantic import Field
+from lin import BaseModel, ParameterError
+from pydantic import AnyHttpUrl, Field, validator
+
+from . import EmailSchema, GroupIdListSchema, ResetPasswordSchema
 
 
 class LoginSchema(BaseModel):
@@ -18,3 +21,48 @@ class LoginTokenSchema(BaseModel):
 class CaptchaSchema(BaseModel):
     image: str = Field("", description="验证码图片base64编码")
     tag: str = Field("", description="验证码标记码")
+
+
+class PermissionNameSchema(BaseModel):
+    name: str = Field(description="权限名称")
+
+
+class PermissionModuleSchema(BaseModel):
+    module: List[PermissionNameSchema] = Field(description="权限模块")
+
+
+class UserBaseInfoSchema(EmailSchema):
+    nickname: Optional[str] = Field(description="用户昵称", min_length=2, max_length=10)
+    avatar: Optional[AnyHttpUrl] = Field(description="头像url")
+
+
+class UserSchema(UserBaseInfoSchema):
+    id: int = Field(description="用户id")
+    username: str = Field(description="用户名")
+
+
+class UserPermissionSchema(UserSchema):
+    admin: bool = Field(description="是否是管理员")
+    permissions: List[PermissionModuleSchema] = Field(description="用户权限")
+
+
+class ChangePasswordSchema(ResetPasswordSchema):
+    old_password: str = Field(description="旧密码")
+
+
+class UserRegisterSchema(GroupIdListSchema, EmailSchema):
+    username: str = Field(description="用户名", min_length=2, max_length=10)
+    password: str = Field(description="密码", min_length=6, max_length=22)
+    confirm_password: str = Field(description="确认密码", min_length=6, max_length=22)
+
+    @validator("confirm_password")
+    def passwords_match(cls, v, values, **kwargs):
+        if v != values["password"]:
+            raise ParameterError("两次输入的密码不一致，请输入相同的密码")
+        return v
+
+    @validator("username")
+    def check_username(cls, v, values, **kwargs):
+        if not re.match(r"^[a-zA-Z0-9_]{2,10}$", v):
+            raise ParameterError("用户名只能由字母、数字、下划线组成，且长度为2-10位")
+        return v

app/api/cms/user.py

@@ -4,7 +4,7 @@
     :copyright: © 2020 by the Lin team.
     :license: MIT, see LICENSE for more details.
 """
-from flask import Blueprint, current_app, request
+from flask import Blueprint, current_app, g, request
 from flask_jwt_extended import (
     create_access_token,
     create_refresh_token,
@@ -32,12 +32,15 @@
 
 from app.api import AuthorizationBearerSecurity, api
 from app.api.cms.exception import RefreshFailed
-from app.api.cms.schema.user import CaptchaSchema, LoginSchema, LoginTokenSchema
-from app.api.cms.validator import (
-    ChangePasswordForm,
-    LoginForm,
-    RegisterForm,
-    UpdateInfoForm,
+from app.api.cms.schema.user import (
+    CaptchaSchema,
+    ChangePasswordSchema,
+    LoginSchema,
+    LoginTokenSchema,
+    UserBaseInfoSchema,
+    UserPermissionSchema,
+    UserRegisterSchema,
+    UserSchema,
 )
 from app.util.captcha import CaptchaTool
 from app.util.common import split_group
@@ -52,19 +55,38 @@
 @api.validate(
     tags=["用户"],
     security=[AuthorizationBearerSecurity],
-    resp=DocResponse(Success("用户创建成功")),
+    resp=DocResponse(Success("用户创建成功"), Duplicated("字段重复，请重新输入")),
 )
-def register():
+def register(json: UserRegisterSchema):
     """
     注册新用户
     """
-    form = RegisterForm().validate_for_api()
-    if manager.user_model.count_by_username(form.username.data) > 0:
+    if manager.user_model.count_by_username(g.username) > 0:
         raise Duplicated("用户名重复，请重新输入")  # type: ignore
-    if form.email.data and form.email.data.strip() != "":
-        if manager.user_model.count_by_email(form.email.data) > 0:
+    if g.email and g.email.strip() != "":
+        if manager.user_model.count_by_email(g.email) > 0:
             raise Duplicated("注册邮箱重复，请重新输入")  # type: ignore
-    _register_user(form)
+    # create a user
+    with db.auto_commit():
+        user = manager.user_model()
+        user.username = g.username
+        if g.email and g.email.strip() != "":
+            user.email = g.email
+        db.session.add(user)
+        db.session.flush()
+        user.password = g.password
+        group_ids = g.group_ids
+        # 如果没传分组数据，则将其设定为 guest 分组
+        if len(group_ids) == 0:
+            from lin import GroupLevelEnum
+
+            group_ids = [GroupLevelEnum.GUEST.value]
+        for group_id in group_ids:
+            user_group = manager.user_group_model()
+            user_group.user_id = user.id
+            user_group.group_id = group_id
+            db.session.add(user_group)
+
     return Success("用户创建成功")  # type: ignore
 
 
@@ -74,16 +96,15 @@ def login(json: LoginSchema):
     """
     用户登录
     """
-    form = LoginForm().validate_for_api()
     # 校对验证码
     if current_app.config.get("LOGIN_CAPTCHA"):
         tag = request.headers.get("tag")
         secret_key = current_app.config.get("SECRET_KEY")
         serializer = JWSSerializer(secret_key)
-        if form.captcha.data != serializer.loads(tag):
+        if g.captcha != serializer.loads(tag):
             raise Failed("验证码校验失败")  # type: ignore
 
-    user = manager.user_model.verify(form.username.data, form.password.data)
+    user = manager.user_model.verify(g.username, g.password)
     # 用户未登录，此处不能用装饰器记录日志
     Log.create_log(
         message=f"{user.username}登录成功获取了令牌",
@@ -105,29 +126,26 @@ def login(json: LoginSchema):
 @api.validate(
     tags=["用户"],
     security=[AuthorizationBearerSecurity],
+    resp=DocResponse(Success("用户信息更新成功"), ParameterError("邮箱已被注册，请重新输入邮箱")),
 )
-def update():
+def update(json: UserBaseInfoSchema):
     """
     更新用户信息
     """
-    form = UpdateInfoForm().validate_for_api()
     user = get_current_user()
-    email = form.email.data
-    nickname = form.nickname.data
-    avatar = form.avatar.data
 
-    if email and user.email != email:
-        exists = manager.user_model.get(email=form.email.data)
+    if g.email and user.email != g.email:
+        exists = manager.user_model.get(email=g.email)
         if exists:
             raise ParameterError("邮箱已被注册，请重新输入邮箱")
     with db.auto_commit():
-        if email:
-            user.email = form.email.data
-        if nickname:
-            user.nickname = form.nickname.data
-        if avatar:
-            user._avatar = form.avatar.data
-    return Success("操作成功")
+        if g.email:
+            user.email = g.email
+        if g.nickname:
+            user.nickname = g.nickname
+        if g.avatar:
+            user._avatar = g.avatar
+    return Success("用户信息更新成功")
 
 
 @user_api.route("/change_password", methods=["PUT"])
@@ -137,14 +155,14 @@ def update():
 @api.validate(
     tags=["用户"],
     security=[AuthorizationBearerSecurity],
+    resp=DocResponse(Success("密码修改成功"), Failed("密码修改失败")),
 )
-def change_password():
+def change_password(json: ChangePasswordSchema):
     """
     修改密码
     """
-    form = ChangePasswordForm().validate_for_api()
     user = get_current_user()
-    ok = user.change_password(form.old_password.data, form.new_password.data)
+    ok = user.change_password(g.old_password, g.new_password)
     if ok:
         db.session.commit()
         return Success("密码修改成功")
@@ -158,6 +176,7 @@ def change_password():
 @api.validate(
     tags=["用户"],
     security=[AuthorizationBearerSecurity],
+    resp=DocResponse(r=UserSchema),
 )
 def get_information():
     """
@@ -197,6 +216,7 @@ def refresh():
 @api.validate(
     tags=["用户"],
     security=[AuthorizationBearerSecurity],
+    resp=DocResponse(r=UserPermissionSchema),
 )
 def get_allowed_apis():
     """
@@ -218,28 +238,6 @@ def get_allowed_apis():
     return user
 
 
-def _register_user(form: RegisterForm):
-    with db.auto_commit():
-        user = manager.user_model()
-        user.username = form.username.data
-        if form.email.data and form.email.data.strip() != "":
-            user.email = form.email.data
-        db.session.add(user)
-        db.session.flush()
-        user.password = form.password.data
-        group_ids = form.group_ids.data
-        # 如果没传分组数据，则将其设定为 guest 分组
-        if len(group_ids) == 0:
-            from lin import GroupLevelEnum
-
-            group_ids = [GroupLevelEnum.GUEST.value]
-        for group_id in group_ids:
-            user_group = manager.user_group_model()
-            user_group.user_id = user.id
-            user_group.group_id = group_id
-            db.session.add(user_group)
-
-
 @user_api.route("/captcha", methods=["GET", "POST"])
 @api.validate(
     resp=DocResponse(r=CaptchaSchema),

pyproject.toml

@@ -11,7 +11,7 @@ default = true
 
 [tool.poetry.dependencies]
 python = "^3.8"
-Lin-CMS = "^0.4.4"
+Lin-CMS = "^0.4.5"
 pillow = "^9.0.0"
 flask-cors = "^3.0.10"
 gunicorn = "^20.1.0"

requirements-dev.txt

@@ -1,5 +1,5 @@
 python==3.8
-Lin-CMS==0.4.4
+Lin-CMS==0.4.5
 pillow==9.0.0
 flask-cors==3.0.10
 gunicorn==20.1.0