moves schemas to models_library

Author: pcrespov

api/specs/web-server/_users.py

@@ -7,7 +7,13 @@
 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 (
+    MyProfileGet,
+    MyProfilePatch,
+    PreRegisteredUserGet,
+    SearchQueryParams,
+    UserGet,
+)
 from models_library.api_schemas_webserver.users_preferences import PatchRequestBody
 from models_library.generics import Envelope
 from models_library.user_preferences import PreferenceIdentifier
@@ -20,11 +26,6 @@
 from simcore_service_webserver.users._notifications_handlers import (
     _NotificationPathParams,
 )
-from simcore_service_webserver.users._schemas import (
-    PreUserProfile,
-    SearchQueryParams,
-    UserProfile,
-)
 from simcore_service_webserver.users._tokens_handlers import _TokenPathParams
 from simcore_service_webserver.users.schemas import (
     PermissionGet,
@@ -66,8 +67,8 @@ 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,
 ):
     ...
 
@@ -142,7 +143,7 @@ async def list_user_permissions():
 
 @router.get(
     "/users:search",
-    response_model=Envelope[list[UserProfile]],
+    response_model=Envelope[list[UserGet]],
     tags=[
         "po",
     ],
@@ -154,10 +155,10 @@ async def search_users(_params: Annotated[SearchQueryParams, Depends()]):
 
 @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/models-library/src/models_library/api_schemas_webserver/users.py

@@ -1,14 +1,27 @@
 import re
+import sys
+from contextlib import suppress
 from datetime import date
 from enum import Enum
-from typing import Annotated, Literal
+from typing import Annotated, Any, Final, Literal
 
+import pycountry
+from models_library.api_schemas_webserver._base import InputSchema, OutputSchema
 from models_library.api_schemas_webserver.groups import MyGroupsGet
 from models_library.api_schemas_webserver.users_preferences import AggregatedPreferences
 from models_library.basic_types import IDStr
 from models_library.emails import LowerCaseEmailStr
+from models_library.products import ProductName
 from models_library.users import FirstNameStr, LastNameStr, UserID
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    ValidationInfo,
+    field_validator,
+    model_validator,
+)
+from simcore_postgres_database.models.users import UserStatus
 
 from ._base import InputSchema, OutputSchema
 
@@ -128,3 +141,132 @@ def _validate_user_name(cls, value: str):
             raise ValueError(msg)
 
         return value
+
+
+class SearchQueryParams(BaseModel):
+    email: str = Field(
+        min_length=3,
+        max_length=200,
+        description="complete or glob pattern for an email",
+    )
+
+
+class UserGet(OutputSchema):
+    first_name: str | None
+    last_name: str | None
+    email: LowerCaseEmailStr
+    institution: str | None
+    phone: str | None
+    address: str | None
+    city: str | None
+    state: str | None = Field(description="State, province, canton, ...")
+    postal_code: str | None
+    country: str | None
+    extras: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Keeps extra information provided in the request form",
+    )
+
+    # authorization
+    invited_by: str | None = Field(default=None)
+
+    # user status
+    registered: bool
+    status: UserStatus | None
+    products: list[ProductName] | None = Field(
+        default=None,
+        description="List of products this users is included or None if fields is unset",
+    )
+
+    @field_validator("status")
+    @classmethod
+    def _consistency_check(cls, v, info: ValidationInfo):
+        registered = info.data["registered"]
+        status = v
+        if not registered and status is not None:
+            msg = f"{registered=} and {status=} is not allowed"
+            raise ValueError(msg)
+        return v
+
+
+MAX_BYTES_SIZE_EXTRAS: Final[int] = 512
+
+
+class PreRegisteredUserGet(InputSchema):
+    first_name: str
+    last_name: str
+    email: LowerCaseEmailStr
+    institution: str | None = Field(
+        default=None, description="company, university, ..."
+    )
+    phone: str | None
+    # billing details
+    address: str
+    city: str
+    state: str | None = Field(default=None)
+    postal_code: str
+    country: str
+    extras: Annotated[
+        dict[str, Any],
+        Field(
+            default_factory=dict,
+            description="Keeps extra information provided in the request form. At most MAX_NUM_EXTRAS fields",
+        ),
+    ]
+
+    model_config = ConfigDict(str_strip_whitespace=True, str_max_length=200)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _preprocess_aliases_and_extras(cls, values):
+        # multiple aliases for "institution"
+        alias_by_priority = ("companyName", "company", "university", "universityName")
+        if "institution" not in values:
+
+            for alias in alias_by_priority:
+                if alias in values:
+                    values["institution"] = values.pop(alias)
+
+        # collect extras
+        extra_fields = {}
+        field_names_and_aliases = (
+            set(cls.model_fields.keys())
+            | {f.alias for f in cls.model_fields.values() if f.alias}
+            | set(alias_by_priority)
+        )
+        for key, value in values.items():
+            if key not in field_names_and_aliases:
+                extra_fields[key] = value
+                if sys.getsizeof(extra_fields) > MAX_BYTES_SIZE_EXTRAS:
+                    extra_fields.pop(key)
+                    break
+
+        for key in extra_fields:
+            values.pop(key)
+
+        values.setdefault("extras", {})
+        values["extras"].update(extra_fields)
+
+        return values
+
+    @field_validator("first_name", "last_name", "institution", mode="before")
+    @classmethod
+    def _pre_normalize_given_names(cls, v):
+        if v:
+            with suppress(Exception):  # skip if funny characters
+                name = re.sub(r"\s+", " ", v)
+                return re.sub(r"\b\w+\b", lambda m: m.group(0).capitalize(), name)
+        return v
+
+    @field_validator("country", mode="before")
+    @classmethod
+    def _pre_check_and_normalize_country(cls, v):
+        if v:
+            try:
+                return pycountry.countries.lookup(v).name
+            except LookupError as err:
+                raise ValueError(v) from err
+        return v
+
+
+assert set(PreRegisteredUserGet.model_fields).issubset(UserGet.model_fields)  # nosec

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

