Improved swagger doc

Author: Yanni8

src/foxops/database/repositories/group/errors.py

@@ -1,6 +1,23 @@
-class GroupNotFoundError(Exception):
-    id: int
+from typing import Optional
 
+from foxops.errors import FoxopsError
 
-class GroupAlreadyExistsError(Exception):
-    system_name: str
+
+class GroupNotFoundError(FoxopsError):
+    def __init__(self, system_name: Optional[str] = None, id: Optional[int] = None):
+        if system_name is None and id is None:
+            raise ValueError("system_name or id must be provided")
+        if system_name is not None and id is not None:
+            raise ValueError("system_name and id cannot both be provided")
+        if system_name is not None:
+            self.system_name = system_name
+            super().__init__(f"Group with system name '{system_name}' not found.")
+        else:
+            self.id = id
+            super().__init__(f"Group with id '{id}' not found.")
+
+
+class GroupAlreadyExistsError(FoxopsError):
+    def __init__(self, system_name: str):
+        self.system_name = system_name
+        super().__init__(f"Group with System name ${system_name} already exists")

src/foxops/database/repositories/group/repository.py

@@ -21,7 +21,7 @@ async def create(self, system_name: str, display_name: str) -> GroupInDB:
             try:
                 result = await conn.execute(query)
             except IntegrityError as e:
-                raise GroupAlreadyExistsError(f"Group with System name ${system_name} already exists") from e
+                raise GroupAlreadyExistsError(system_name) from e
 
             row = result.one()
             return GroupInDB.model_validate(row)
@@ -35,7 +35,7 @@ async def get_by_system_name(self, system_name: str) -> GroupInDB:
             try:
                 row = result.one()
             except NoResultFound as e:
-                raise GroupNotFoundError(f"Group with System name ${system_name} not found") from e
+                raise GroupNotFoundError(system_name=system_name) from e
 
             return GroupInDB.model_validate(row)
 
@@ -59,5 +59,5 @@ async def get_by_id(self, group_id: int) -> GroupInDB:
             try:
                 row = result.one()
             except NoResultFound as e:
-                raise GroupNotFoundError(f"Group with ID ${group_id} not found") from e
+                raise GroupNotFoundError(id=group_id) from e
             return GroupInDB.model_validate(row)

src/foxops/database/repositories/incarnation/errors.py

@@ -1,5 +1,7 @@
 class IncarnationNotFoundError(Exception):
-    pass
+    def __init__(self, id: int):
+        self.id = id
+        super().__init__(f"Incarnation with id '{id}' not found.")
 
 
 class IncarnationAlreadyExistsError(Exception):

src/foxops/database/repositories/incarnation/repository.py

@@ -73,7 +73,7 @@ async def get_by_id(self, id_: int) -> IncarnationInDB:
             try:
                 row = result.one()
             except NoResultFound:
-                raise IncarnationNotFoundError(f"could not find incarnation in DB with id: {id_}")
+                raise IncarnationNotFoundError(id_)
             else:
                 return IncarnationInDB.model_validate(row)
 
@@ -84,7 +84,7 @@ async def delete_by_id(self, id_: int) -> None:
             result = await conn.execute(query)
 
             if result.rowcount == 0:
-                raise IncarnationNotFoundError(f"could not find incarnation in DB with id: {id_}")
+                raise IncarnationNotFoundError(id_)
 
     async def get_group_permissions(self, incarnation_id: int) -> List[GroupPermissionInDB]:
         query = (
@@ -162,10 +162,10 @@ async def set_owner(self, incarnation_id: int, user_id: int):
             try:
                 result = await conn.execute(query)
             except IntegrityError as e:
-                raise UserNotFoundError(f"User with ID {user_id} not found") from e
+                raise UserNotFoundError(id=user_id) from e
             try:
                 row = result.one()
             except NoResultFound as e:
-                raise IncarnationNotFoundError(f"Incarnation with ID {incarnation_id} not found") from e
+                raise IncarnationNotFoundError(incarnation_id) from e
 
             return IncarnationInDB.model_validate(row)

src/foxops/database/repositories/user/errors.py

@@ -1,6 +1,20 @@
-class UserNotFoundError(Exception):
-    pass
+from typing import Optional
+
+from foxops.errors import FoxopsError
+
+
+class UserNotFoundError(FoxopsError):
+    def __init__(self, id: Optional[int] = None, username: Optional[str] = None):
+        if id is None and username is None:
+            raise ValueError("id or username must be provided")
+        if id is not None and username is not None:
+            raise ValueError("id and username cannot both be provided")
+        if id is not None:
+            super().__init__(f"User with id '{id}' not found.")
+        else:
+            super().__init__(f"User with username '{username}' not found.")
 
 
 class UserAlreadyExistsError(Exception):
-    pass
+    def __init__(self, username: str):
+        super().__init__(f"User with username '{username}' already exists.")

src/foxops/database/repositories/user/repository.py

@@ -21,7 +21,7 @@ async def create(self, username: str, is_admin: bool) -> UserInDB:
             try:
                 result = await conn.execute(query)
             except IntegrityError as e:
-                raise UserAlreadyExistsError(f"username={username}") from e
+                raise UserAlreadyExistsError(username=username) from e
 
             row = result.one()
             return UserInDB.model_validate(row)
@@ -35,7 +35,7 @@ async def get_by_username(self, username: str) -> UserInDB:
             try:
                 row = result.one()
             except NoResultFound as e:
-                raise UserNotFoundError(f"username={username}") from e
+                raise UserNotFoundError(username=username) from e
 
             return UserInDB.model_validate(row)
 
@@ -67,6 +67,6 @@ async def get_by_id(self, user_id: int) -> UserInDB:
             try:
                 row = result.one()
             except NoResultFound as e:
-                raise UserNotFoundError(f"User with ID ${user_id} not found") from e
+                raise UserNotFoundError(id=user_id) from e
 
             return UserInDB.model_validate(row)

src/foxops/routers/incarnations.py

@@ -3,7 +3,9 @@
 from fastapi import APIRouter, Depends, Response, status
 from pydantic import BaseModel, model_validator
 
+from foxops.database.repositories.group.errors import GroupNotFoundError
 from foxops.database.repositories.incarnation.errors import IncarnationNotFoundError
+from foxops.database.repositories.user.errors import UserNotFoundError
 from foxops.dependencies import (
     authorization,
     get_change_service,
@@ -230,9 +232,9 @@ async def reset_incarnation(
             message=f"could not initialize the incarnation as the provided template data "
             f"is invalid: {'; '.join(error_messages)}"
         )
-    except IncarnationNotFoundError:
+    except IncarnationNotFoundError as exc:
         response.status_code = status.HTTP_404_NOT_FOUND
-        return ApiError(message="The incarnation was not found in the inventory")
+        return ApiError(message=str(exc))
     except ChangeRejectedDueToNoChanges:
         response.status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
         return ApiError(message="The incarnation does not have any customizations. Nothing to reset.")
@@ -295,8 +297,8 @@ class UpdateIncarnationRequest(BaseModel):
     template_data: TemplateData
     owner_id: int
 
-    user_permission: list[UnresolvedUserPermissions]
-    group_permission: list[UnresolvedGroupPermissions]
+    user_permissions: list[UnresolvedUserPermissions]
+    group_permissions: list[UnresolvedGroupPermissions]
 
     automerge: bool
 
@@ -341,10 +343,32 @@ async def update_incarnation(
     """
 
     await incarnation_service.remove_all_permissions(incarnation_id)
-    await incarnation_service.set_user_permissions(incarnation_id, request.user_permission)
-    await incarnation_service.set_group_permissions(incarnation_id, request.group_permission)
+    try:
+        await incarnation_service.set_user_permissions(incarnation_id, request.user_permissions)
+    except UserNotFoundError as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ApiError(message=str(exc))
+    except IncarnationNotFoundError as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ApiError(message=str(exc))
 
-    await incarnation_service.set_owner(incarnation_id, request.owner_id)
+    try:
+        await incarnation_service.set_group_permissions(incarnation_id, request.group_permissions)
+    except GroupNotFoundError as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ApiError(message=str(exc))
+    except IncarnationNotFoundError as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ApiError(message=str(exc))
+
+    try:
+        await incarnation_service.set_owner(incarnation_id, request.owner_id)
+    except UserNotFoundError as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ApiError(message=str(exc))
+    except IncarnationNotFoundError as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ApiError(message=str(exc))
 
     return await _create_change(
         incarnation_id=incarnation_id,
@@ -362,8 +386,8 @@ class PatchIncarnationRequest(BaseModel):
 
     requested_version: str | None = None
     requested_data: TemplateData | None = None
-    user_permission: list[UnresolvedUserPermissions] | None = None
-    group_permission: list[UnresolvedGroupPermissions] | None = None
+    user_permissions: list[UnresolvedUserPermissions] | None = None
+    group_permissions: list[UnresolvedGroupPermissions] | None = None
     owner_id: int | None = None
 
     automerge: bool | None = None
@@ -373,8 +397,8 @@ def check_either_version_or_data_change_requested(self) -> Self:
         if (
             self.requested_version is None
             and self.requested_data is None
-            and self.user_permission is None
-            and self.group_permission is None
+            and self.user_permissions is None
+            and self.group_permissions is None
             and self.owner_id is None
         ):
             raise ValueError(
@@ -427,16 +451,37 @@ async def patch_incarnation(
 
     requested_data = request.requested_data or {}
 
-    if request.group_permission is not None:
-        await incarnation_service.remove_all_group_permissions(incarnation_id)
-        await incarnation_service.set_group_permissions(incarnation_id, request.group_permission)
-
-    if request.user_permission is not None:
-        await incarnation_service.remove_all_user_permissions(incarnation_id)
-        await incarnation_service.set_user_permissions(incarnation_id, request.user_permission)
+    if request.group_permissions is not None:
+        try:
+            await incarnation_service.remove_all_group_permissions(incarnation_id)
+            await incarnation_service.set_group_permissions(incarnation_id, request.group_permissions)
+        except GroupNotFoundError as exc:
+            response.status_code = status.HTTP_404_NOT_FOUND
+            return ApiError(message=str(exc))
+        except IncarnationNotFoundError as exc:
+            response.status_code = status.HTTP_404_NOT_FOUND
+            return ApiError(message=str(exc))
+
+    if request.user_permissions is not None:
+        try:
+            await incarnation_service.remove_all_user_permissions(incarnation_id)
+            await incarnation_service.set_user_permissions(incarnation_id, request.user_permissions)
+        except UserNotFoundError as exc:
+            response.status_code = status.HTTP_404_NOT_FOUND
+            return ApiError(message=str(exc))
+        except IncarnationNotFoundError as exc:
+            response.status_code = status.HTTP_404_NOT_FOUND
+            return ApiError(message=str(exc))
 
     if request.owner_id is not None:
-        await incarnation_service.set_owner(incarnation_id, request.owner_id)
+        try:
+            await incarnation_service.set_owner(incarnation_id, request.owner_id)
+        except UserNotFoundError as exc:
+            response.status_code = status.HTTP_404_NOT_FOUND
+            return ApiError(message=str(exc))
+        except IncarnationNotFoundError as exc:
+            response.status_code = status.HTTP_404_NOT_FOUND
+            return ApiError(message=str(exc))
 
     if request.requested_version is not None or request.requested_data is not None:
         return await _create_change(
@@ -449,7 +494,11 @@ async def patch_incarnation(
             change_service=change_service,
         )
     else:
-        return await change_service.get_incarnation_with_details(incarnation_id)
+        try:
+            return await change_service.get_incarnation_with_details(incarnation_id)
+        except IncarnationNotFoundError as exc:
+            response.status_code = status.HTTP_404_NOT_FOUND
+            return ApiError(message=str(exc))
 
 
 @router.delete(

src/foxops/services/incarnation.py

@@ -84,6 +84,8 @@ async def set_user_permissions(
             for user_permission in user_permissions
         ]
 
+        await self.incarnation_repository.get_by_id(incarnation_id)  # Validate if the incarnation exists
+
         await self.incarnation_repository.set_user_permissions(incarnation_id, resolved_user_permissions)
 
     async def set_group_permissions(
@@ -99,6 +101,8 @@ async def set_group_permissions(
             for group_permission in group_permissions
         ]
 
+        await self.incarnation_repository.get_by_id(incarnation_id)  # Validate if the incarnation exists
+
         await self.incarnation_repository.set_group_permissions(incarnation_id, resolved_group_permissions)
 
     async def remove_all_user_permissions(self, incarnation_id: int) -> None: