add utils

Author: giancarloromeo

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

@@ -0,0 +1,22 @@
+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:
+    return get_origin(info.annotation) 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/common-library/src/common_library/utils/serialization.py

@@ -0,0 +1,32 @@
+from typing import Any
+
+from common_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:
+        if field_name not in data:
+            continue
+
+        field_data = data[field_name]
+
+        if isinstance(field_data, SecretStr):
+            if show_secrets:
+                data[field_name] = field_data.get_secret_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/common-library/tests/test_utils_pydantic_extension.py

@@ -0,0 +1,76 @@
+from typing import Literal
+
+import pytest
+from common_library.utils.pydantic_fields_extension import (
+    get_type,
+    is_literal,
+    is_nullable,
+)
+from pydantic import BaseModel, Field
+
+
+class MyModel(BaseModel):
+    a: int
+    b: float | None = Field(...)
+    c: str = "bla"
+    d: bool | None = None
+    e: Literal["bla"]
+
+
+@pytest.mark.parametrize(
+    "fn,expected,name",
+    [
+        (
+            get_type,
+            int,
+            "a",
+        ),
+        (
+            get_type,
+            float,
+            "b",
+        ),
+        (
+            get_type,
+            str,
+            "c",
+        ),
+        (get_type, bool, "d"),
+        (
+            is_literal,
+            False,
+            "a",
+        ),
+        (
+            is_literal,
+            False,
+            "b",
+        ),
+        (
+            is_literal,
+            False,
+            "c",
+        ),
+        (is_literal, False, "d"),
+        (is_literal, True, "e"),
+        (
+            is_nullable,
+            False,
+            "a",
+        ),
+        (
+            is_nullable,
+            True,
+            "b",
+        ),
+        (
+            is_nullable,
+            False,
+            "c",
+        ),
+        (is_nullable, True, "d"),
+        (is_nullable, False, "e"),
+    ],
+)
+def test_field_fn(fn, expected, name):
+    assert expected == fn(MyModel.model_fields[name])

packages/common-library/tests/test_utils_serialization.py

@@ -0,0 +1,30 @@
+from typing import Final
+
+import pytest
+from models_library.utils.serialization import model_dump_with_secrets
+from pydantic import BaseModel, SecretStr
+
+
+class Credentials(BaseModel):
+    USERNAME: str | None = None
+    PASSWORD: SecretStr | None = None
+
+
+ME: Final[Credentials] = Credentials(USERNAME="DeepThought", PASSWORD=SecretStr("42"))
+
+
+@pytest.mark.parametrize(
+    "expected,show_secrets",
+    [
+        (
+            {"USERNAME": "DeepThought", "PASSWORD": "42"},
+            True,
+        ),
+        (
+            {"USERNAME": "DeepThought", "PASSWORD": "**********"},
+            False,  # hide secrets
+        ),
+    ],
+)
+def test_model_dump_with_secrets(expected: dict, show_secrets: bool):
+    assert expected == model_dump_with_secrets(ME, show_secrets=show_secrets)