feat(dto): update dto

Author: wangxin688

src/_types.py

@@ -4,7 +4,7 @@
 
 import pydantic
 from fastapi import Query
-from pydantic import ConfigDict, Field, constr
+from pydantic import ConfigDict, Field, StringConstraints
 from pydantic.functional_validators import BeforeValidator
 
 from src.validators import items_to_list, mac_address_validator
@@ -18,8 +18,8 @@
 StrList = Annotated[str | list[str], BeforeValidator(items_to_list)]
 IntList = Annotated[int | list[int], BeforeValidator(items_to_list)]
 MacAddress = Annotated[str, BeforeValidator(mac_address_validator)]
-NameStr = constr(pattern="^[a-zA-Z0-9_-].$", max_length=50)
-NameChineseStr = constr(pattern="^[\u4e00-\u9fa5a-zA-Z0-9_-].$", max_length=50)
+NameStr = Annotated[str, StringConstraints(pattern="^[a-zA-Z0-9_-].$", max_length=50)]
+NameChineseStr = Annotated[str, StringConstraints(pattern="^[\u4e00-\u9fa5a-zA-Z0-9_-].$", max_length=50)]
 
 
 class BaseModel(pydantic.BaseModel):
@@ -44,9 +44,18 @@ class AuditUser(BaseModel):
     updated_by: AuditUserBase | None = None
 
 
+class AuditLog(BaseModel):
+    id: int
+    created_at: datetime
+    request_id: str
+    action: str
+    diff: dict | None = None
+    user: AuditUserBase | None = None
+
+
 class ListT(BaseModel, Generic[T]):
     count: int
-    results: T | None = None
+    results: list[T] | None = None
 
 
 class AppStrEnum(str, Enum):
@@ -70,15 +79,6 @@ class AuidtUserQuery(BaseModel):
     updated_by_fk: list[int] = Field(Query(default=[]))
 
 
-class AuditLog(BaseModel):
-    id: int
-    created_at: datetime
-    request_id: str
-    action: str
-    diff: dict | None = None
-    user: AuditUserBase | None = None
-
-
 class QueryParams(BaseModel):
     limit: int | None = Query(default=20, ge=0, le=1000, description="Number of results to return per request.")
     offset: int | None = Query(default=0, ge=0, description="The initial index from which return the results.")
@@ -108,3 +108,7 @@ class VisibleName(TypedDict, total=True):
 
 class IdResponse(BaseModel):
     id: int
+
+
+class IdCreate(IdResponse):
+    ...

src/auth/api.py

@@ -1,15 +1,15 @@
 from fastapi import APIRouter, Depends, status
 from fastapi.security import OAuth2PasswordRequestForm
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
 from src import errors
 from src._types import IdResponse, ListT
 from src.auth import schemas
-from src.auth.models import Group, Menu, Permission, Role, User
-from src.auth.services import GroupDto, MenuDto, RoleDto, UserDto
+from src.auth.models import Group, Menu, Role, User
+from src.auth.services import MenuDto, UserDto
 from src.cbv import cbv
+from src.db.dtobase import DtoBase
 from src.deps import auth, get_session
 from src.exceptions import GenerError
 from src.security import generate_access_token_response
@@ -23,21 +23,21 @@ async def login_pwd(
     user: OAuth2PasswordRequestForm = Depends(),
     session: AsyncSession = Depends(get_session),
 ) -> schemas.AccessToken:
-    user_dto = UserDto(User)
-    result = await user_dto.verify_user(session, user)
+    dto = UserDto(User)
+    result = await dto.verify_user(session, user)
     return generate_access_token_response(result.id)
 
 
 @cbv(router)
 class UserAPI:
     user: User = Depends(auth)
     session: AsyncSession = Depends(get_session)
-    user_dto = UserDto(User)
+    dto = UserDto(User)
 
     @router.post("/users", operation_id="5091fff6-1adc-4a22-8a8c-ef0107122df7", summary="创建新用户/Create new user")
     async def create_user(self, user: schemas.UserCreate) -> IdResponse:
-        new_user = await self.user_dto.create(self.session, user)
-        result = await self.user_dto.commit(self.session, new_user)
+        new_user = await self.dto.create(self.session, user)
+        result = await self.dto.commit(self.session, new_user)
         return IdResponse(id=result.id)
 
     @router.get(
@@ -46,7 +46,7 @@ async def create_user(self, user: schemas.UserCreate) -> IdResponse:
         summary="获取单个用户/Get user information by ID",
     )
     async def get_user(self, id: int) -> schemas.UserDetail:
-        db_user = await self.user_dto.get_one_or_404(
+        db_user = await self.dto.get_one_or_404(
             self.session,
             id,
             selectinload(User.role).load_only(Role.id, Role.name),
@@ -55,8 +55,8 @@ async def get_user(self, id: int) -> schemas.UserDetail:
         return schemas.UserDetail.model_validate(db_user)
 
     @router.get("/users", operation_id="2485e2a2-4d81-4601-a6fd-c633b23ce5fc")
-    async def get_users(self, query: schemas.UserQuery = Depends()) -> ListT[list[schemas.UserDetail]]:
-        count, results = await self.user_dto.list_and_count(
+    async def get_users(self, query: schemas.UserQuery = Depends()) -> ListT[schemas.UserDetail]:
+        count, results = await self.dto.list_and_count(
             self.session,
             query,
             selectinload(User.role).load_only(Role.id, Role.name),
@@ -68,129 +68,110 @@ async def get_users(self, query: schemas.UserQuery = Depends()) -> ListT[list[sc
     async def update_user(self, id: int, user: schemas.UserUpdate) -> IdResponse:
         update_user = user.model_dump(exclude_unset=True)
         if "password" in update_user and update_user["password"] is None:
-            raise GenerError(errors.ERR_10006, status_code=status.HTTP_406_NOT_ACCEPTABLE)
-        db_user = await self.user_dto.get_one_or_404(self.session, id)
-        await self.user_dto.update(self.session, db_user, user)
+            raise GenerError(errors.ERR_10005, status_code=status.HTTP_406_NOT_ACCEPTABLE)
+        db_user = await self.dto.get_one_or_404(self.session, id)
+        await self.dto.update(self.session, db_user, user)
         return IdResponse(id=id)
 
     @router.delete("/users/{id}", operation_id="78e48ceb-d7cf-46fe-bf9e-d04958aade7d")
     async def delete_user(self, id: int) -> IdResponse:
-        db_user = await self.user_dto.get_one_or_404(self.session, id)
-        await self.user_dto.delete(self.session, db_user)
+        db_user = await self.dto.get_one_or_404(self.session, id)
+        await self.dto.delete(self.session, db_user)
         return IdResponse(id=id)
 
 
 @cbv(router)
 class GroupAPI:
     user: User = Depends(auth)
     session: AsyncSession = Depends(get_session)
-    group_dto = GroupDto(Group)
+    dto = DtoBase(Group)
 
     @router.post("/groups", operation_id="9e3e639d-c694-467d-9209-717b038cf267")
     async def create_group(self, group: schemas.GroupCreate) -> IdResponse:
-        if not group.user_ids:
-            new_group = await self.group_dto.create(self.session, group)
-        else:
-            users = (await self.session.scalars(select(User).where(User.id.in_(group.user_ids)))).all()
-            new_group = await self.group_dto.create_with_users(self.session, group, users)
+        new_group = await self.dto.create(self.session, group)
         return IdResponse(id=new_group.id)
 
     @router.get("/groups/{id}", operation_id="00327087-9443-4d24-8d04-e396e3244744")
     async def get_group(self, id: int) -> schemas.GroupDetail:
-        db_group = await self.group_dto.get_one_or_404(self.session, id, undefer_load=True)
+        db_group = await self.dto.get_one_or_404(self.session, id, undefer_load=True)
         return schemas.GroupDetail.model_validate(db_group)
 
     @router.get("/groups", operation_id="a1d1f8f1-4d4d-4fab-868b-3f977df26e05")
-    async def get_groups(self, query: schemas.GroupQuery = Depends()) -> ListT[list[schemas.GroupDetail]]:
-        count, results = await self.group_dto.list_and_count(self.session, query)
+    async def get_groups(self, query: schemas.GroupQuery = Depends()) -> ListT[schemas.GroupDetail]:
+        count, results = await self.dto.list_and_count(self.session, query)
         return ListT(count=count, results=[schemas.GroupDetail.model_validate(r) for r in results])
 
     @router.put("/groups/{id}", operation_id="3d5badd1-665c-49f8-85c4-6f6d7f3a1b2a")
     async def update_group(self, id: int, group: schemas.GroupUpdate) -> IdResponse:
-        db_group = await self.group_dto.get_one_or_404(self.session, id, selectinload(Group.user))
-        update_group = group.model_dump(exclude_unset=True)
-        if "user_ids" in update_group:
-            db_group = await self.group_dto.update_relationship_field(
-                self.session, db_group, User, "user", group.user_ids
-            )
-        await self.group_dto.update(self.session, db_group, group, excludes={"user_ids"})
+        db_group = await self.dto.get_one_or_404(self.session, id, selectinload(Group.user))
+        await self.dto.update(self.session, db_group, group)
         return IdResponse(id=id)
 
     @router.delete("/groups/{id}", operation_id="e16830da-2973-4369-8e75-da9b4174ab72")
     async def delete_group(self, id: int) -> IdResponse:
-        db_group = await self.group_dto.get_one_or_404(self.session, id)
-        await self.group_dto.delete(self.session, db_group)
+        db_group = await self.dto.get_one_or_404(self.session, id)
+        await self.dto.delete(self.session, db_group)
         return IdResponse(id=id)
 
 
 @cbv(router)
 class RoleAPI:
     user: User = Depends(auth)
     session: AsyncSession = Depends(get_session)
-    role_dto = RoleDto(Role)
+    dto = DtoBase(Role)
 
     @router.post("/roles", operation_id="a18a152b-e9e9-4128-b8be-8a8e9c842abb")
     async def create_role(self, role: schemas.RoleCreate) -> IdResponse:
-        if not role.permission_ids:
-            new_role = await self.role_dto.create(self.session, role)
-        else:
-            permissions = (
-                await self.session.scalars(select(Permission).where(Permission.id.in_(role.permission_ids)))
-            ).all()
-            new_role = await self.role_dto.create_with_permissions(self.session, role, permissions)
+        new_role = await self.dto.create(self.session, role)
         return IdResponse(id=new_role.id)
 
     @router.get("/roles/{id}", operation_id="2b45f59a-77a1-45d4-bf43-94373da517e3")
     async def get_role(self, id: int) -> schemas.RoleDetail:
-        db_role = await self.role_dto.get_one_or_404(self.session, id, selectinload(Role.permission), undefer_load=True)
+        db_role = await self.dto.get_one_or_404(self.session, id, selectinload(Role.permission), undefer_load=True)
         return schemas.RoleDetail.model_validate(db_role)
 
     @router.get("/roles", operation_id="c5f793b1-7adf-4b4e-a498-732b0fa7d758")
-    async def get_roles(self, query: schemas.RoleQuery = Depends()) -> ListT[list[schemas.RoleList]]:
-        count, results = await self.role_dto.list_and_count(self.session, query)
+    async def get_roles(self, query: schemas.RoleQuery = Depends()) -> ListT[schemas.RoleList]:
+        count, results = await self.dto.list_and_count(self.session, query)
         return ListT(count=count, results=[schemas.RoleList.model_validate(r) for r in results])
 
     @router.put("/roles/{id}", operation_id="2fda2e00-ad86-4296-a1d4-c7f02366b52e")
     async def update_role(self, id: int, role: schemas.RoleUpdate) -> IdResponse:
-        db_role = await self.role_dto.get_one_or_404(self.session, id, selectinload(Role.permission))
-        if "permission_ids" in role.model_dump(exclude_unset=True):
-            db_role = await self.role_dto.update_relationship_field(
-                self.session, db_role, Permission, "permission", role.permission_ids
-            )
-        await self.role_dto.update(self.session, db_role, role, excludes={"permission_ids"})
+        db_role = await self.dto.get_one_or_404(self.session, id, selectinload(Role.permission))
+        await self.dto.update(self.session, db_role, role)
         return IdResponse(id=id)
 
     @router.delete("/roles/{id}", operation_id="c4e9e0e8-6b0c-4f6f-9e6c-8d9f9f9f9f9f")
     async def delete_role(self, id: int) -> IdResponse:
-        db_role = await self.role_dto.get_one_or_404(self.session, id)
-        await self.role_dto.delete(self.session, db_role)
+        db_role = await self.dto.get_one_or_404(self.session, id)
+        await self.dto.delete(self.session, db_role)
         return IdResponse(id=id)
 
 
 @cbv(router)
 class MenuAPI:
     user: User = Depends(auth)
     session: AsyncSession = Depends(get_session)
-    menu_dto = MenuDto(Menu)
+    dto = MenuDto(Menu)
 
     @router.post("/menus", operation_id="008bf4d4-cc01-48b0-82b8-1a67c0348b31")
     async def create_menu(self, meun: schemas.MenuCreate) -> IdResponse:
-        new_menu = await self.menu_dto.create(self.session, meun)
+        new_menu = await self.dto.create(self.session, meun)
         return IdResponse(id=new_menu.id)
 
     @router.get("/menus", operation_id="cb7f25ab-798b-4668-a838-6339425e2889")
     async def get_menus(self) -> schemas.MenuTree:
-        results = await self.menu_dto.get_all(self.session)
+        results = await self.dto.get_all(self.session)
         data = list_to_tree([r.dict() for r in results])
         return schemas.MenuTree.model_validate(data)
 
     @router.put("menus/{id}", operation_id="b4d7ac97-a182-4bd1-a75c-6ae44b5fcf0a")
     async def update_menu(self, id: int, meun: schemas.MenuUpdate) -> IdResponse:
-        db_menu = await self.menu_dto.get_one_or_404(self.session, id)
-        await self.menu_dto.update(self.session, db_menu, meun)
+        db_menu = await self.dto.get_one_or_404(self.session, id)
+        await self.dto.update(self.session, db_menu, meun)
         return IdResponse(id=id)
 
     async def delete_menu(self, id: int) -> IdResponse:
-        db_menu = await self.menu_dto.get_one_or_404(self.session, id)
-        await self.menu_dto.delete(self.session, db_menu)
+        db_menu = await self.dto.get_one_or_404(self.session, id)
+        await self.dto.delete(self.session, db_menu)
         return IdResponse(id=id)

src/auth/schemas.py

@@ -1,9 +1,8 @@
 from datetime import datetime
-from uuid import UUID
 
 from pydantic_extra_types.phone_numbers import PhoneNumber
 
-from src._types import AuditTime, BaseModel, QueryParams
+from src._types import AuditTime, BaseModel, IdCreate, QueryParams
 
 
 class AccessToken(BaseModel):
@@ -43,7 +42,7 @@ class UserBase(BaseModel):
     avatar: str | None = None
 
 
-class UserBrief(UserBase, AuditTime):
+class UserBrief(UserBase):
     id: int
 
 
@@ -130,11 +129,11 @@ class UserCreate(UserBase):
 class GroupCreate(GroupBase):
     password: str
     role_id: int
-    user_ids: list[int] | None = None
+    user: list[IdCreate]
 
 
 class RoleCreate(RoleBase):
-    permission_ids: list[UUID] | None = None
+    permission: list[IdCreate]
 
 
 class UserUpdate(UserCreate):

src/auth/services.py

@@ -5,7 +5,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.auth import schemas
-from src.auth.models import Group, Menu, Permission, Role, User
+from src.auth.models import Menu, Permission, User
 from src.auth.schemas import PermissionCreate, PermissionUpdate
 from src.context import locale_ctx
 from src.db.dtobase import DtoBase
@@ -24,48 +24,6 @@ async def verify_user(self, session: AsyncSession, user: OAuth2PasswordRequestFo
         return db_user
 
 
-class GroupDto(DtoBase[Group, schemas.GroupCreate, schemas.GroupUpdate, schemas.GroupQuery]):
-    async def create_with_users(
-        self, session: AsyncSession, group: schemas.GroupCreate, users: Sequence[User]
-    ) -> Group:
-        """
-        Create a new group with the specified users.
-
-        Parameters:
-            session (AsyncSession): The database session.
-            group (schemas.GroupCreate): The group data to create.
-            users (Sequence[User]): The list of users to associate with the group.
-
-        Returns:
-            Group: The newly created group.
-
-        """
-        new_group = await self.create(session, group, excludes={"user_ids"}, commit=False)
-        new_group.user.extend(users)
-        return await self.commit(session, new_group)
-
-
-class RoleDto(DtoBase[Role, schemas.RoleCreate, schemas.RoleUpdate, schemas.RoleQuery]):
-    async def create_with_permissions(
-        self, session: AsyncSession, role: schemas.RoleCreate, permissions: Sequence[Permission]
-    ) -> Role:
-        """
-        Create a new role with the specified permissions.
-
-        Parameters:
-            session (AsyncSession): The database session.
-            role (schemas.RoleCreate): The role data to create.
-            permissions (Sequence[Permission]): The list of permissions to associate with the role.
-
-        Returns:
-            Role: The newly created role.
-
-        """
-        new_role = await self.create(session, role, excludes={"permission_ids"}, commit=False)
-        new_role.permission.extend(permissions)
-        return await self.commit(session, new_role)
-
-
 class PermissionDto(DtoBase[Permission, schemas.PermissionCreate, schemas.PermissionUpdate, schemas.PermissionQuery]):
     async def create(
         self,