common

Author: pcrespov

packages/postgres-database/src/simcore_postgres_database/utils_users.py

@@ -10,6 +10,7 @@
 import sqlalchemy as sa
 from aiopg.sa.connection import SAConnection
 from aiopg.sa.result import RowProxy
+from sqlalchemy import Column
 
 from .errors import UniqueViolation
 from .models.users import UserRole, UserStatus, users
@@ -214,7 +215,17 @@ async def is_email_used(conn: SAConnection, email: str) -> bool:
                 users_pre_registration_details.c.pre_email == email
             )
         )
-        if pre_registered:
-            return True
+        return bool(pre_registered)
+
+
+#
+# Privacy settings
+#
+
+
+def is_private(hide_attribute: Column, caller_id: int):
+    return hide_attribute.is_(True) & (users.c.id != caller_id)
+
 
-        return False
+def is_public(hide_attribute: Column, caller_id: int):
+    return hide_attribute.is_(False) | (users.c.id == caller_id)

services/web/server/src/simcore_service_webserver/groups/_groups_repository.py

@@ -25,6 +25,7 @@
     pass_or_acquire_connection,
     transaction_context,
 )
+from simcore_postgres_database.utils_users import is_private, is_public
 from sqlalchemy import and_
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.engine.row import Row
@@ -417,10 +418,7 @@ async def get_user_from_email(
         result = await conn.stream(
             sa.select(users.c.id).where(
                 (users.c.email == email)
-                & (
-                    users.c.privacy_hide_email.is_(False)
-                    | (users.c.id == caller_user_id)
-                )
+                & is_public(users.c.privacy_hide_email, caller_id=caller_user_id)
             )
         )
         user = await result.fetchone()
@@ -434,30 +432,28 @@ async def get_user_from_email(
 #
 
 
-def _group_user_cols(caller_user_id: int):
+def _group_user_cols(caller_id: UserID):
     return (
         users.c.id,
         users.c.name,
         # privacy settings
         sa.case(
             (
-                users.c.privacy_hide_email.is_(True) & (users.c.id != caller_user_id),
+                is_private(users.c.privacy_hide_email, caller_id),
                 None,
             ),
             else_=users.c.email,
         ).label("email"),
         sa.case(
             (
-                users.c.privacy_hide_fullname.is_(True)
-                & (users.c.id != caller_user_id),
+                is_private(users.c.privacy_hide_fullname, caller_id),
                 None,
             ),
             else_=users.c.first_name,
         ).label("first_name"),
         sa.case(
             (
-                users.c.privacy_hide_fullname.is_(True)
-                & (users.c.id != caller_user_id),
+                is_private(users.c.privacy_hide_fullname, caller_id),
                 None,
             ),
             else_=users.c.last_name,

services/web/server/src/simcore_service_webserver/users/_users_repository.py

@@ -30,8 +30,10 @@
 from simcore_postgres_database.utils_users import (
     UsersRepo,
     generate_alternative_username,
+    is_private,
+    is_public,
 )
-from sqlalchemy import Column, delete
+from sqlalchemy import delete
 from sqlalchemy.engine.row import Row
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
@@ -52,42 +54,29 @@ def _parse_as_user(user_id: Any) -> UserID:
         raise UserNotFoundError(uid=user_id, user_id=user_id) from err
 
 
-#
-# Privacy settings
-#
-
-
-def _is_private(hide_attribute: Column, caller_id: UserID):
-    return hide_attribute.is_(True) & (users.c.id != caller_id)
-
-
-def _is_public(hide_attribute: Column, caller_id: UserID):
-    return hide_attribute.is_(False) | (users.c.id == caller_id)
-
-
-def _public_user_cols(caller_id: UserID):
+def _public_user_cols(caller_id: int):
     return (
         # Fits PublicUser model
         users.c.id.label("user_id"),
         users.c.name.label("user_name"),
         # privacy settings
         sa.case(
             (
-                _is_private(users.c.privacy_hide_email, caller_id),
+                is_private(users.c.privacy_hide_email, caller_id),
                 None,
             ),
             else_=users.c.email,
         ).label("email"),
         sa.case(
             (
-                _is_private(users.c.privacy_hide_fullname, caller_id),
+                is_private(users.c.privacy_hide_fullname, caller_id),
                 None,
             ),
             else_=users.c.first_name,
         ).label("first_name"),
         sa.case(
             (
-                _is_private(users.c.privacy_hide_fullname, caller_id),
+                is_private(users.c.privacy_hide_fullname, caller_id),
                 None,
             ),
             else_=users.c.last_name,
@@ -136,11 +125,11 @@ async def search_public_user(
         .where(
             users.c.name.ilike(_pattern)
             | (
-                _is_public(users.c.privacy_hide_email, caller_id)
+                is_public(users.c.privacy_hide_email, caller_id)
                 & users.c.email.ilike(_pattern)
             )
             | (
-                _is_public(users.c.privacy_hide_fullname, caller_id)
+                is_public(users.c.privacy_hide_fullname, caller_id)
                 & (
                     users.c.first_name.ilike(_pattern)
                     | users.c.last_name.ilike(_pattern)

services/web/server/tests/unit/with_dbs/03/test_users.py

@@ -143,7 +143,7 @@ async def test_search_users(
     assert len(found) == 1
     assert semi_private_user["name"] == found[0].user_name
     assert found[0].first_name == semi_private_user.get("first_name")
-    assert found[0].last_name == semi_private_user.get("first_name")
+    assert found[0].last_name == semi_private_user.get("last_name")
     assert found[0].email is None
 
     # SEARCH by partial email