Create generic SerializeAsOptional type for Pydantic

kpops/components/common/kubernetes_model.py

@@ -7,7 +7,11 @@
 from pydantic import Field, model_validator
 
 from kpops.utils.docstring import describe_attr
-from kpops.utils.pydantic import CamelCaseConfigModel, DescConfigModel
+from kpops.utils.pydantic import (
+    CamelCaseConfigModel,
+    DescConfigModel,
+    SerializeAsOptional,
+)
 
 if TYPE_CHECKING:
     try:
@@ -98,11 +102,11 @@ class NodeSelectorTerm(DescConfigModel, CamelCaseConfigModel):
     :param match_fields: A list of node selector requirements by node's fields.
     """
 
-    match_expressions: list[NodeSelectorRequirement] | None = Field(
-        default=None, description=describe_attr("match_expressions", __doc__)
+    match_expressions: SerializeAsOptional[list[NodeSelectorRequirement]] = Field(
+        default=[], description=describe_attr("match_expressions", __doc__)
     )
-    match_fields: list[NodeSelectorRequirement] | None = Field(
-        default=None, description=describe_attr("match_fields", __doc__)
+    match_fields: SerializeAsOptional[list[NodeSelectorRequirement]] = Field(
+        default=[], description=describe_attr("match_fields", __doc__)
     )
 
 
@@ -143,10 +147,10 @@ class NodeAffinity(DescConfigModel, CamelCaseConfigModel):
             "required_during_scheduling_ignored_during_execution", __doc__
         ),
     )
-    preferred_during_scheduling_ignored_during_execution: (
-        list[PreferredSchedulingTerm] | None
-    ) = Field(
-        default=None,
+    preferred_during_scheduling_ignored_during_execution: SerializeAsOptional[
+        list[PreferredSchedulingTerm]
+    ] = Field(
+        default=[],
         description=describe_attr(
             "preferred_during_scheduling_ignored_during_execution", __doc__
         ),
@@ -197,12 +201,12 @@ class LabelSelector(DescConfigModel, CamelCaseConfigModel):
     :param match_expressions: matchExpressions is a list of label selector requirements. The requirements are ANDed.
     """
 
-    match_labels: dict[str, str] | None = Field(
-        default=None,
+    match_labels: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("match_labels", __doc__),
     )
-    match_expressions: list[LabelSelectorRequirement] | None = Field(
-        default=None,
+    match_expressions: SerializeAsOptional[list[LabelSelectorRequirement]] = Field(
+        default=[],
         description=describe_attr("match_expressions", __doc__),
     )
 
@@ -222,19 +226,19 @@ class PodAffinityTerm(DescConfigModel, CamelCaseConfigModel):
         default=None,
         description=describe_attr("label_selector", __doc__),
     )
-    match_label_keys: list[str] | None = Field(
-        default=None,
+    match_label_keys: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("match_label_keys", __doc__),
     )
-    mismatch_label_keys: list[str] | None = Field(
-        default=None,
+    mismatch_label_keys: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("mismatch_label_keys", __doc__),
     )
     topology_key: str = Field(
         description=describe_attr("topology_key", __doc__),
     )
