♻️ web-server: Refactor users domain for improved layer separation and upgrading to asyncpg

api/specs/web-server/_users.py

@@ -7,27 +7,27 @@
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, status
-from models_library.api_schemas_webserver.users import MyProfileGet, MyProfilePatch
+from models_library.api_schemas_webserver.users import (
+    MyPermissionGet,
+    MyProfileGet,
+    MyProfilePatch,
+    MyTokenCreate,
+    MyTokenGet,
+    UserGet,
+    UsersSearchQueryParams,
+)
 from models_library.api_schemas_webserver.users_preferences import PatchRequestBody
 from models_library.generics import Envelope
 from models_library.user_preferences import PreferenceIdentifier
 from simcore_service_webserver._meta import API_VTAG
-from simcore_service_webserver.users._handlers import PreUserProfile, _SearchQueryParams
+from simcore_service_webserver.users._common.schemas import PreRegisteredUserGet
 from simcore_service_webserver.users._notifications import (
     UserNotification,
     UserNotificationCreate,
     UserNotificationPatch,
 )
-from simcore_service_webserver.users._notifications_handlers import (
-    _NotificationPathParams,
-)
-from simcore_service_webserver.users._schemas import UserProfile
-from simcore_service_webserver.users._tokens_handlers import _TokenPathParams
-from simcore_service_webserver.users.schemas import (
-    PermissionGet,
-    ThirdPartyToken,
-    TokenCreate,
-)
+from simcore_service_webserver.users._notifications_rest import _NotificationPathParams
+from simcore_service_webserver.users._tokens_rest import _TokenPathParams
 
 router = APIRouter(prefix=f"/{API_VTAG}", tags=["user"])
 
@@ -63,32 +63,32 @@ async def replace_my_profile(_profile: MyProfilePatch):
     status_code=status.HTTP_204_NO_CONTENT,
 )
 async def set_frontend_preference(
-    preference_id: PreferenceIdentifier,  # noqa: ARG001
-    body_item: PatchRequestBody,  # noqa: ARG001
+    preference_id: PreferenceIdentifier,
+    body_item: PatchRequestBody,
 ):
     ...
 
 
 @router.get(
     "/me/tokens",
-    response_model=Envelope[list[ThirdPartyToken]],
+    response_model=Envelope[list[MyTokenGet]],
 )
 async def list_tokens():
     ...
 
 
 @router.post(
     "/me/tokens",
-    response_model=Envelope[ThirdPartyToken],
+    response_model=Envelope[MyTokenGet],
     status_code=status.HTTP_201_CREATED,
 )
-async def create_token(_token: TokenCreate):
+async def create_token(_token: MyTokenCreate):
     ...
 
 
 @router.get(
     "/me/tokens/{service}",
-    response_model=Envelope[ThirdPartyToken],
+    response_model=Envelope[MyTokenGet],
 )
 async def get_token(_params: Annotated[_TokenPathParams, Depends()]):
     ...
