model conversion ready

Author: pcrespov

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

@@ -1,4 +1,6 @@
-from pydantic import BaseModel
+from typing import Annotated, Any, Self
+
+from pydantic import BaseModel, ConfigDict, Field
 
 
 #
@@ -12,3 +14,48 @@ class ProfilePrivacyGet(BaseModel):
 class ProfilePrivacyUpdate(BaseModel):
     hide_fullname: bool | None = None
     hide_email: bool | None = None
+
+
+#
+# DB models
+#
+
+
+def flatten_dict(d: dict, parent_key="", sep="_"):
+    items = []
+    for key, value in d.items():
+        new_key = f"{parent_key}{sep}{key}" if parent_key else key
+        if isinstance(value, dict):
+            # Recursively process nested dictionaries
+            items.extend(flatten_dict(value, new_key, sep=sep).items())
+        else:
+            items.append((new_key, value))
+    return dict(items)
+
+
+class ToUserUpdateDB(BaseModel):
+    """
+    Maps ProfileUpdate api model into UserDB db model
+    """
+
+    # NOTE: field names are UserDB columns
+    # NOTE: aliases are ProfileUpdate field names
+
+    name: Annotated[str | None, Field(alias="user_name")] = None
+    first_name: str | None = None
+    last_name: str | None = None
+
+    privacy_hide_fullname: bool | None = None
+    privacy_hide_email: bool | None = None
+
+    model_config = ConfigDict(extra="forbid")
+
+    @classmethod
+    def from_api(cls, profile_update) -> Self:
+        # The mapping of embed fields to flatten keys is done here
+        return cls.model_validate(
+            flatten_dict(profile_update.model_dump(exclude_unset=True, by_alias=False))
+        )
+
+    def to_columns(self) -> dict[str, Any]:
+        return self.model_dump(exclude_unset=True, by_alias=False)

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

@@ -30,6 +30,7 @@
 from ..security.api import clean_auth_policy_cache
 from . import _db
 from ._api import get_user_credentials, get_user_invoice_address, set_user_as_deleted
+from ._models import ToUserUpdateDB
 from ._preferences_api import get_frontend_user_preferences_aggregation
 from .exceptions import MissingGroupExtraPropertiesForProductError, UserNotFoundError
 from .schemas import ProfileGet, ProfileUpdate
@@ -159,26 +160,12 @@ async def update_user_profile(
         UserNotFoundError
     """
     user_id = _parse_as_user(user_id)
-    async with get_database_engine(app).acquire() as conn:
-        updated_values = update.model_dump(
-            include={
-                "first_name": True,
-                "last_name": True,
-                "user_name": True,
-                "privacy": {
-                    "hide_email",
-                    "hide_fullname",
-                },
-            },
-            exclude_unset=True,
-        )
-        # flatten dict
-        if privacy := updated_values.pop("privacy", None):
-            updated_values |= {f"privacy_{k}": v for k, v in privacy.items()}
 
-        query = users.update().where(users.c.id == user_id).values(**updated_values)
-        resp = await conn.execute(query)
-        assert resp.rowcount == 1  # nosec
+    if updated_values := ToUserUpdateDB.from_api(update).to_columns():
+        async with get_database_engine(app).acquire() as conn:
+            query = users.update().where(users.c.id == user_id).values(**updated_values)
+            resp = await conn.execute(query)
+            assert resp.rowcount == 1  # nosec
 
 
 async def get_user_role(app: web.Application, user_id: UserID) -> UserRole:

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

@@ -104,7 +104,8 @@ def _to_upper_string(cls, v):
 class ProfileUpdate(BaseModel):
     first_name: FirstNameStr | None = None
     last_name: LastNameStr | None = None
-    user_name: IDStr | None = None
+    user_name: Annotated[IDStr | None, Field(alias="userName")] = None
+
     privacy: ProfilePrivacyUpdate | None = None
 
     model_config = ConfigDict(

services/web/server/tests/unit/isolated/test_users_models.py

@@ -15,8 +15,12 @@
 from pydantic import BaseModel
 from servicelib.rest_constants import RESPONSE_MODEL_POLICY
 from simcore_postgres_database.models.users import UserRole
-from simcore_service_webserver.users._models import ProfilePrivacyGet
-from simcore_service_webserver.users.schemas import ProfileGet, ThirdPartyToken
+from simcore_service_webserver.users._models import ProfilePrivacyGet, ToUserUpdateDB
+from simcore_service_webserver.users.schemas import (
+    ProfileGet,
+    ProfileUpdate,
+    ThirdPartyToken,
+)
 
 
 @pytest.mark.parametrize(
@@ -152,3 +156,25 @@ def test_parsing_output_of_get_user_profile():
 
     profile = ProfileGet.model_validate(result_from_db_query_and_composition)
     assert "password" not in profile.model_dump(exclude_unset=True)
+
+
+def test_mapping_update_models_from_rest_to_db():
+
+    profile_update = ProfileUpdate.model_validate(
+        # input in rest
+        {
+            "first_name": "foo",
+            "userName": "foo1234",
+            "privacy": {"hide_fullname": False},
+        }
+    )
+
+    # to db
+    profile_update_db = ToUserUpdateDB.from_api(profile_update)
+
+    # expected
+    assert profile_update_db.to_columns() == {
+        "first_name": "foo",
+        "name": "foo1234",
+        "privacy_hide_fullname": False,
+    }