fix code

Author: giancarloromeo

services/invitations/src/simcore_service_invitations/cli.py

@@ -5,7 +5,7 @@
 from cryptography.fernet import Fernet
 from models_library.emails import LowerCaseEmailStr
 from models_library.invitations import InvitationContent, InvitationInputs
-from pydantic import EmailStr, HttpUrl, ValidationError, parse_obj_as
+from pydantic import EmailStr, HttpUrl, TypeAdapter, ValidationError
 from rich.console import Console
 from servicelib.utils_secrets import generate_password
 from settings_library.utils_cli import (
@@ -96,19 +96,19 @@ def invite(
     ctx: typer.Context,
     email: str = typer.Argument(
         ...,
-        callback=lambda v: parse_obj_as(LowerCaseEmailStr, v),
+        callback=lambda v: TypeAdapter(LowerCaseEmailStr).validate_python(v),
         help="Custom invitation for a given guest",
     ),
     issuer: str = typer.Option(
-        ..., help=InvitationInputs.__fields__["issuer"].field_info.description
+        ..., help=InvitationInputs.model_fields["issuer"].description
     ),
     trial_account_days: int = typer.Option(
         None,
-        help=InvitationInputs.__fields__["trial_account_days"].field_info.description,
+        help=InvitationInputs.model_fields["trial_account_days"].description,
     ),
     product: str = typer.Option(
         None,
-        help=InvitationInputs.__fields__["product"].field_info.description,
+        help=InvitationInputs.model_fields["product"].description,
     ),
 ):
     """Creates an invitation link for user with 'email' and issued by 'issuer'"""
@@ -117,7 +117,7 @@ def invite(
 
     invitation_data = InvitationInputs(
         issuer=issuer,
-        guest=parse_obj_as(EmailStr, email),
+        guest=TypeAdapter(EmailStr).validate_python(email),
         trial_account_days=trial_account_days,
         extra_credits_in_usd=None,
         product=product,
@@ -142,14 +142,14 @@ def extract(ctx: typer.Context, invitation_url: str):
     try:
         invitation: InvitationContent = extract_invitation_content(
             invitation_code=extract_invitation_code_from_query(
-                parse_obj_as(HttpUrl, invitation_url)
+                TypeAdapter(HttpUrl).validate_python(invitation_url)
             ),
             secret_key=settings.INVITATIONS_SECRET_KEY.get_secret_value().encode(),
             default_product=settings.INVITATIONS_DEFAULT_PRODUCT,
         )
         assert invitation.product is not None  # nosec
 
-        print(invitation.json(indent=1))  # noqa: T201
+        print(invitation.model_dump_json(indent=1))  # noqa: T201
 
     except (InvalidInvitationCodeError, ValidationError):
         _err_console.print("[bold red]Invalid code[/bold red]")

services/invitations/src/simcore_service_invitations/core/settings.py

@@ -1,7 +1,14 @@
 from functools import cached_property
 
 from models_library.products import ProductName
-from pydantic import Field, HttpUrl, PositiveInt, SecretStr, validator
+from pydantic import (
+    AliasChoices,
+    Field,
+    HttpUrl,
+    PositiveInt,
+    SecretStr,
+    field_validator,
+)
 from settings_library.base import BaseCustomSettings
 from settings_library.basic_types import BuildTargetEnum, LogLevel, VersionTag
 from settings_library.utils_logging import MixinLoggingSettings
@@ -38,22 +45,23 @@ class _BaseApplicationSettings(BaseCustomSettings, MixinLoggingSettings):
     # RUNTIME  -----------------------------------------------------------
 
     INVITATIONS_LOGLEVEL: LogLevel = Field(
-        default=LogLevel.INFO, env=["INVITATIONS_LOGLEVEL", "LOG_LEVEL", "LOGLEVEL"]
+        default=LogLevel.INFO,
+        validation_alias=AliasChoices("INVITATIONS_LOGLEVEL", "LOG_LEVEL", "LOGLEVEL"),
     )
     INVITATIONS_LOG_FORMAT_LOCAL_DEV_ENABLED: bool = Field(
         default=False,
-        env=[
+        validation_alias=AliasChoices(
             "INVITATIONS_LOG_FORMAT_LOCAL_DEV_ENABLED",
             "LOG_FORMAT_LOCAL_DEV_ENABLED",
-        ],
+        ),
         description="Enables local development log format. WARNING: make sure it is disabled if you want to have structured logs!",
     )
 
     @cached_property
     def LOG_LEVEL(self):
         return self.INVITATIONS_LOGLEVEL
 
-    @validator("INVITATIONS_LOGLEVEL")
+    @field_validator("INVITATIONS_LOGLEVEL")
     @classmethod
     def valid_log_level(cls, value: str) -> str:
         return cls.validate_log_level(value)

services/invitations/src/simcore_service_invitations/services/invitations.py

@@ -1,18 +1,22 @@
 import base64
 import binascii
 import logging
-from typing import Any, ClassVar, cast
 from urllib import parse
 
+from common_library.pydantic_networks_extension import HttpUrlLegacy
 from cryptography.fernet import Fernet, InvalidToken
 from models_library.invitations import InvitationContent, InvitationInputs
 from models_library.products import ProductName
-from pydantic import HttpUrl, ValidationError, parse_obj_as
+from pydantic import ConfigDict, HttpUrl, TypeAdapter, ValidationError
 from starlette.datastructures import URL
 
 _logger = logging.getLogger(__name__)
 
 
+def _to_initial(v: str):
+    return v[0]
+
+
 class InvalidInvitationCodeError(Exception):
     ...
 
@@ -23,9 +27,9 @@ class _ContentWithShortNames(InvitationContent):
     @classmethod
     def serialize(cls, model_obj: InvitationContent) -> str:
         """Exports to json using *short* aliases and values in order to produce shorter codes"""
-        model_w_short_aliases_json: str = cls.construct(
-            **model_obj.dict(exclude_unset=True)
-        ).json(exclude_unset=True, by_alias=True)
+        model_w_short_aliases_json: str = cls.model_construct(
+            **model_obj.model_dump(exclude_unset=True)
+        ).model_dump_json(exclude_unset=True, by_alias=True)
         # NOTE: json arguments try to minimize the amount of data
         # serialized. The CONS is that it relies on models in the code
         # that might change over time. This might lead to some datasets in codes
@@ -35,36 +39,18 @@ def serialize(cls, model_obj: InvitationContent) -> str:
     @classmethod
     def deserialize(cls, raw_json: str) -> InvitationContent:
         """Parses a json string and returns InvitationContent model"""
-        model_w_short_aliases = cls.parse_raw(raw_json)
-        return InvitationContent.construct(
-            **model_w_short_aliases.dict(exclude_unset=True)
+        model_w_short_aliases = cls.model_validate_json(raw_json)
+        return InvitationContent.model_construct(
+            **model_w_short_aliases.model_dump(exclude_unset=True)
         )
 
-    class Config:
-        allow_population_by_field_name = True  # NOTE: can parse using field names
-        allow_mutation = False
-        anystr_strip_whitespace = True
+    model_config = ConfigDict(
         # NOTE: Can export with alias: short aliases to minimize the size of serialization artifact
-        fields: ClassVar[dict[str, Any]] = {
-            "issuer": {
-                "alias": "i",
-            },
-            "guest": {
-                "alias": "g",
-            },
-            "trial_account_days": {
-                "alias": "t",
-            },
-            "extra_credits_in_usd": {
-                "alias": "e",
-            },
-            "product": {
-                "alias": "p",
-            },
-            "created": {
-                "alias": "c",
-            },
-        }
+        alias_generator=_to_initial,
+        populate_by_name=True,  # NOTE: can parse using field names
+        frozen=True,
+        str_strip_whitespace=True,
+    )
 
 
 #
@@ -79,9 +65,9 @@ def _build_link(
     r = URL("/registration").include_query_params(invitation=code_url_safe)
 
     # Adds query to fragment
-    base_url = f"{base_url.rstrip('/')}/"
+    base_url = f"{base_url}/"
     url = URL(base_url).replace(fragment=f"{r}")
-    return cast(HttpUrl, parse_obj_as(HttpUrl, f"{url}"))
+    return TypeAdapter(HttpUrlLegacy).validate_python(f"{url}")
 
 
 def _fernet_encrypt_as_urlsafe_code(
@@ -124,7 +110,7 @@ def create_invitation_link_and_content(
     code = _create_invitation_code(content, secret_key)
     # Adds message as the invitation in query
     link = _build_link(
-        base_url=base_url,
+        base_url=f"{base_url}",
         code_url_safe=code.decode(),
     )
     return link, content

services/invitations/tests/unit/api/conftest.py

@@ -18,7 +18,7 @@ def client(app_environment: EnvVarsDict) -> Iterator[TestClient]:
     print(f"app_environment={json.dumps(app_environment)}")
 
     app = create_app()
-    print("settings:\n", app.state.settings.json(indent=1))
+    print("settings:\n", app.state.settings.model_dump_json(indent=1))
     with TestClient(app, base_url="http://testserver.test") as client:
         yield client
 

services/invitations/tests/unit/api/test_api_invitations.py

@@ -64,7 +64,7 @@ def test_check_invitation(
 
     # up ot here, identifcal to above.
     # Let's use invitation link
-    invitation_url = ApiInvitationContentAndLink.parse_obj(
+    invitation_url = ApiInvitationContentAndLink.model_validate(
         response.json()
     ).invitation_url
 
@@ -77,7 +77,7 @@ def test_check_invitation(
     assert response.status_code == 200, f"{response.json()=}"
 
     # decrypted invitation should be identical to request above
-    invitation = InvitationContent.parse_obj(response.json())
+    invitation = InvitationContent.model_validate(response.json())
     assert invitation.issuer == invitation_data.issuer
     assert invitation.guest == invitation_data.guest
     assert invitation.trial_account_days == invitation_data.trial_account_days
@@ -106,7 +106,7 @@ def test_check_valid_invitation(
     assert response.status_code == 200, f"{response.json()=}"
 
     # decrypted invitation should be identical to request above
-    invitation = InvitationContent.parse_obj(response.json())
+    invitation = InvitationContent.model_validate(response.json())
 
     assert invitation.issuer == invitation_data.issuer
     assert invitation.guest == invitation_data.guest

services/invitations/tests/unit/api/test_api_meta.py

@@ -19,7 +19,7 @@ def test_healthcheck(client: TestClient):
 def test_meta(client: TestClient):
     response = client.get(f"/{API_VTAG}/meta")
     assert response.status_code == status.HTTP_200_OK
-    meta = _Meta.parse_obj(response.json())
+    meta = _Meta.model_validate(response.json())
 
     response = client.get(meta.docs_url)
     assert response.status_code == status.HTTP_200_OK

services/invitations/tests/unit/conftest.py

@@ -109,4 +109,4 @@ def invitation_data(
     if product:
         kwargs["product"] = product
 
-    return InvitationInputs.parse_obj(kwargs)
+    return InvitationInputs.model_validate(kwargs)

services/invitations/tests/unit/test__model_examples.py

@@ -26,4 +26,4 @@ def test_model_examples(
     model_cls: type[BaseModel], example_name: int, example_data: Any
 ):
     print(example_name, ":", json.dumps(example_data))
-    assert model_cls.parse_obj(example_data)
+    assert model_cls.model_validate(example_data)

services/invitations/tests/unit/test_cli.py

@@ -8,6 +8,7 @@
 import pytest
 from faker import Faker
 from models_library.products import ProductName
+from pydantic import TypeAdapter
 from pytest_simcore.helpers.monkeypatch_envs import load_dotenv, setenvs_from_dict
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from simcore_service_invitations._meta import API_VERSION
@@ -71,7 +72,10 @@ def test_invite_user_and_check_invitation(
         env=environs,
     )
     assert result.exit_code == os.EX_OK, result.output
-    assert expected == InvitationInputs.parse_raw(result.stdout).dict()
+    assert (
+        expected
+        == TypeAdapter(InvitationInputs).validate_json(result.stdout).model_dump()
+    )
 
 
 def test_echo_dotenv(cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch):
@@ -82,7 +86,7 @@ def test_echo_dotenv(cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch):
     environs = load_dotenv(result.stdout)
 
     envs = setenvs_from_dict(monkeypatch, environs)
-    settings_from_obj = ApplicationSettings.parse_obj(envs)
+    settings_from_obj = ApplicationSettings.model_validate(envs)
     settings_from_envs = ApplicationSettings.create_from_envs()
 
     assert settings_from_envs == settings_from_obj
@@ -93,5 +97,5 @@ def test_list_settings(cli_runner: CliRunner, app_environment: EnvVarsDict):
     assert result.exit_code == os.EX_OK, result.output
 
     print(result.output)
-    settings = ApplicationSettings.parse_raw(result.output)
+    settings = TypeAdapter(ApplicationSettings).validate_json(result.output)
     assert settings == ApplicationSettings.create_from_envs()