fixes nullable and is_literal

Author: pcrespov

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

@@ -1,22 +1,36 @@
 from types import UnionType
-from typing import Any, Literal, get_args, get_origin
+from typing import Annotated, Any, Literal, Union, get_args, get_origin
 
 from pydantic.fields import FieldInfo
 
+NoneType: type = type(None)
+
 
 def get_type(info: FieldInfo) -> Any:
     field_type = info.annotation
     if args := get_args(info.annotation):
-        field_type = next(a for a in args if a is not type(None))
+        field_type = next(a for a in args if a is not NoneType)
     return field_type
 
 
+def _unwrap_annotation(ann):
+    """Peel off Annotated wrappers until reaching the core type."""
+    while get_origin(ann) is Annotated:
+        ann = get_args(ann)[0]
+    return ann
+
+
 def is_literal(info: FieldInfo) -> bool:
-    return get_origin(info.annotation) is Literal
+    ann = _unwrap_annotation(info.annotation)
+    return get_origin(ann) is Literal
 
 
 def is_nullable(info: FieldInfo) -> bool:
-    origin = get_origin(info.annotation)  # X | None or Optional[X] will return Union
-    if origin is UnionType:
-        return any(x in get_args(info.annotation) for x in (type(None), Any))
-    return False
+    """Checks whether a field allows None as a value."""
+    ann = _unwrap_annotation(info.annotation)
+    origin = get_origin(ann)  # X | None or Optional[X] will return Union
+
+    if origin in (Union, UnionType):
+        return any(arg is NoneType or arg is Any for arg in get_args(ann))
+
+    return ann is NoneType or ann is Any

packages/common-library/tests/test_pydantic_fields_extension.py

@@ -1,9 +1,9 @@
 from collections.abc import Callable
-from typing import Any, Literal
+from typing import Annotated, Any, Literal
 
 import pytest
 from common_library.pydantic_fields_extension import get_type, is_literal, is_nullable
-from pydantic import BaseModel
+from pydantic import BaseModel, PositiveInt
 
 
 class MyModel(BaseModel):
@@ -12,6 +12,11 @@ class MyModel(BaseModel):
     c: str = "bla"
     d: bool | None = None
     e: Literal["bla"]
+    f: Annotated[
+        PositiveInt | None,
+        "nullable inside Annotated (PositiveInt = Annotated[int, ...])",
+    ]
+    g: Annotated[Literal["foo", "bar"], "literal inside Annotated"]
 
 
 @pytest.mark.parametrize(
@@ -50,6 +55,8 @@ class MyModel(BaseModel):
         ),
         (is_literal, False, "d"),
         (is_literal, True, "e"),
+        (is_literal, False, "f"),
+        (is_literal, True, "g"),
         (
             is_nullable,
             False,
@@ -67,6 +74,11 @@ class MyModel(BaseModel):
         ),
         (is_nullable, True, "d"),
         (is_nullable, False, "e"),
+        (
+            is_nullable,
+            True,
+            "f",
+        ),
     ],
 )
 def test_field_fn(fn: Callable[[Any], Any], expected: Any, name: str):

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

@@ -381,17 +381,21 @@ async def search_merged_pre_and_registered_users(
         users_pre_registration_details.c.state,
         users_pre_registration_details.c.postal_code,
         users_pre_registration_details.c.country,
-        users_pre_registration_details.c.user_id,  # user_id from pre-registration
+        users_pre_registration_details.c.user_id.label(
+            "pre_reg_user_id"
+        ),  # "copy" of user_id from pre-registration
         users_pre_registration_details.c.extras,
         users_pre_registration_details.c.account_request_status,
         users_pre_registration_details.c.account_request_reviewed_by,
         users_pre_registration_details.c.account_request_reviewed_at,
-        users.c.status,
         invited_by,
         account_request_reviewed_by_username,  # account_request_reviewed_by converted to username
         users_pre_registration_details.c.created,
+        # NOTE: some users have no pre-registration details (e.g. s4l-lite)
+        users.c.id.label("user_id"),  # real user_id from users table
         users.c.name.label("user_name"),
         users.c.primary_gid.label("user_primary_group_id"),
+        users.c.status,
     )
 
     left_outer_join = _build_left_outer_join_query(