fix encoder issues

Author: giancarloromeo

packages/models-library/src/models_library/utils/pydantic_fields_extension.py

@@ -0,0 +1,23 @@
+from types import UnionType
+from typing import Any, Literal, get_args, get_origin
+
+from pydantic.fields import FieldInfo
+
+
+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 != type(None))
+    return field_type
+
+
+def is_literal(info: FieldInfo) -> bool:
+    origin = get_origin(info.annotation)
+    return origin 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

packages/models-library/src/models_library/utils/serialization.py

@@ -0,0 +1,29 @@
+from typing import Any
+
+from models_library.utils.pydantic_fields_extension import get_type
+from pydantic import BaseModel, SecretStr
+
+
+def model_dump_with_secrets(
+    settings_obj: BaseModel, show_secrets: bool, **pydantic_export_options
+) -> dict[str, Any]:
+    data = settings_obj.model_dump(**pydantic_export_options)
+
+    for field_name in settings_obj.model_fields:
+        field_data = data[field_name]
+
+        if isinstance(field_data, SecretStr):
+            if show_secrets:
+                data[field_name] = field_data.get_secret_value()  # Expose the raw value
+            else:
+                data[field_name] = str(field_data)
+        elif isinstance(field_data, dict):
+            field_type = get_type(settings_obj.model_fields[field_name])
+            if issubclass(field_type, BaseModel):
+                data[field_name] = model_dump_with_secrets(
+                    field_type.model_validate(field_data),
+                    show_secrets,
+                    **pydantic_export_options,
+                )
+
+    return data

packages/service-library/tests/deferred_tasks/test_deferred_tasks.py

@@ -16,6 +16,8 @@
 import psutil
 import pytest
 from aiohttp.test_utils import unused_port
+from models_library.utils.json_serialization import json_dumps
+from models_library.utils.serialization import model_dump_with_secrets
 from pydantic import NonNegativeFloat, NonNegativeInt
 from pytest_mock import MockerFixture
 from servicelib import redis as servicelib_redis
@@ -24,7 +26,6 @@
 from servicelib.sequences_utils import partition_gen
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
-from settings_library.utils_encoders import create_json_encoder_wo_secrets
 from tenacity.asyncio import AsyncRetrying
 from tenacity.retry import retry_if_exception_type
 from tenacity.stop import stop_after_delay
@@ -123,7 +124,6 @@ async def _tcp_command(
 
 def _get_serialization_options() -> dict[str, Any]:
     return {
-        "encoder": create_json_encoder_wo_secrets(RabbitSettings),
         "exclude_defaults": True,
         "exclude_none": True,
         "exclude_unset": True,
@@ -158,8 +158,20 @@ async def start(self) -> None:
         response = await _tcp_command(
             "init-context",
             {
-                "rabbit": self.rabbit_service.model_dump_json(**_get_serialization_options()),
-                "redis": self.redis_service.model_dump_json(**_get_serialization_options()),
+                "rabbit": json_dumps(
+                    model_dump_with_secrets(
+                        self.rabbit_service,
+                        show_secrets=True,
+                        **_get_serialization_options(),
+                    )
+                ),
+                "redis": json_dumps(
+                    model_dump_with_secrets(
+                        self.redis_service,
+                        show_secrets=True,
+                        **_get_serialization_options(),
+                    )
+                ),
                 "max-workers": self.max_workers,
             },
             port=self.remote_process.port,

packages/settings-library/src/settings_library/base.py

@@ -1,8 +1,12 @@
 import logging
 from functools import cached_property
-from types import UnionType
-from typing import Any, Final, Literal, get_args, get_origin
+from typing import Any, Final, get_origin
 
+from models_library.utils.pydantic_fields_extension import (
+    get_type,
+    is_literal,
+    is_nullable,
+)
 from pydantic import ValidationInfo, field_validator
 from pydantic.fields import FieldInfo
 from pydantic_core import PydanticUndefined, ValidationError
@@ -21,38 +25,19 @@ def __init__(self, errors):
         self.errors = errors
 
 
-def _allows_none(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
-
-
-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 != type(None))
-    return field_type
-
-
-def _is_literal(info: FieldInfo) -> bool:
-    origin = get_origin(info.annotation)
-    return origin is Literal
-
-
 def _create_settings_from_env(field_name: str, info: FieldInfo):
     # NOTE: Cannot pass only field.type_ because @prepare_field (when this function is called)
     #  this value is still not resolved (field.type_ at that moment has a weak_ref).
     #  Therefore we keep the entire 'field' but MUST be treated here as read-only
 
     def _default_factory():
         """Creates default from sub-settings or None (if nullable)"""
-        field_settings_cls = _get_type(info)
+        field_settings_cls = get_type(info)
         try:
             return field_settings_cls()
 
         except ValidationError as err:
-            if _allows_none(info):
+            if is_nullable(info):
                 # e.g. Optional[PostgresSettings] would warn if defaults to None
                 _logger.warning(
                     _DEFAULTS_TO_NONE_MSG,
@@ -80,7 +65,7 @@ def _parse_none(cls, v, info: ValidationInfo):
         # WARNING: In nullable fields, envs equal to null or none are parsed as None !!
         if (
             info.field_name
-            and _allows_none(cls.model_fields[info.field_name])
+            and is_nullable(cls.model_fields[info.field_name])
             and isinstance(v, str)
             and v.lower() in ("none",)
         ):
@@ -107,13 +92,13 @@ def __pydantic_init_subclass__(cls, **kwargs: Any):
                     "auto_default_from_env", False
                 )
             )
-            field_type = _get_type(field)
+            field_type = get_type(field)
 
             # Avoids issubclass raising TypeError. SEE test_issubclass_type_error_with_pydantic_models
             is_not_composed = (
                 get_origin(field_type) is None
             )  # is not composed as dict[str, Any] or Generic[Base]
-            is_not_literal = not _is_literal(field)
+            is_not_literal = not is_literal(field)
 
             if (
                 is_not_literal

packages/settings-library/src/settings_library/utils_cli.py

@@ -7,36 +7,12 @@
 
 import rich
 import typer
-from pydantic import SecretStr, ValidationError
+from models_library.utils.serialization import model_dump_with_secrets
+from pydantic import ValidationError
 from pydantic_settings import BaseSettings
 
 from ._constants import HEADER_STR
-from .base import BaseCustomSettings, _get_type
-
-
-def model_dump_with_secrets(
-    settings_obj: BaseSettings, show_secret: bool, **pydantic_export_options
-) -> dict[str, Any]:
-    data = settings_obj.model_dump(**pydantic_export_options)
-
-    for field_name in settings_obj.model_fields:
-        field_data = data[field_name]
-
-        if isinstance(field_data, SecretStr):
-            if show_secret:
-                data[field_name] = field_data.get_secret_value()  # Expose the raw value
-            else:
-                data[field_name] = str(field_data)
-        elif isinstance(field_data, dict):
-            field_type = _get_type(settings_obj.model_fields[field_name])
-            if issubclass(field_type, BaseSettings):
-                data[field_name] = model_dump_with_secrets(
-                    field_type.model_validate(field_data),
-                    show_secret,
-                    **pydantic_export_options,
-                )
-
-    return data
+from .base import BaseCustomSettings
 
 
 def print_as_envfile(