♻️ Enhanced groups/organizations web-api specs and validation 🚨

api/specs/web-server/_groups.py

@@ -8,14 +8,21 @@
 
 from fastapi import APIRouter, Depends, status
 from models_library.api_schemas_webserver.groups import (
-    AllUsersGroups,
+    GroupCreate,
+    GroupGet,
+    GroupUpdate,
     GroupUserGet,
-    UsersGroup,
+    MyGroupsGet,
 )
 from models_library.generics import Envelope
-from models_library.users import GroupID, UserID
 from simcore_service_webserver._meta import API_VTAG
-from simcore_service_webserver.groups._handlers import _ClassifiersQuery
+from simcore_service_webserver.groups._handlers import (
+    GroupUserAdd,
+    GroupUserUpdate,
+    _ClassifiersQuery,
+    _GroupPathParams,
+    _GroupUserPathParams,
+)
 from simcore_service_webserver.scicrunch.models import ResearchResource, ResourceHit
 
 router = APIRouter(
@@ -28,106 +35,130 @@
 
 @router.get(
     "/groups",
-    response_model=Envelope[AllUsersGroups],
+    response_model=Envelope[MyGroupsGet],
 )
 async def list_groups():
-    ...
+    """
+    List all groups (organizations, primary, everyone and products) I belong to
+    """
 
 
 @router.post(
     "/groups",
-    response_model=Envelope[UsersGroup],
+    response_model=Envelope[GroupGet],
     status_code=status.HTTP_201_CREATED,
 )
-async def create_group():
-    ...
+async def create_group(_b: GroupCreate):
+    """
+    Creates an organization group
+    """
 
 
 @router.get(
     "/groups/{gid}",
-    response_model=Envelope[UsersGroup],
+    response_model=Envelope[GroupGet],
 )
-async def get_group(gid: GroupID):
-    ...
+async def get_group(_p: Annotated[_GroupPathParams, Depends()]):
+    """
+    Get an organization group
+    """
 
 
 @router.patch(
     "/groups/{gid}",
-    response_model=Envelope[UsersGroup],
+    response_model=Envelope[GroupGet],
 )
-async def update_group(gid: GroupID, _update: UsersGroup):
-    ...
+async def update_group(
+    _p: Annotated[_GroupPathParams, Depends()],
+    _b: GroupUpdate,
+):
+    """
+    Updates organization groups
+    """
 
 
 @router.delete(
     "/groups/{gid}",
     status_code=status.HTTP_204_NO_CONTENT,
 )
-async def delete_group(gid: GroupID):
-    ...
+async def delete_group(_p: Annotated[_GroupPathParams, Depends()]):
+    """
+    Deletes organization groups
+    """
 
 
 @router.get(
     "/groups/{gid}/users",
     response_model=Envelope[list[GroupUserGet]],
 )
-async def get_group_users(gid: GroupID):
-    ...
+async def get_all_group_users(_p: Annotated[_GroupPathParams, Depends()]):
+    """
+    Gets users in organization groups
+    """
 
 
 @router.post(
     "/groups/{gid}/users",
     status_code=status.HTTP_204_NO_CONTENT,
 )
 async def add_group_user(
-    gid: GroupID,
-    _new: GroupUserGet,
+    _p: Annotated[_GroupPathParams, Depends()],
+    _b: GroupUserAdd,
 ):
-    ...
+    """
+    Adds a user to an organization group
+    """
 
 
 @router.get(
     "/groups/{gid}/users/{uid}",
     response_model=Envelope[GroupUserGet],
 )
 async def get_group_user(
-    gid: GroupID,
-    uid: UserID,
+    _p: Annotated[_GroupUserPathParams, Depends()],
 ):
-    ...
+    """
+    Gets specific user in an organization group
+    """
 
 
 @router.patch(
     "/groups/{gid}/users/{uid}",
     response_model=Envelope[GroupUserGet],
 )
 async def update_group_user(
-    gid: GroupID,
-    uid: UserID,
-    _update: GroupUserGet,
+    _p: Annotated[_GroupUserPathParams, Depends()],
+    _b: GroupUserUpdate,
 ):
-    # FIXME: update type
-    ...
+    """
+    Updates user (access-rights) to an organization group
+    """
 
 
 @router.delete(
     "/groups/{gid}/users/{uid}",
     status_code=status.HTTP_204_NO_CONTENT,
 )
 async def delete_group_user(
-    gid: GroupID,
-    uid: UserID,
+    _p: Annotated[_GroupUserPathParams, Depends()],
 ):
-    ...
+    """
+    Removes a user from an organization group
+    """
+
+
+#
+# Classifiers
+#
 
 
 @router.get(
     "/groups/{gid}/classifiers",
     response_model=Envelope[dict[str, Any]],
 )
 async def get_group_classifiers(
-    gid: GroupID,
-    _query: Annotated[_ClassifiersQuery, Depends()],
+    _p: Annotated[_GroupPathParams, Depends()],
+    _q: Annotated[_ClassifiersQuery, Depends()],
 ):
     ...
 

packages/models-library/src/models_library/api_schemas_webserver/groups.py

@@ -1,13 +1,20 @@
 from contextlib import suppress
 from typing import Any, ClassVar
 
-from pydantic import AnyUrl, BaseModel, Field, ValidationError, parse_obj_as, validator
+from pydantic import (
+    AnyUrl,
+    BaseModel,
+    Field,
+    ValidationError,
+    parse_obj_as,
+    root_validator,
+    validator,
+)
 
 from ..emails import LowerCaseEmailStr
-
-#
-# GROUPS MODELS defined in OPENAPI specs
-#
+from ..users import UserID
+from ..utils.common_validators import create__check_only_one_is_set__root_validator
+from ._base import InputSchema, OutputSchema
 
 
 class GroupAccessRights(BaseModel):
@@ -29,7 +36,7 @@ class Config:
         }
 
 
