chore: cleanup ModelObject class to provide default implementations f…

…or common operations
eclipse-csi · Oct 9, 2024 · 49aaa1e · 49aaa1e
1 parent df6b215
commit 49aaa1e
Show file tree

Hide file tree

Showing 14 changed files with 79 additions and 215 deletions.
diff --git a/otterdog/models/__init__.py b/otterdog/models/__init__.py
@@ -12,7 +12,7 @@
 import os
 from abc import ABC, abstractmethod
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Protocol, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Protocol, TypeVar, cast, final
 
 from jsonbender import OptionalS, S, bend  # type: ignore
 
@@ -475,16 +475,24 @@ def get_model_header(self, parent_object: ModelObject | None = None) -> str:
         return header
 
     @classmethod
-    @abstractmethod
-    def from_model_data(cls, data: dict[str, Any]): ...
+    @final
+    def from_model_data(cls: type[MT], data: dict[str, Any]) -> MT:
+        mapping = cls.get_mapping_from_model()
+        return cls(**bend(mapping, data))  # type: ignore
 
     @classmethod
-    @abstractmethod
-    def from_provider_data(cls, org_id: str, data: dict[str, Any]): ...
+    def get_mapping_from_model(cls) -> dict[str, Any]:
+        return {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
 
     @classmethod
-    @abstractmethod
-    def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]: ...
+    @final
+    def from_provider_data(cls: type[MT], org_id: str, data: dict[str, Any]) -> MT:
+        mapping = cls.get_mapping_from_provider(org_id, data)
+        return cls(**bend(mapping, data))  # type: ignore
+
+    @classmethod
+    def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
+        return {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
 
     async def to_provider_data(self, org_id: str, provider: GitHubProvider) -> dict[str, Any]:
         return await self.dict_to_provider_data(org_id, self.to_model_dict(), provider)
@@ -501,10 +509,12 @@ async def dict_to_provider_data(cls, org_id: str, data: dict[str, Any], provider
         return bend(mapping, data)
 
     @classmethod
-    @abstractmethod
     async def get_mapping_to_provider(
         cls, org_id: str, data: dict[str, Any], provider: GitHubProvider
-    ) -> dict[str, Any]: ...
+    ) -> dict[str, Any]:
+        return {
+            field.name: S(field.name) for field in cls.provider_fields() if not is_unset(data.get(field.name, UNSET))
+        }
 
     def include_field_for_diff_computation(self, field: dataclasses.Field) -> bool:
         return True

diff --git a/otterdog/models/branch_protection_rule.py b/otterdog/models/branch_protection_rule.py
@@ -12,7 +12,7 @@
 import re
 from typing import TYPE_CHECKING, Any, cast
 
-from jsonbender import Forall, K, OptionalS, S, bend  # type: ignore
+from jsonbender import Forall, K, OptionalS, S  # type: ignore
 
 from otterdog.models import (
     FailureType,
@@ -274,20 +274,6 @@ def include_field_for_diff_computation(self, field: dataclasses.Field) -> bool:
     def include_field_for_patch_computation(self, field: dataclasses.Field) -> bool:
         return True
 
-    @classmethod
-    def from_model_data(cls, data: dict[str, Any]) -> BranchProtectionRule:
-        mapping: dict[str, Any] = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
-
-        if "requires_approving_reviews" in data:
-            mapping["requires_pull_request"] = S("requires_approving_reviews")
-
-        return cls(**bend(mapping, data))
-
-    @classmethod
-    def from_provider_data(cls, org_id: str, data: dict[str, Any]) -> BranchProtectionRule:
-        mapping = cls.get_mapping_from_provider(org_id, data)
-        return cls(**bend(mapping, data))
-
     @classmethod
     def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
         mapping = {k: OptionalS(snake_to_camel_case(k), default=UNSET) for k in (x.name for x in cls.all_fields())}

diff --git a/otterdog/models/custom_property.py b/otterdog/models/custom_property.py
@@ -11,7 +11,7 @@
 import dataclasses
 from typing import TYPE_CHECKING, Any
 
-from jsonbender import OptionalS, S, bend  # type: ignore
+from jsonbender import OptionalS, S  # type: ignore
 
 from otterdog.models import (
     FailureType,
@@ -22,7 +22,7 @@
     ModelObject,
     ValidationContext,
 )
-from otterdog.utils import UNSET, Change, is_set_and_present, is_set_and_valid, is_unset
+from otterdog.utils import Change, is_set_and_present, is_set_and_valid
 
 if TYPE_CHECKING:
     from otterdog.jsonnet import JsonnetConfig
@@ -141,31 +141,22 @@ def include_field_for_diff_computation(self, field: dataclasses.Field) -> bool:
     def include_field_for_patch_computation(self, field: dataclasses.Field) -> bool:
         return True
 
-    @classmethod
-    def from_model_data(cls, data: dict[str, Any]) -> CustomProperty:
-        mapping = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
-        return cls(**bend(mapping, data))
-
-    @classmethod
-    def from_provider_data(cls, org_id: str, data: dict[str, Any]) -> CustomProperty:
-        mapping = cls.get_mapping_from_provider(org_id, data)
-        mapping.update({"allowed_values": OptionalS("allowed_values", default=[])})
-        return cls(**bend(mapping, data))
-
     @classmethod
     def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
-        mapping = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
-
-        mapping.update({"name": S("property_name")})
+        mapping = super().get_mapping_from_provider(org_id, data)
+        mapping.update(
+            {
+                "name": S("property_name"),
+                "allowed_values": OptionalS("allowed_values", default=[]),
+            }
+        )
         return mapping
 
     @classmethod
     async def get_mapping_to_provider(
         cls, org_id: str, data: dict[str, Any], provider: GitHubProvider
     ) -> dict[str, Any]:
-        mapping: dict[str, Any] = {
-            field.name: S(field.name) for field in cls.provider_fields() if not is_unset(data.get(field.name, UNSET))
-        }
+        mapping = await super().get_mapping_to_provider(org_id, data, provider)
 
         if "name" in data:
             mapping.pop("name")

diff --git a/otterdog/models/environment.py b/otterdog/models/environment.py
@@ -11,7 +11,7 @@
 import dataclasses
 from typing import TYPE_CHECKING, Any
 
-from jsonbender import F, Filter, Forall, If, K, OptionalS, S, bend  # type: ignore
+from jsonbender import F, Filter, Forall, If, K, OptionalS, S  # type: ignore
 
 from otterdog.models import (
     FailureType,
@@ -20,7 +20,7 @@
     ModelObject,
     ValidationContext,
 )
-from otterdog.utils import UNSET, is_set_and_valid, is_unset
+from otterdog.utils import is_set_and_valid, is_unset
 
 if TYPE_CHECKING:
     from otterdog.jsonnet import JsonnetConfig
@@ -98,19 +98,9 @@ def include_existing_object_for_live_patch(self, org_id: str, parent_object: Mod
         else:
             return True
 
-    @classmethod
-    def from_model_data(cls, data: dict[str, Any]) -> Environment:
-        mapping = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
-        return cls(**bend(mapping, data))
-
-    @classmethod
-    def from_provider_data(cls, org_id: str, data: dict[str, Any]) -> Environment:
-        mapping = cls.get_mapping_from_provider(org_id, data)
-        return cls(**bend(mapping, data))
-
     @classmethod
     def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
-        mapping = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
+        mapping = super().get_mapping_from_provider(org_id, data)
 
         def transform_reviewers(x):
             match x["type"]:
@@ -165,9 +155,7 @@ def transform_branch_policy(x):
     async def get_mapping_to_provider(
         cls, org_id: str, data: dict[str, Any], provider: GitHubProvider
     ) -> dict[str, Any]:
-        mapping: dict[str, Any] = {
-            field.name: S(field.name) for field in cls.provider_fields() if not is_unset(data.get(field.name, UNSET))
-        }
+        mapping = await super().get_mapping_to_provider(org_id, data, provider)
 
         if "reviewers" in mapping:
             reviewers = data["reviewers"]

diff --git a/otterdog/models/organization_secret.py b/otterdog/models/organization_secret.py
@@ -11,11 +11,11 @@
 import dataclasses
 from typing import TYPE_CHECKING, Any, cast
 
-from jsonbender import Forall, If, K, OptionalS, S, bend  # type: ignore
+from jsonbender import Forall, If, K, OptionalS, S  # type: ignore
 
 from otterdog.models import FailureType, LivePatch, LivePatchType, ValidationContext
 from otterdog.models.secret import Secret
-from otterdog.utils import UNSET, is_set_and_valid, is_unset
+from otterdog.utils import UNSET, is_set_and_valid
 
 if TYPE_CHECKING:
     from otterdog.jsonnet import JsonnetConfig
@@ -65,14 +65,10 @@ def validate(self, context: ValidationContext, parent_object: Any) -> None:
                     f"setting will be ignored.",
                 )
 
-    @classmethod
-    def from_provider_data(cls, org_id: str, data: dict[str, Any]):
-        mapping = cls.get_mapping_from_provider(org_id, data)
-        return cls(**bend(mapping, data))
-
     @classmethod
     def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
-        mapping = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
+        mapping = super().get_mapping_from_provider(org_id, data)
+
         mapping.update(
             {
                 "visibility": If(
@@ -85,15 +81,14 @@ def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[st
                 "value": K("********"),
             }
         )
+
         return mapping
 
     @classmethod
     async def get_mapping_to_provider(
         cls, org_id: str, data: dict[str, Any], provider: GitHubProvider
     ) -> dict[str, Any]:
-        mapping: dict[str, Any] = {
-            field.name: S(field.name) for field in cls.provider_fields() if not is_unset(data.get(field.name, UNSET))
-        }
+        mapping = await super().get_mapping_to_provider(org_id, data, provider)
 
         if "visibility" in mapping:
             mapping["visibility"] = If(S("visibility") == K("public"), K("all"), S("visibility"))

diff --git a/otterdog/models/organization_settings.py b/otterdog/models/organization_settings.py
@@ -11,7 +11,7 @@
 import dataclasses
 from typing import TYPE_CHECKING, Any, cast
 
-from jsonbender import F, Forall, If, K, OptionalS, S, bend  # type: ignore
+from jsonbender import F, Forall, If, K, OptionalS, S  # type: ignore
 
 from otterdog.models import (
     FailureType,
@@ -29,7 +29,6 @@
     IndentingPrinter,
     is_set_and_present,
     is_set_and_valid,
-    is_unset,
     write_patch_object_as_json,
 )
 
@@ -145,8 +144,8 @@ def get_model_objects(self) -> Iterator[tuple[ModelObject, ModelObject]]:
             yield from self.workflows.get_model_objects()
 
     @classmethod
-    def from_model_data(cls, data: dict[str, Any]) -> OrganizationSettings:
-        mapping: dict[str, Any] = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
+    def get_mapping_from_model(cls) -> dict[str, Any]:
+        mapping = super().get_mapping_from_model()
 
         mapping.update(
             {
@@ -160,26 +159,12 @@ def from_model_data(cls, data: dict[str, Any]) -> OrganizationSettings:
             }
         )
 
-        return cls(**bend(mapping, data))
-
-    @classmethod
-    def from_provider_data(cls, org_id: str, data: dict[str, Any]) -> OrganizationSettings:
-        mapping = cls.get_mapping_from_provider(org_id, data)
-        return cls(**bend(mapping, data))
-
-    @classmethod
-    def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
-        mapping = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
-        mapping.update({"plan": OptionalS("plan", "name", default=UNSET)})
         return mapping
 
     @classmethod
-    async def get_mapping_to_provider(
-        cls, org_id: str, data: dict[str, Any], provider: GitHubProvider
-    ) -> dict[str, Any]:
-        mapping = {
-            field.name: S(field.name) for field in cls.provider_fields() if not is_unset(data.get(field.name, UNSET))
-        }
+    def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
+        mapping = super().get_mapping_from_provider(org_id, data)
+        mapping["plan"] = OptionalS("plan", "name", default=UNSET)
         return mapping
 
     def get_jsonnet_template_function(self, jsonnet_config: JsonnetConfig, extend: bool) -> str | None:

diff --git a/otterdog/models/organization_variable.py b/otterdog/models/organization_variable.py
@@ -11,11 +11,11 @@
 import dataclasses
 from typing import TYPE_CHECKING, Any, cast
 
-from jsonbender import Forall, If, K, OptionalS, S, bend  # type: ignore
+from jsonbender import Forall, If, K, OptionalS, S  # type: ignore
 
 from otterdog.models import FailureType, LivePatch, LivePatchType, ValidationContext
 from otterdog.models.variable import Variable
-from otterdog.utils import UNSET, is_set_and_valid, is_unset
+from otterdog.utils import UNSET, is_set_and_valid
 
 if TYPE_CHECKING:
     from otterdog.jsonnet import JsonnetConfig
@@ -65,14 +65,10 @@ def validate(self, context: ValidationContext, parent_object: Any) -> None:
                     f"setting will be ignored.",
                 )
 
-    @classmethod
-    def from_provider_data(cls, org_id: str, data: dict[str, Any]):
-        mapping = cls.get_mapping_from_provider(org_id, data)
-        return cls(**bend(mapping, data))
-
     @classmethod
     def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
-        mapping = {k: OptionalS(k, default=UNSET) for k in (x.name for x in cls.all_fields())}
+        mapping = super().get_mapping_from_provider(org_id, data)
+
         mapping.update(
             {
                 "visibility": If(
@@ -83,15 +79,14 @@ def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[st
                 "selected_repositories": OptionalS("selected_repositories", default=[]) >> Forall(lambda x: x["name"]),
             }
         )
+
         return mapping
 
     @classmethod
     async def get_mapping_to_provider(
         cls, org_id: str, data: dict[str, Any], provider: GitHubProvider
     ) -> dict[str, Any]:
-        mapping: dict[str, Any] = {
-            field.name: S(field.name) for field in cls.provider_fields() if not is_unset(data.get(field.name, UNSET))
-        }
+        mapping = await super().get_mapping_to_provider(org_id, data, provider)
 
         if "visibility" in mapping:
             mapping["visibility"] = If(S("visibility") == K("public"), K("all"), S("visibility"))

diff --git a/otterdog/models/organization_workflow_settings.py b/otterdog/models/organization_workflow_settings.py
@@ -81,6 +81,7 @@ def validate(self, context: ValidationContext, parent_object: Any) -> None:
     @classmethod
     def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]:
         mapping = super().get_mapping_from_provider(org_id, data)
+
         mapping.update(
             {
                 "selected_repositories": If(
@@ -90,6 +91,7 @@ def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[st
                 ),
             }
         )
+
         return mapping
 
     @classmethod