@@ -131,30 +131,30 @@ async def mark_notification_as_read(
 
 @router.get(
     "/me/permissions",
-    response_model=Envelope[list[PermissionGet]],
+    response_model=Envelope[list[MyPermissionGet]],
 )
 async def list_user_permissions():
     ...
 
 
 @router.get(
     "/users:search",
-    response_model=Envelope[list[UserProfile]],
+    response_model=Envelope[list[UserGet]],
     tags=[
         "po",
     ],
 )
-async def search_users(_params: Annotated[_SearchQueryParams, Depends()]):
+async def search_users(_params: Annotated[UsersSearchQueryParams, Depends()]):
     # NOTE: see `Search` in `Common Custom Methods` in https://cloud.google.com/apis/design/custom_methods
     ...
 
 
 @router.post(
     "/users:pre-register",
-    response_model=Envelope[UserProfile],
+    response_model=Envelope[UserGet],
     tags=[
         "po",
     ],
 )
-async def pre_register_user(_body: PreUserProfile):
+async def pre_register_user(_body: PreRegisteredUserGet):
     ...

packages/common-library/src/common_library/users_enums.py

@@ -0,0 +1,59 @@
+from enum import Enum
+from functools import total_ordering
+
+_USER_ROLE_TO_LEVEL = {
+    "ANONYMOUS": 0,
+    "GUEST": 10,
+    "USER": 20,
+    "TESTER": 30,
+    "PRODUCT_OWNER": 40,
+    "ADMIN": 100,
+}
+
+
+@total_ordering
+class UserRole(Enum):
+    """SORTED enumeration of user roles
+
+    A role defines a set of privileges the user can perform
+    Roles are sorted from lower to highest privileges
+    USER is the role assigned by default A user with a higher/lower role is denoted super/infra user
+
+    ANONYMOUS : The user is not logged in
+    GUEST     : Temporary user with very limited access. Main used for demos and for a limited amount of time
+    USER      : Registered user. Basic permissions to use the platform [default]
+    TESTER    : Upgraded user. First level of super-user with privileges to test the framework.
+                Can use everything but does not have an effect in other users or actual data
+    ADMIN     : Framework admin.
+
+    See security_access.py
+    """
+
+    ANONYMOUS = "ANONYMOUS"
+    GUEST = "GUEST"
+    USER = "USER"
+    TESTER = "TESTER"
+    PRODUCT_OWNER = "PRODUCT_OWNER"
+    ADMIN = "ADMIN"
+
+    @property
+    def privilege_level(self) -> int:
+        return _USER_ROLE_TO_LEVEL[self.name]
+
+    def __lt__(self, other: "UserRole") -> bool:
+        if self.__class__ is other.__class__:
+            return self.privilege_level < other.privilege_level
+        return NotImplemented
+
+
+class UserStatus(str, Enum):
+    # This is a transition state. The user is registered but not confirmed. NOTE that state is optional depending on LOGIN_REGISTRATION_CONFIRMATION_REQUIRED
+    CONFIRMATION_PENDING = "CONFIRMATION_PENDING"
+    # This user can now operate the platform
+    ACTIVE = "ACTIVE"
+    # This user is inactive because it expired after a trial period
+    EXPIRED = "EXPIRED"
+    # This user is inactive because he has been a bad boy
+    BANNED = "BANNED"
+    # This user is inactive because it was marked for deletion
+    DELETED = "DELETED"

packages/common-library/tests/test_users_enums.py

@@ -0,0 +1,79 @@
+# pylint: disable=no-value-for-parameter
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+
+from common_library.users_enums import _USER_ROLE_TO_LEVEL, UserRole
+
+
+def test_user_role_to_level_map_in_sync():
+    # If fails, then update _USER_ROLE_TO_LEVEL map
+    assert set(_USER_ROLE_TO_LEVEL.keys()) == set(UserRole.__members__.keys())
+
+
+def test_user_roles_compares_to_admin():
+    assert UserRole.ANONYMOUS < UserRole.ADMIN
+    assert UserRole.GUEST < UserRole.ADMIN
+    assert UserRole.USER < UserRole.ADMIN
+    assert UserRole.TESTER < UserRole.ADMIN
+    assert UserRole.PRODUCT_OWNER < UserRole.ADMIN
+    assert UserRole.ADMIN == UserRole.ADMIN
+
+
+def test_user_roles_compares_to_product_owner():
+    assert UserRole.ANONYMOUS < UserRole.PRODUCT_OWNER
+    assert UserRole.GUEST < UserRole.PRODUCT_OWNER
+    assert UserRole.USER < UserRole.PRODUCT_OWNER
+    assert UserRole.TESTER < UserRole.PRODUCT_OWNER
+    assert UserRole.PRODUCT_OWNER == UserRole.PRODUCT_OWNER
+    assert UserRole.ADMIN > UserRole.PRODUCT_OWNER
+
+
+def test_user_roles_compares_to_tester():
+    assert UserRole.ANONYMOUS < UserRole.TESTER
+    assert UserRole.GUEST < UserRole.TESTER
+    assert UserRole.USER < UserRole.TESTER
+    assert UserRole.TESTER == UserRole.TESTER
+    assert UserRole.PRODUCT_OWNER > UserRole.TESTER
+    assert UserRole.ADMIN > UserRole.TESTER
+
+
+def test_user_roles_compares_to_user():
+    assert UserRole.ANONYMOUS < UserRole.USER
+    assert UserRole.GUEST < UserRole.USER
+    assert UserRole.USER == UserRole.USER
+    assert UserRole.TESTER > UserRole.USER
+    assert UserRole.PRODUCT_OWNER > UserRole.USER
+    assert UserRole.ADMIN > UserRole.USER
+
+
+def test_user_roles_compares_to_guest():
+    assert UserRole.ANONYMOUS < UserRole.GUEST
+    assert UserRole.GUEST == UserRole.GUEST
+    assert UserRole.USER > UserRole.GUEST
+    assert UserRole.TESTER > UserRole.GUEST
+    assert UserRole.PRODUCT_OWNER > UserRole.GUEST
+    assert UserRole.ADMIN > UserRole.GUEST
+
+
+def test_user_roles_compares_to_anonymous():
+    assert UserRole.ANONYMOUS == UserRole.ANONYMOUS
+    assert UserRole.GUEST > UserRole.ANONYMOUS
+    assert UserRole.USER > UserRole.ANONYMOUS
+    assert UserRole.TESTER > UserRole.ANONYMOUS
+    assert UserRole.PRODUCT_OWNER > UserRole.ANONYMOUS
+    assert UserRole.ADMIN > UserRole.ANONYMOUS
+
+
+def test_user_roles_compares():
+    # < and >
+    assert UserRole.TESTER < UserRole.ADMIN
+    assert UserRole.ADMIN > UserRole.TESTER
+
+    # >=, == and <=
+    assert UserRole.TESTER <= UserRole.ADMIN
+    assert UserRole.ADMIN >= UserRole.TESTER
+
+    assert UserRole.ADMIN <= UserRole.ADMIN
+    assert UserRole.ADMIN == UserRole.ADMIN

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

@@ -29,6 +29,14 @@ class InputSchema(BaseModel):
     )
 
 
+class OutputSchemaWithoutCamelCase(BaseModel):
+    model_config = ConfigDict(
+        populate_by_name=True,
+        extra="ignore",
+        frozen=True,
+    )
+
+
 class OutputSchema(BaseModel):
     model_config = ConfigDict(
         alias_generator=snake_to_camel,
@@ -70,3 +78,12 @@ def data_json(
             exclude_none=exclude_none,
             **kwargs,
         )
+
+
+#
+# mapping tools
+#
+
+
+def copy_dict(data: dict, update_keys: dict[str, str]) -> dict[str, Any]:
+    return {update_keys.get(k, k): v for k, v in data.items()}