-class UsersGroup(BaseModel):
+class GroupGet(OutputSchema):
     gid: int = Field(..., description="the group ID")
     label: str = Field(..., description="the group name")
     description: str = Field(..., description="the group description")
@@ -45,7 +52,7 @@ class UsersGroup(BaseModel):
 
     @validator("thumbnail", pre=True)
     @classmethod
-    def sanitize_legacy_data(cls, v):
+    def _sanitize_legacy_data(cls, v):
         if v:
             # Enforces null if thumbnail is not valid URL or empty
             with suppress(ValidationError):
@@ -86,11 +93,23 @@ class Config:
         }
 
 
-class AllUsersGroups(BaseModel):
-    me: UsersGroup | None = None
-    organizations: list[UsersGroup] | None = None
-    all: UsersGroup | None = None
-    product: UsersGroup | None = None
+class GroupCreate(InputSchema):
+    label: str
+    description: str
+    thumbnail: AnyUrl | None = None
+
+
+class GroupUpdate(InputSchema):
+    label: str | None = None
+    description: str | None = None
+    thumbnail: AnyUrl | None = None
+
+
+class MyGroupsGet(OutputSchema):
+    me: GroupGet
+    organizations: list[GroupGet] | None = None
+    all: GroupGet
+    product: GroupGet | None = None
 
     class Config:
         schema_extra: ClassVar[dict[str, Any]] = {
@@ -158,3 +177,38 @@ class Config:
                 },
             }
         }
+
+
+class GroupUserAdd(InputSchema):
+    """
+    Identify the user with either `email` or `uid` — only one.
+    """
+
+    uid: UserID | None = None
+    email: LowerCaseEmailStr | None = None
+
+    _check_uid_or_email = root_validator(allow_reuse=True)(
+        create__check_only_one_is_set__root_validator(["uid", "email"])
+    )
+
+    class Config:
+        schema_extra: ClassVar[dict[str, Any]] = {
+            "examples": [{"uid": 42}, {"email": "[email protected]"}]
+        }
+
+
+class GroupUserUpdate(InputSchema):
+    # NOTE: since it is a single item, it is required. Cannot
+    # update for the moment partial attributes e.g. {read: False}
+    access_rights: GroupAccessRights
+
+    class Config:
+        schema_extra: ClassVar[dict[str, Any]] = {
+            "example": {
+                "accessRights": {
+                    "read": True,
+                    "write": False,
+                    "delete": False,
+                },
+            }
+        }

packages/models-library/src/models_library/utils/common_validators.py

@@ -16,6 +16,8 @@ class MyModel(BaseModel):
 """
 
 import enum
+import functools
+import operator
 from typing import Any
 
 
@@ -69,3 +71,36 @@ def null_or_none_str_to_none_validator(value: Any):
     if isinstance(value, str) and value.lower() in ("null", "none"):
         return None
     return value
+
+
+def create__check_only_one_is_set__root_validator(alternative_field_names: list[str]):
+    """Ensure exactly one and only one of the alternatives is set
+
+    NOTE: a field is considered here `unset` when it is `not None`. When None
+    is used to indicate something else, please do not use this validator.
+
+    This is useful when you want to give the client alternative
+    ways to set the same thing e.g. set the user by email or id or username
+    and each of those has a different field
+
+    NOTE: Alternatevely, the previous example can also be solved using a
+    single field as `user: Email | UserID | UserName`
+
+    SEE test_uid_or_email_are_set.py for more details
+    """
+
+    def _validator(cls, values):
+        assert set(alternative_field_names).issubset(cls.__fields__)  # nosec
+
+        got = {
+            field_name: values.get(field_name) for field_name in alternative_field_names
+        }
+
+        if not functools.reduce(operator.xor, (v is not None for v in got.values())):
+            msg = (
+                f"Either { 'or'.join(got.keys()) } must be set, but not both. Got {got}"
+            )
+            raise ValueError(msg)
+        return values
+
+    return _validator

packages/models-library/tests/test_api_schemas_webserver_groups.py

@@ -0,0 +1,27 @@
+from typing import Any
+
+import pytest
+from models_library.api_schemas_webserver.groups import GroupUserAdd
+from pydantic import ValidationError
+
+unset = object()
+
+
+@pytest.mark.parametrize("uid", [1, None, unset])
+@pytest.mark.parametrize("email", ["[email protected]", None, unset])
+def test_uid_or_email_are_set(uid: Any, email: Any):
+    kwargs = {}
+    if uid != unset:
+        kwargs["uid"] = uid
+    if email != unset:
+        kwargs["email"] = email
+
+    none_are_defined = kwargs.get("uid") is None and kwargs.get("email") is None
+    both_are_defined = kwargs.get("uid") is not None and kwargs.get("email") is not None
+
+    if none_are_defined or both_are_defined:
+        with pytest.raises(ValidationError, match="not both"):
+            GroupUserAdd(**kwargs)
+    else:
+        got = GroupUserAdd(**kwargs)
+        assert bool(got.email) ^ bool(got.uid)