@@ -2,159 +2,14 @@
 
 """
 
-import re
-import sys
-from contextlib import suppress
-from typing import Annotated, Any, Final
 
-import pycountry
-from models_library.api_schemas_webserver._base import InputSchema, OutputSchema
-from models_library.emails import LowerCaseEmailStr
-from models_library.products import ProductName
 from models_library.users import UserID
-from pydantic import (
-    BaseModel,
-    ConfigDict,
-    Field,
-    ValidationInfo,
-    field_validator,
-    model_validator,
-)
+from pydantic import BaseModel, Field
 from servicelib.request_keys import RQT_USERID_KEY
-from simcore_postgres_database.models.users import UserStatus
 
 from .._constants import RQ_PRODUCT_KEY
 
 
 class UsersRequestContext(BaseModel):
     user_id: UserID = Field(..., alias=RQT_USERID_KEY)  # type: ignore[literal-required]
     product_name: str = Field(..., alias=RQ_PRODUCT_KEY)  # type: ignore[literal-required]
-
-
-class SearchQueryParams(BaseModel):
-    email: str = Field(
-        min_length=3,
-        max_length=200,
-        description="complete or glob pattern for an email",
-    )
-
-
-class UserProfile(OutputSchema):
-    first_name: str | None
-    last_name: str | None
-    email: LowerCaseEmailStr
-    institution: str | None
-    phone: str | None
-    address: str | None
-    city: str | None
-    state: str | None = Field(description="State, province, canton, ...")
-    postal_code: str | None
-    country: str | None
-    extras: dict[str, Any] = Field(
-        default_factory=dict,
-        description="Keeps extra information provided in the request form",
-    )
-
-    # authorization
-    invited_by: str | None = Field(default=None)
-
-    # user status
-    registered: bool
-    status: UserStatus | None
-    products: list[ProductName] | None = Field(
-        default=None,
-        description="List of products this users is included or None if fields is unset",
-    )
-
-    @field_validator("status")
-    @classmethod
-    def _consistency_check(cls, v, info: ValidationInfo):
-        registered = info.data["registered"]
-        status = v
-        if not registered and status is not None:
-            msg = f"{registered=} and {status=} is not allowed"
-            raise ValueError(msg)
-        return v
-
-
-MAX_BYTES_SIZE_EXTRAS: Final[int] = 512
-
-
-class PreUserProfile(InputSchema):
-    first_name: str
-    last_name: str
-    email: LowerCaseEmailStr
-    institution: str | None = Field(
-        default=None, description="company, university, ..."
-    )
-    phone: str | None
-    # billing details
-    address: str
-    city: str
-    state: str | None = Field(default=None)
-    postal_code: str
-    country: str
-    extras: Annotated[
-        dict[str, Any],
-        Field(
-            default_factory=dict,
-            description="Keeps extra information provided in the request form. At most MAX_NUM_EXTRAS fields",
-        ),
-    ]
-
-    model_config = ConfigDict(str_strip_whitespace=True, str_max_length=200)
-
-    @model_validator(mode="before")
-    @classmethod
-    def _preprocess_aliases_and_extras(cls, values):
-        # multiple aliases for "institution"
-        alias_by_priority = ("companyName", "company", "university", "universityName")
-        if "institution" not in values:
-
-            for alias in alias_by_priority:
-                if alias in values:
-                    values["institution"] = values.pop(alias)
-
-        # collect extras
-        extra_fields = {}
-        field_names_and_aliases = (
-            set(cls.model_fields.keys())
-            | {f.alias for f in cls.model_fields.values() if f.alias}
-            | set(alias_by_priority)
-        )
-        for key, value in values.items():
-            if key not in field_names_and_aliases:
-                extra_fields[key] = value
-                if sys.getsizeof(extra_fields) > MAX_BYTES_SIZE_EXTRAS:
-                    extra_fields.pop(key)
-                    break
-
-        for key in extra_fields:
-            values.pop(key)
-
-        values.setdefault("extras", {})
-        values["extras"].update(extra_fields)
-
-        return values
-
-    @field_validator("first_name", "last_name", "institution", mode="before")
-    @classmethod
-    def _pre_normalize_given_names(cls, v):
-        if v:
-            with suppress(Exception):  # skip if funny characters
-                name = re.sub(r"\s+", " ", v)
-                return re.sub(r"\b\w+\b", lambda m: m.group(0).capitalize(), name)
-        return v
-
-    @field_validator("country", mode="before")
-    @classmethod
-    def _pre_check_and_normalize_country(cls, v):
-        if v:
-            try:
-                return pycountry.countries.lookup(v).name
-            except LookupError as err:
-                raise ValueError(v) from err
-        return v
-
-
-assert set(PreUserProfile.model_fields).issubset(UserProfile.model_fields)  # nosec