refactor:admin pydantic validator

Author: sunlin92

app/api/cms/admin.py

@@ -3,16 +3,24 @@
     :license: MIT, see LICENSE for more details.
 """
 from flask import Blueprint, request
-
 from lin import login_required
+
+from app.api import AuthorizationBearerSecurity, api
 from app.extension.file.local_uploader import LocalUploader
 
 file_api = Blueprint("file", __name__)
 
 
 @file_api.route("", methods=["POST"])
 @login_required
+@api.validate(
+    tags=["文件"],
+    security=[AuthorizationBearerSecurity],
+)
 def post_file():
+    """
+    上传文件
+    """
     files = request.files
     uploader = LocalUploader(files)
     ret = uploader.upload()

app/api/cms/file.py

@@ -1,11 +1,15 @@
 import math
 
 from flask import Blueprint, g
+from lin import DocResponse, Log, db, group_required, permission_meta
 from sqlalchemy import text
 
 from app.api import AuthorizationBearerSecurity, api
-from app.api.cms.schema import LogPageSchema, LogQuerySearchSchema, UsernameListSchema
-from lin import DocResponse, Log, db, group_required, permission_meta
+from app.api.cms.schema.log import (
+    LogPageSchema,
+    LogQuerySearchSchema,
+    UsernameListSchema,
+)
 
 log_api = Blueprint("log", __name__)
 

app/api/cms/log.py

@@ -1,7 +1,6 @@
-from sqlalchemy import func
-
 from lin import User as LinUser
 from lin import db, manager
+from sqlalchemy import func
 
 
 class User(LinUser):

app/api/cms/model/user.py

@@ -1,66 +0,0 @@
-import re
-from datetime import datetime
-from typing import List, Optional
-
-from flask import g
-from pydantic import Field, validator
-
-from lin import BaseModel
-from app.schema import BasePageSchema, datetime_regex
-
-
-class UsernameListSchema(BaseModel):
-    items: List[str]
-
-
-class LogQuerySearchSchema(BaseModel):
-    keyword: Optional[str] = None
-    name: Optional[str] = None
-    start: Optional[str] = Field(None, description="YY-MM-DD HH:MM:SS")
-    end: Optional[str] = Field(None, description="YY-MM-DD HH:MM:SS")
-    count: int = Field(5, gt=0, lt=16, description="0 < count < 16")
-    page: int = 0
-
-    @validator("start", "end")
-    def datetime_match(cls, v, values, **kwargs):
-        if re.match(datetime_regex, v):
-            return v
-        raise ValueError("时间格式有误")
-
-    @staticmethod
-    def offset_handler(req, resp, req_validation_error, instance):
-        g.offset = req.context.query.count * req.context.query.page
-
-
-class LogSchema(BaseModel):
-    message: str
-    user_id: int
-    username: str
-    status_code: int
-    method: str
-    path: str
-    permission: str
-    time: datetime = Field(alias="create_time")
-
-
-class LogPageSchema(BasePageSchema):
-    items: List[LogSchema]
-
-
-class LoginSchema(BaseModel):
-    username: str = Field(description="用户名")
-    password: str = Field(description="密码")
-    captcha: Optional[str] = Field(description="验证码")
-
-
-class AccessTokenSchema(BaseModel):
-    __root__: str = Field(description="access_token")
-
-
-class RefreshTokenSchema(BaseModel):
-    __root__: str = Field(description="refresh_token")
-
-
-class LoginTokenSchema(BaseModel):
-    access_token: AccessTokenSchema
-    refresh_token: RefreshTokenSchema

app/api/cms/schema/__init__.py

@@ -0,0 +1,108 @@
+import re
+from typing import List, Optional
+
+from lin import BaseModel, ParameterError
+from pydantic import EmailStr, 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
+
+
+class AdminGroupSchema(BaseModel):
+    id: int = Field(description="用户组ID")
+    info: str = Field(description="用户组信息")
+    name: str = Field(description="用户组名称")
+
+
+class AdminGroupListSchema(BaseModel):
+    __root__: List[AdminGroupSchema]
+
+
+class AdminUserSchema(BaseModel):
+    id: int = Field(description="用户ID")
+    username: str = Field(description="用户名")
+    email: Optional[EmailStr] = Field(description="邮箱")
+    groups: List[AdminGroupSchema] = Field(description="用户组列表")
+
+
+class AdminUserPageSchema(BasePageSchema):
+    items: List[AdminUserSchema]
+
+
+class GroupQuerySearchSchema(QueryPageSchema):
+    group_id: Optional[int] = Field(gt=0, description="用户组ID")
+
+
+class UpdateUserInfoSchema(GroupIdListSchema):
+    email: EmailStr = Field(description="电子邮件")
+
+
+class PermissionSchema(BaseModel):
+    id: int = Field(description="权限ID")
+    name: str = Field(description="权限名称")
+    module: str = Field(description="权限所属模块")
+    mount: bool = Field(description="是否为挂载权限")
+
+
+class AdminGroupPermissionSchema(AdminGroupSchema):
+    permissions: List[PermissionSchema]
+
+
+class AdminGroupPermissionPageSchema(BasePageSchema):
+    items: List[AdminGroupPermissionSchema]
+
+
+class GroupBaseSchema(BaseModel):
+    name: str = Field(description="用户组名称")
+    info: Optional[str] = Field(description="用户组信息")
+
+
+class CreateGroupSchema(GroupBaseSchema):
+    permission_ids: List[int] = Field(description="权限ID列表")
+
+    @validator("permission_ids", each_item=True)
+    def check_permission_id(cls, v, values, **kwargs):
+        if v <= 0:
+            raise ParameterError("权限ID必须大于0")
+        return v
+
+
+class GroupIdWithPermissionIdListSchema(BaseModel):
+    group_id: int = Field(description="用户组ID")
+    permission_ids: List[int] = Field(description="权限ID列表")
+
+    @validator("permission_ids", each_item=True)
+    def check_permission_id(cls, v, values, **kwargs):
+        if v <= 0:
+            raise ParameterError("权限ID必须大于0")
+        return v

app/api/cms/schema/admin.py

@@ -0,0 +1,40 @@
+import re
+from datetime import datetime
+from typing import List, Optional
+
+from lin import BaseModel
+from pydantic import Field, validator
+
+from app.schema import BasePageSchema, QueryPageSchema, datetime_regex
+
+
+class UsernameListSchema(BaseModel):
+    items: List[str]
+
+
+class LogQuerySearchSchema(QueryPageSchema):
+    keyword: Optional[str] = None
+    name: Optional[str] = None
+    start: Optional[str] = Field(None, description="YY-MM-DD HH:MM:SS")
+    end: Optional[str] = Field(None, description="YY-MM-DD HH:MM:SS")
+
+    @validator("start", "end")
+    def datetime_match(cls, v):
+        if re.match(datetime_regex, v):
+            return v
+        raise ValueError("时间格式有误")
+
+
+class LogSchema(BaseModel):
+    message: str
+    user_id: int
+    username: str
+    status_code: int
+    method: str
+    path: str
+    permission: str
+    time: datetime = Field(alias="create_time")
+
+
+class LogPageSchema(BasePageSchema):
+    items: List[LogSchema]

app/api/cms/schema/log.py

@@ -0,0 +1,20 @@
+from typing import Optional
+
+from lin import BaseModel
+from pydantic import Field
+
+
+class LoginSchema(BaseModel):
+    username: str = Field(description="用户名")
+    password: str = Field(description="密码")
+    captcha: Optional[str] = Field(description="验证码")
+
+
+class LoginTokenSchema(BaseModel):
+    access_token: str = Field(description="access_token")
+    refresh_token: str = Field(description="refresh_token")
+
+
+class CaptchaSchema(BaseModel):
+    image: str = Field("", description="验证码图片base64编码")
+    tag: str = Field("", description="验证码标记码")