Fix pydantic v2 warnings related to UserGroup

api/main.py

@@ -33,7 +33,7 @@
 from pymongo.errors import DuplicateKeyError
 from fastapi_users import FastAPIUsers
 from beanie import PydanticObjectId
 from kernelci.api.models import (
     Node,
     Hierarchy,
     PublishEvent,
@@ -52,7 +52,9 @@
     User,
     UserRead,
     UserCreate,
+    UserCreateRequest,
     UserUpdate,
+    UserUpdateRequest,
     UserGroup,
 )
 from .metrics import Metrics
@@ -157,7 +159,7 @@
 
 @app.post("/user/register", response_model=UserRead, tags=["user"],
           response_model_by_alias=False)
-async def register(request: Request, user: UserCreate,
+async def register(request: Request, user: UserCreateRequest,
                    current_user: User = Depends(get_current_superuser)):
     """User registration route
 
@@ -183,9 +185,11 @@
                     detail=f"User group does not exist with name: \
     {group_name}")
             groups.append(group)
-    user.groups = groups
+    user_create = UserCreate(**(user.model_dump(
+         exclude={'groups'}, exclude_none=True)))
+    user_create.groups = groups
     created_user = await register_router.routes[0].endpoint(
-        request, user, user_manager)
+        request, user_create, user_manager)
     # Update user to be an admin user explicitly if requested as
     # `fastapi-users` register route does not allow it
     if user.is_superuser:
@@ -233,7 +237,7 @@
 
 @app.patch("/user/me", response_model=UserRead, tags=["user"],
            response_model_by_alias=False)
-async def update_me(request: Request, user: UserUpdate,
+async def update_me(request: Request, user: UserUpdateRequest,
                     current_user: User = Depends(get_current_user)):
     """User update route
 
@@ -258,15 +262,17 @@
                     detail=f"User group does not exist with name: \
     {group_name}")
             groups.append(group)
+    user_update = UserUpdate(**(user.model_dump(
+         exclude={'groups'}, exclude_none=True)))
     if groups:
-        user.groups = groups
+        user_update.groups = groups
     return await users_router.routes[1].endpoint(
-        request, user, current_user, user_manager)
+        request, user_update, current_user, user_manager)
 
 
 @app.patch("/user/{user_id}", response_model=UserRead, tags=["user"],
            response_model_by_alias=False)
-async def update_user(user_id: str, request: Request, user: UserUpdate,
+async def update_user(user_id: str, request: Request, user: UserUpdateRequest,
                       current_user: User = Depends(get_current_superuser)):
     """Router to allow admin users to update other user account"""
     metrics.add('http_requests_total', 1)
@@ -295,11 +301,14 @@
                     detail=f"User group does not exist with name: \
     {group_name}")
             groups.append(group)
+    user_update = UserUpdate(**(user.model_dump(
+         exclude={'groups'}, exclude_none=True)))
+
     if groups:
-        user.groups = groups
+        user_update.groups = groups
 
     updated_user = await users_router.routes[3].endpoint(
-        user, request, user_from_id, user_manager
+        user_update, request, user_from_id, user_manager
     )
     # Update user to be an admin user explicitly if requested as
     # `fastapi-users` user update route does not allow it

api/models.py

@@ -28,7 +28,7 @@
     Document,
     PydanticObjectId,
 )
 from kernelci.api.models_base import DatabaseModel, ModelId
 
 
 # PubSub model definitions
@@ -87,7 +87,7 @@
     )
 
     @field_validator('groups')
     def validate_groups(cls, groups):   # pylint: disable=no-self-argument
         """Unique group constraint"""
         unique_names = {group.name for group in groups}
         if len(unique_names) != len(groups):
@@ -107,13 +107,13 @@
         ]
 
 
 class UserRead(schemas.BaseUser[PydanticObjectId], ModelId):
     """Schema for reading a user"""
     username: Annotated[str, Indexed(unique=True)]
     groups: List[UserGroup] = Field(default=[])
 
     @field_validator('groups')
     def validate_groups(cls, groups):   # pylint: disable=no-self-argument
         """Unique group constraint"""
         unique_names = {group.name for group in groups}
         if len(unique_names) != len(groups):
@@ -121,13 +121,13 @@
         return groups
 
 
-class UserCreate(schemas.BaseUserCreate):
-    """Schema for creating a user"""
+class UserCreateRequest(schemas.BaseUserCreate):
+    """Create user request schema for API router"""
     username: Annotated[str, Indexed(unique=True)]
     groups: List[str] = Field(default=[])
 
     @field_validator('groups')
     def validate_groups(cls, groups):   # pylint: disable=no-self-argument
         """Unique group constraint"""
         unique_names = set(groups)
         if len(unique_names) != len(groups):
@@ -135,14 +135,28 @@
         return groups
 
 
-class UserUpdate(schemas.BaseUserUpdate):
-    """Schema for updating a user"""
+class UserCreate(schemas.BaseUserCreate):
+    """Schema used for sending create user request to 'fastapi-users' router"""
+    username: Annotated[str, Indexed(unique=True)]
+    groups: List[UserGroup] = Field(default=[])
+
+    @field_validator('groups')
+    def validate_groups(cls, groups):   # pylint: disable=no-self-argument
+        """Unique group constraint"""
+        unique_names = {group.name for group in groups}
+        if len(unique_names) != len(groups):
+            raise ValueError("Groups must have unique names.")
+        return groups
+
+
+class UserUpdateRequest(schemas.BaseUserUpdate):
+    """Update user request schema for API router"""
     username: Annotated[Optional[str], Indexed(unique=True),
                         Field(default=None)]
     groups: List[str] = Field(default=[])
 
     @field_validator('groups')
     def validate_groups(cls, groups):   # pylint: disable=no-self-argument
         """Unique group constraint"""
         unique_names = set(groups)
         if len(unique_names) != len(groups):
@@ -150,6 +164,21 @@
         return groups
 
 
+class UserUpdate(schemas.BaseUserUpdate):
+    """Schema used for sending update user request to 'fastapi-users' router"""
+    username: Annotated[Optional[str], Indexed(unique=True),
+                        Field(default=None)]
+    groups: List[UserGroup] = Field(default=[])
+
+    @field_validator('groups')
+    def validate_groups(cls, groups):   # pylint: disable=no-self-argument
+        """Unique group constraint"""
+        unique_names = {group.name for group in groups}
+        if len(unique_names) != len(groups):
+            raise ValueError("Groups must have unique names.")
+        return groups
+
+
 # Pagination models
 
 class CustomLimitOffsetParams(LimitOffsetParams):