更新架构并修复问题 (#36)

Author: wu-clan

.pre-commit-config.yaml

@@ -8,7 +8,7 @@ repos:
       - id: check-toml
 
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.9.5
+    rev: v0.11.4
     hooks:
       - id: ruff
         args:
@@ -18,13 +18,12 @@ repos:
           - '--unsafe-fixes'
       - id: ruff-format
 
-  - repo: https://github.com/pdm-project/pdm
-    rev: 2.22.3
+  - repo: https://github.com/astral-sh/uv-pre-commit
+    rev: 0.6.14
     hooks:
-      - id: pdm-export
+      - id: uv-lock
+      - id: uv-export
         args:
           - '-o'
           - 'requirements.txt'
-          - '--without-hashes'
-        files: ^pdm.lock$
-      - id: pdm-lock-check
+          - '--no-hashes'

README.md

@@ -52,7 +52,7 @@
    fastapi dev main.py
    ```
 
-9. 浏览器访问: http://127.0.0.1:8000/api/v1/docs
+9. 浏览器访问: http://127.0.0.1:8000/docs
 
 ---
 

backend/app/admin/api/v1/auth/auth.py

@@ -4,7 +4,7 @@
 from fastapi.security import OAuth2PasswordRequestForm
 
 from backend.app.admin.schema.token import GetLoginToken, GetSwaggerToken
-from backend.app.admin.schema.user import Auth2
+from backend.app.admin.schema.user import AuthLoginParam
 from backend.app.admin.service.auth_service import auth_service
 from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
 from backend.common.security.jwt import DependsJwtAuth
@@ -19,7 +19,7 @@ async def swagger_login(form_data: OAuth2PasswordRequestForm = Depends()) -> Get
 
 
 @router.post('/login', summary='验证码登录')
-async def user_login(request: Request, obj: Auth2) -> ResponseSchemaModel[GetLoginToken]:
+async def user_login(request: Request, obj: AuthLoginParam) -> ResponseSchemaModel[GetLoginToken]:
     data = await auth_service.login(request=request, obj=obj)
     return response_base.success(data=data)
 

backend/app/admin/api/v1/user.py

@@ -4,7 +4,13 @@
 
 from fastapi import APIRouter, Query
 
-from backend.app.admin.schema.user import CreateUser, ResetPassword, UpdateUser, GetUserInfo, Avatar
+from backend.app.admin.schema.user import (
+    RegisterUserParam,
+    ResetPassword,
+    UpdateUserParam,
+    GetUserInfoDetail,
+    AvatarParam,
+)
 from backend.app.admin.service.user_service import UserService
 from backend.common.pagination import DependsPagination, paging_data, PageData
 from backend.common.response.response_schema import response_base, ResponseModel, ResponseSchemaModel
@@ -14,7 +20,7 @@
 
 
 @router.post('/register', summary='用户注册')
-async def create_user(obj: CreateUser) -> ResponseModel:
+async def create_user(obj: RegisterUserParam) -> ResponseModel:
     await UserService.register(obj=obj)
     return response_base.success()
 
@@ -28,21 +34,21 @@ async def password_reset(obj: ResetPassword) -> ResponseModel:
 
 
 @router.get('/{username}', summary='查看用户信息', dependencies=[DependsJwtAuth])
-async def get_user(username: str) -> ResponseSchemaModel[GetUserInfo]:
+async def get_user(username: str) -> ResponseSchemaModel[GetUserInfoDetail]:
     data = await UserService.get_userinfo(username=username)
     return response_base.success(data=data)
 
 
 @router.put('/{username}', summary='更新用户信息', dependencies=[DependsJwtAuth])
-async def update_userinfo(username: str, obj: UpdateUser) -> ResponseModel:
+async def update_userinfo(username: str, obj: UpdateUserParam) -> ResponseModel:
     count = await UserService.update(username=username, obj=obj)
     if count > 0:
         return response_base.success()
     return response_base.fail()
 
 
 @router.put('/{username}/avatar', summary='更新头像', dependencies=[DependsJwtAuth])
-async def update_avatar(username: str, avatar: Avatar) -> ResponseModel:
+async def update_avatar(username: str, avatar: AvatarParam) -> ResponseModel:
     count = await UserService.update_avatar(username=username, avatar=avatar)
     if count > 0:
         return response_base.success()
@@ -61,7 +67,7 @@ async def get_all_users(
     username: Annotated[str | None, Query()] = None,
     phone: Annotated[str | None, Query()] = None,
     status: Annotated[int | None, Query()] = None,
-) -> ResponseSchemaModel[PageData[GetUserInfo]]:
+) -> ResponseSchemaModel[PageData[GetUserInfoDetail]]:
     user_queryset = await UserService.get_list(username=username, phone=phone, status=status)
     page_data = await paging_data(user_queryset)
     return response_base.success(data=page_data)

backend/app/admin/crud/crud_user.py

@@ -8,7 +8,7 @@
 from tortoise.transactions import atomic
 
 from backend.app.admin.model.user import User
-from backend.app.admin.schema.user import Avatar, CreateUser, UpdateUser
+from backend.app.admin.schema.user import AvatarParam, RegisterUserParam, UpdateUserParam
 from backend.common.crud import CRUDBase
 from backend.common.security.jwt import get_hash_password
 
@@ -28,7 +28,7 @@ async def check_email(self, email: str) -> bool:
         return await self.model.filter(email=email).exists()
 
     @atomic()
-    async def register(self, obj: CreateUser) -> None:
+    async def register(self, obj: RegisterUserParam) -> None:
         salt = bcrypt.gensalt()
         obj.password = get_hash_password(obj.password, salt)
         dict_obj = obj.model_dump()
@@ -41,11 +41,11 @@ async def reset_password(self, pk: int, new_pwd: str) -> int:
         return await self.model.filter(id=pk).update(password=new_pwd)
 
     @atomic()
-    async def update_userinfo(self, input_user: int, obj_in: UpdateUser) -> int:
+    async def update_userinfo(self, input_user: int, obj_in: UpdateUserParam) -> int:
         return await self.update_model(input_user, obj_in)
 
     @atomic()
-    async def update_avatar(self, input_user: int, avatar: Avatar) -> int:
+    async def update_avatar(self, input_user: int, avatar: AvatarParam) -> int:
         return await self.update_model(input_user, {'avatar': avatar.url})
 
     async def get_list(self, username: str = None, phone: str = None, status: int = None) -> QuerySet:

backend/app/admin/schema/token.py

@@ -1,13 +1,13 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-from backend.app.admin.schema.user import GetUserInfo
+from backend.app.admin.schema.user import GetUserInfoDetail
 from backend.common.schema import SchemaBase
 
 
 class GetSwaggerToken(SchemaBase):
     access_token: str
     token_type: str = 'Bearer'
-    user: GetUserInfo
+    user: GetUserInfoDetail
 
 
 class GetLoginToken(GetSwaggerToken):

backend/app/admin/schema/user.py

@@ -1,49 +1,49 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-import datetime
+from datetime import datetime
 
-from pydantic import UUID4, ConfigDict, EmailStr, Field, HttpUrl
+from pydantic import ConfigDict, EmailStr, Field, HttpUrl
 
 from backend.common.schema import CustomPhoneNumber, SchemaBase
 
 
-class Auth(SchemaBase):
-    username: str
-    password: str
+class AuthSchemaBase(SchemaBase):
+    username: str = Field(description='用户名')
+    password: str = Field(description='密码')
 
 
-class Auth2(Auth):
-    captcha: str
+class AuthLoginParam(AuthSchemaBase):
+    captcha: str = Field(description='验证码')
 
 
-class CreateUser(Auth):
+class RegisterUserParam(AuthSchemaBase):
     email: EmailStr = Field(examples=['[email protected]'])
 
 
-class UpdateUser(SchemaBase):
-    username: str
-    email: EmailStr = Field(examples=['[email protected]'])
-    phone: CustomPhoneNumber | None = None
+class UpdateUserParam(SchemaBase):
+    username: str = Field(description='用户名')
+    email: EmailStr = Field(examples=['[email protected]'], description='邮箱')
+    phone: CustomPhoneNumber | None = Field(None, description='手机号')
 
 
-class Avatar(SchemaBase):
+class AvatarParam(SchemaBase):
     url: HttpUrl = Field(description='头像 http 地址')
 
 
-class GetUserInfo(UpdateUser):
+class GetUserInfoDetail(UpdateUserParam):
     model_config = ConfigDict(from_attributes=True)
 
-    id: int
-    uuid: UUID4
-    status: int
-    is_superuser: bool
-    avatar: str | None = None
-    join_time: datetime.datetime
-    last_login_time: datetime.datetime | None = None
+    id: int = Field(description='用户 ID')
+    uuid: str = Field(description='用户 UUID')
+    avatar: str | None = Field(None, description='头像')
+    status: int = Field(description='状态')
+    is_superuser: bool = Field(description='是否超级管理员')
+    join_time: datetime = Field(description='加入时间')
+    last_login_time: datetime | None = Field(None, description='最后登录时间')
 
 
 class ResetPassword(SchemaBase):
-    username: str
-    old_password: str
-    new_password: str
-    confirm_password: str
+    username: str = Field(description='用户名')
+    old_password: str = Field(description='旧密码')
+    new_password: str = Field(description='新密码')
+    confirm_password: str = Field(description='确认密码')

backend/app/admin/service/auth_service.py

@@ -6,7 +6,7 @@
 from backend.app.admin.crud.crud_user import user_dao
 from backend.app.admin.model.user import User
 from backend.app.admin.schema.token import GetLoginToken
-from backend.app.admin.schema.user import Auth2
+from backend.app.admin.schema.user import AuthLoginParam
 from backend.common.exception import errors
 from backend.common.response.response_code import CustomErrorCode
 from backend.common.security.jwt import create_access_token, password_verify
@@ -33,7 +33,7 @@ async def swagger_login(self, *, form_data: OAuth2PasswordRequestForm) -> tuple[
         token = create_access_token(str(user.id))
         return token, user
 
-    async def login(self, *, request: Request, obj: Auth2) -> GetLoginToken:
+    async def login(self, *, request: Request, obj: AuthLoginParam) -> GetLoginToken:
         user = await self.user_verify(obj.username, obj.password)
         try:
             captcha_uuid = request.app.state.captcha_uuid

backend/app/admin/service/user_service.py

@@ -3,14 +3,14 @@
 from tortoise.queryset import QuerySet
 from backend.app.admin.crud.crud_user import user_dao
 from backend.app.admin.model.user import User
-from backend.app.admin.schema.user import CreateUser, ResetPassword, UpdateUser, Avatar
+from backend.app.admin.schema.user import RegisterUserParam, ResetPassword, UpdateUserParam, AvatarParam
 from backend.common.exception import errors
 from backend.common.security.jwt import get_hash_password, password_verify, superuser_verify
 
 
 class UserService:
     @staticmethod
-    async def register(*, obj: CreateUser) -> None:
+    async def register(*, obj: RegisterUserParam) -> None:
         if not obj.password:
             raise errors.ForbiddenError(msg='密码为空')
         username = await user_dao.get_by_username(name=obj.username)
@@ -42,7 +42,7 @@ async def get_userinfo(*, username: str) -> User:
         return user
 
     @staticmethod
-    async def update(*, username: str, obj: UpdateUser) -> int:
+    async def update(*, username: str, obj: UpdateUserParam) -> int:
         input_user = await user_dao.get_by_username(username)
         if not input_user:
             raise errors.NotFoundError(msg='用户不存在')
@@ -59,7 +59,7 @@ async def update(*, username: str, obj: UpdateUser) -> int:
         return count
 
     @staticmethod
-    async def update_avatar(*, username: str, avatar: Avatar) -> int:
+    async def update_avatar(*, username: str, avatar: AvatarParam) -> int:
         input_user = await user_dao.get_by_username(username)
         if not input_user:
             raise errors.NotFoundError(msg='用户不存在')

backend/common/exception/errors.py

@@ -1,12 +1,5 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""
-全局业务异常类
-
-业务代码执行异常时，可以使用 raise xxxError 触发内部错误，它尽可能实现带有后台任务的异常，但它不适用于**自定义响应状态码**
-如果要求使用**自定义响应状态码**，则可以通过 return response_base.fail(res=CustomResponseCode.xxx) 直接返回
-"""  # noqa: E501
-
 from typing import Any
 
 from fastapi import HTTPException
@@ -16,6 +9,8 @@
 
 
 class BaseExceptionMixin(Exception):
+    """基础异常混入类"""
+
     code: int
 
     def __init__(self, *, msg: str = None, data: Any = None, background: BackgroundTask | None = None):
@@ -26,38 +21,50 @@ def __init__(self, *, msg: str = None, data: Any = None, background: BackgroundT
 
 
 class HTTPError(HTTPException):
+    """HTTP 异常"""
+
     def __init__(self, *, code: int, msg: Any = None, headers: dict[str, Any] | None = None):
         super().__init__(status_code=code, detail=msg, headers=headers)
 
 
 class CustomError(BaseExceptionMixin):
+    """自定义异常"""
+
     def __init__(self, *, error: CustomErrorCode, data: Any = None, background: BackgroundTask | None = None):
         self.code = error.code
         super().__init__(msg=error.msg, data=data, background=background)
 
 
 class RequestError(BaseExceptionMixin):
+    """请求异常"""
+
     code = StandardResponseCode.HTTP_400
 
     def __init__(self, *, msg: str = 'Bad Request', data: Any = None, background: BackgroundTask | None = None):
         super().__init__(msg=msg, data=data, background=background)
 
 
 class ForbiddenError(BaseExceptionMixin):
+    """禁止访问异常"""
+
     code = StandardResponseCode.HTTP_403
 
     def __init__(self, *, msg: str = 'Forbidden', data: Any = None, background: BackgroundTask | None = None):
         super().__init__(msg=msg, data=data, background=background)
 
 
 class NotFoundError(BaseExceptionMixin):
+    """资源不存在异常"""
+
     code = StandardResponseCode.HTTP_404
 
     def __init__(self, *, msg: str = 'Not Found', data: Any = None, background: BackgroundTask | None = None):
         super().__init__(msg=msg, data=data, background=background)
 
 
 class ServerError(BaseExceptionMixin):
+    """服务器异常"""
+
     code = StandardResponseCode.HTTP_500
 
     def __init__(
@@ -67,20 +74,26 @@ def __init__(
 
 
 class GatewayError(BaseExceptionMixin):
+    """网关异常"""
+
     code = StandardResponseCode.HTTP_502
 
     def __init__(self, *, msg: str = 'Bad Gateway', data: Any = None, background: BackgroundTask | None = None):
         super().__init__(msg=msg, data=data, background=background)
 
 
 class AuthorizationError(BaseExceptionMixin):
+    """授权异常"""
+
     code = StandardResponseCode.HTTP_401
 
     def __init__(self, *, msg: str = 'Permission Denied', data: Any = None, background: BackgroundTask | None = None):
         super().__init__(msg=msg, data=data, background=background)
 
 
 class TokenError(HTTPError):
+    """Token 异常"""
+
     code = StandardResponseCode.HTTP_401
 
     def __init__(self, *, msg: str = 'Not Authenticated', headers: dict[str, Any] | None = None):