-    namespaces: list[str] | None = Field(
-        default=None,
+    namespaces: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("namespaces", __doc__),
     )
     namespace_selector: LabelSelector | None = Field(
@@ -265,18 +269,18 @@ class PodAffinity(DescConfigModel, CamelCaseConfigModel):
     :param preferred_during_scheduling_ignored_during_execution: The scheduler will prefer to schedule pods to nodes that satisfy the affinity expressions specified by this field, but it may choose a node that violates one or more of the expressions. The node that is most preferred is the one with the greatest sum of weights, i.e. for each node that meets all of the scheduling requirements (resource request, requiredDuringScheduling affinity expressions, etc.), compute a sum by iterating through the elements of this field and adding weight to the sum if the node has pods which matches the corresponding podAffinityTerm; the node(s) with the highest sum are the most preferred.
     """
 
-    required_during_scheduling_ignored_during_execution: (
-        list[PodAffinityTerm] | None
-    ) = Field(
-        default=None,
+    required_during_scheduling_ignored_during_execution: SerializeAsOptional[
+        list[PodAffinityTerm]
+    ] = Field(
+        default=[],
         description=describe_attr(
             "required_during_scheduling_ignored_during_execution", __doc__
         ),
     )
-    preferred_during_scheduling_ignored_during_execution: (
-        list[WeightedPodAffinityTerm] | None
-    ) = Field(
-        default=None,
+    preferred_during_scheduling_ignored_during_execution: SerializeAsOptional[
+        list[WeightedPodAffinityTerm]
+    ] = Field(
+        default=[],
         description=describe_attr(
             "preferred_during_scheduling_ignored_during_execution", __doc__
         ),

kpops/components/streams_bootstrap/model.py

@@ -11,6 +11,7 @@
     ImagePullPolicy,
     ProtocolSchema,
     Resources,
+    SerializeAsOptional,
     ServiceType,
     Toleration,
 )
@@ -132,8 +133,8 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("image_pull_policy", __doc__),
     )
 
-    image_pull_secrets: list[dict[str, str]] | None = Field(
-        default=None,
+    image_pull_secrets: SerializeAsOptional[list[dict[str, str]]] = Field(
+        default=[],
         description=describe_attr("image_pull_secret", __doc__),
     )
 
@@ -146,8 +147,8 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("resources", __doc__),
     )
 
-    ports: list[PortConfig] | None = Field(
-        default=None,
+    ports: SerializeAsOptional[list[PortConfig]] = Field(
+        default=[],
         description=describe_attr("ports", __doc__),
     )
 
@@ -161,33 +162,33 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("configuration_env_prefix", __doc__),
     )
 
-    command_line: dict[str, str | bool | int] | None = Field(
-        default=None,
+    command_line: SerializeAsOptional[dict[str, str | bool | int]] = Field(
+        default={},
         description=describe_attr("command_line", __doc__),
     )
 
-    env: dict[str, str] | None = Field(
-        default=None,
+    env: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("env", __doc__),
     )
 
-    secrets: dict[str, str] | None = Field(
-        default=None,
+    secrets: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("secrets", __doc__),
     )
 
-    secret_refs: dict[str, Any] | None = Field(
-        default=None,
+    secret_refs: SerializeAsOptional[dict[str, Any]] = Field(
+        default={},
         description=describe_attr("secret_refs", __doc__),
     )
 
-    secret_files_refs: list[str] | None = Field(
-        default=None,
+    secret_files_refs: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("secret_files_refs", __doc__),
     )
 
-    files: dict[str, Any] | None = Field(
-        default=None,
+    files: SerializeAsOptional[dict[str, Any]] = Field(
+        default={},
         description=describe_attr("files", __doc__),
     )
 
@@ -196,23 +197,23 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("java_options", __doc__),
     )
 
-    pod_annotations: dict[str, str] | None = Field(
-        default=None,
+    pod_annotations: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("pod_annotations", __doc__),
     )
 
-    pod_labels: dict[str, str] | None = Field(
-        default=None,
+    pod_labels: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("pod_labels", __doc__),
     )
 
-    liveness_probe: dict[str, Any] | None = Field(
-        default=None,
+    liveness_probe: SerializeAsOptional[dict[str, Any]] = Field(
+        default={},
         description=describe_attr("liveness_probe", __doc__),
     )
 
-    readiness_probe: dict[str, Any] | None = Field(
-        default=None,
+    readiness_probe: SerializeAsOptional[dict[str, Any]] = Field(
+        default={},
         description=describe_attr("readiness_probe", __doc__),
     )
 
@@ -221,8 +222,8 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("affinity", __doc__),
     )
 
-    tolerations: list[Toleration] | None = Field(
-        default=None,
+    tolerations: SerializeAsOptional[list[Toleration]] = Field(
+        default=[],
         description=describe_attr("tolerations", __doc__),
     )
 

kpops/components/streams_bootstrap/producer/producer_app.py

@@ -63,9 +63,12 @@ def is_cron_job(self) -> bool:
     @computed_field
     @cached_property
     def _cleaner(self) -> ProducerAppCleaner:
-        return ProducerAppCleaner(
-            **self.model_dump(by_alias=True, exclude={"_cleaner", "from_", "to"})
-        )
+        kwargs = {
+            name: getattr(self, name)
+            for name in self.model_fields_set
+            if name not in {"_cleaner", "from_", "to"}
+        }
+        return ProducerAppCleaner.model_validate(kwargs)
 
     @override
     def apply_to_outputs(self, name: str, topic: TopicConfig) -> None:

kpops/components/streams_bootstrap/streams/model.py

@@ -8,6 +8,7 @@
 from kpops.components.common.kubernetes_model import (
     ImagePullPolicy,
     Resources,
+    SerializeAsOptional,
 )
 from kpops.components.common.topic import KafkaTopic, KafkaTopicStr
 from kpops.components.streams_bootstrap.model import (
@@ -191,16 +192,16 @@ class StreamsAppAutoScaling(CamelCaseConfigModel, DescConfigModel):
         title="Idle replica count",
         description=describe_attr("idle_replicas", __doc__),
     )
-    internal_topics: list[str] | None = Field(
-        default=None,
+    internal_topics: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("internal_topics", __doc__),
     )
-    topics: list[str] | None = Field(
-        default=None,
+    topics: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("topics", __doc__),
     )
-    additional_triggers: list[str] | None = Field(
-        default=None,
+    additional_triggers: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("additional_triggers", __doc__),
     )
     model_config = ConfigDict(extra="allow")
@@ -289,8 +290,8 @@ class JMXConfig(CamelCaseConfigModel, DescConfigModel):
         description=describe_attr("port", __doc__),
     )
 
-    metric_rules: list[str] | None = Field(
-        default=None,
+    metric_rules: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("metric_rules", __doc__),
     )
 

kpops/components/streams_bootstrap/streams/streams_app.py

@@ -67,9 +67,12 @@ class StreamsApp(StreamsBootstrap):
     @computed_field
     @cached_property
     def _cleaner(self) -> StreamsAppCleaner:
-        return StreamsAppCleaner(
-            **self.model_dump(by_alias=True, exclude={"_cleaner", "from_", "to"})
-        )
+        kwargs = {
+            name: getattr(self, name)
+            for name in self.model_fields_set
+            if name not in {"_cleaner", "from_", "to"}
+        }
+        return StreamsAppCleaner.model_validate(kwargs)
 
     @property
     @override

kpops/utils/pydantic.py

@@ -1,11 +1,20 @@
 import json
 import logging
 from pathlib import Path
-from typing import Any
+from typing import Annotated, Any
 
 import humps
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    GetCoreSchemaHandler,
+    SerializationInfo,
+    SerializerFunctionWrapHandler,
+    WrapSerializer,
+)
 from pydantic.fields import FieldInfo
+from pydantic_core import core_schema
 from pydantic_settings import PydanticBaseSettingsSource
 from typing_extensions import TypeVar, override
 
@@ -224,3 +233,34 @@ def __call__(self) -> dict[str, Any]:
             if field_value is not None:
                 d[field_key] = field_value
         return d
+
+
+_T = TypeVar("_T")
+
+
+def serialize_to_optional(
+    value: _T,
+    default_serialize_handler: SerializerFunctionWrapHandler,
+    info: SerializationInfo,
+) -> _T | None:
+    result = default_serialize_handler(value)
+    return result or None
+
+
+class WrapNullableSchema:
+    def __get_pydantic_core_schema__(
+        self,
+        source: type[Any],
+        handler: GetCoreSchemaHandler,
+    ) -> core_schema.CoreSchema:
+        schema = handler(source)
+        # wrap generated schema in nullable
+        return core_schema.NullableSchema(type="nullable", schema=schema)
+
+
+SerializeAsOptional = Annotated[
+    _T,
+    WrapSerializer(serialize_to_optional),
+    WrapNullableSchema(),
+    "Optional that is serialized to None if falsy",
+]