From 49aaa1e05174fffbbd764221e6ebd9141a73ee13 Mon Sep 17 00:00:00 2001 From: Thomas Neidhart Date: Wed, 9 Oct 2024 09:54:13 +0200 Subject: [PATCH] chore: cleanup ModelObject class to provide default implementations for common operations --- otterdog/models/__init__.py | 28 +++++++++++------ otterdog/models/branch_protection_rule.py | 16 +--------- otterdog/models/custom_property.py | 29 ++++++----------- otterdog/models/environment.py | 20 +++--------- otterdog/models/organization_secret.py | 17 ++++------ otterdog/models/organization_settings.py | 27 ++++------------ otterdog/models/organization_variable.py | 17 ++++------ .../models/organization_workflow_settings.py | 2 ++ otterdog/models/repository.py | 30 ++++++------------ otterdog/models/ruleset.py | 15 +++------ otterdog/models/secret.py | 25 ++------------- otterdog/models/variable.py | 31 +------------------ otterdog/models/webhook.py | 18 ++--------- otterdog/models/workflow_settings.py | 19 ++---------- 14 files changed, 79 insertions(+), 215 deletions(-) diff --git a/otterdog/models/__init__.py b/otterdog/models/__init__.py index 2511d7a..1bef6dd 100644 --- 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 index 72a4018..6a1d5ce 100644 --- 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 index 5a257c4..979a732 100644 --- 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 index c91cbc2..4a62fa8 100644 --- 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 index 4fb2a95..d292bce 100644 --- 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 index b44b088..e8092fe 100644 --- 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 index 1bb460f..f2ebfca 100644 --- 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 index aa5dedb..66e33e2 100644 --- 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 diff --git a/otterdog/models/repository.py b/otterdog/models/repository.py index 5b5270b..8fd98b9 100644 --- a/otterdog/models/repository.py +++ b/otterdog/models/repository.py @@ -12,7 +12,7 @@ import re from typing import TYPE_CHECKING, Any, ClassVar, 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, @@ -31,7 +31,6 @@ associate_by_key, is_set_and_present, is_set_and_valid, - is_unset, write_patch_object_as_json, ) @@ -569,8 +568,8 @@ def get_model_objects(self) -> Iterator[tuple[ModelObject, ModelObject]]: yield self.workflows, self @classmethod - def from_model_data(cls, data: dict[str, Any]) -> Repository: - mapping = {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( { @@ -591,16 +590,11 @@ def from_model_data(cls, data: dict[str, Any]) -> Repository: } ) - return cls(**bend(mapping, data)) - - @classmethod - def from_provider_data(cls, org_id: str, data: dict[str, Any]) -> Repository: - mapping = cls.get_mapping_from_provider(org_id, data) - return cls(**bend(mapping, data)) + return mapping @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 status_to_bool(status): if status == "enabled": @@ -671,9 +665,7 @@ def property_list_to_map(properties): 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) # add mapping for items that GitHub expects in a nested structure. @@ -865,28 +857,26 @@ def to_jsonnet( # FIXME: support overriding branch protection rules for repos coming from # the default configuration. if has_branch_protection_rules and not extend: - default_org_rule = BranchProtectionRule.from_model_data( - jsonnet_config.default_branch_protection_rule_config - ) + default_bpr = BranchProtectionRule.from_model_data(jsonnet_config.default_branch_protection_rule_config) printer.println("branch_protection_rules: [") printer.level_up() for rule in self.branch_protection_rules: - rule.to_jsonnet(printer, jsonnet_config, context, False, default_org_rule) + rule.to_jsonnet(printer, jsonnet_config, context, False, default_bpr) printer.level_down() printer.println("],") # FIXME: support overriding rulesets for repos coming from the default configuration. if has_rulesets and not extend: - default_org_rule = RepositoryRuleset.from_model_data(jsonnet_config.default_repo_ruleset_config) + default_ruleset = RepositoryRuleset.from_model_data(jsonnet_config.default_repo_ruleset_config) printer.println("rulesets: [") printer.level_up() for ruleset in self.rulesets: - ruleset.to_jsonnet(printer, jsonnet_config, context, False, default_org_rule) + ruleset.to_jsonnet(printer, jsonnet_config, context, False, default_ruleset) printer.level_down() printer.println("],") diff --git a/otterdog/models/ruleset.py b/otterdog/models/ruleset.py index 2e6e965..850e888 100644 --- a/otterdog/models/ruleset.py +++ b/otterdog/models/ruleset.py @@ -13,7 +13,7 @@ import re from typing import TYPE_CHECKING, Any, ClassVar, TypeVar, 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 ( EmbeddedModelObject, @@ -438,8 +438,8 @@ def include_field_for_patch_computation(self, field: dataclasses.Field) -> bool: return True @classmethod - def from_model_data(cls, data: dict[str, Any]): - mapping = {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( { @@ -461,16 +461,11 @@ def from_model_data(cls, data: dict[str, Any]): } ) - return cls(**bend(mapping, data)) - - @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)) + return mapping @classmethod def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[str, Any]: - mapping: dict[str, Any] = {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( { diff --git a/otterdog/models/secret.py b/otterdog/models/secret.py index bf3909c..ab364b5 100644 --- a/otterdog/models/secret.py +++ b/otterdog/models/secret.py @@ -14,7 +14,7 @@ import re from typing import TYPE_CHECKING, Any, TypeVar, cast -from jsonbender import K, OptionalS, S, bend # type: ignore +from jsonbender import K # type: ignore from otterdog.models import ( FailureType, @@ -24,12 +24,11 @@ ModelObject, ValidationContext, ) -from otterdog.utils import UNSET, Change, is_set_and_present, is_unset +from otterdog.utils import Change, is_set_and_present, is_unset if TYPE_CHECKING: from collections.abc import Callable - from otterdog.providers.github import GitHubProvider ST = TypeVar("ST", bound="Secret") @@ -90,31 +89,13 @@ def include_field_for_patch_computation(self, field: dataclasses.Field) -> bool: def include_for_live_patch(self, context: LivePatchContext) -> bool: return not self.has_dummy_secret() - @classmethod - def from_model_data(cls, data: dict[str, Any]): - 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]): - 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: dict[str, Any] = {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) # the provider will never send the value itself, use a dummy secret. mapping["value"] = K("********") return mapping - @classmethod - async def get_mapping_to_provider( - cls, org_id: str, data: dict[str, Any], provider: GitHubProvider - ) -> 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 contains_secrets(self) -> bool: return True diff --git a/otterdog/models/variable.py b/otterdog/models/variable.py index a2e5781..65cf11e 100644 --- a/otterdog/models/variable.py +++ b/otterdog/models/variable.py @@ -10,15 +10,9 @@ import abc import dataclasses -from typing import TYPE_CHECKING, Any, TypeVar - -from jsonbender import OptionalS, S, bend # type: ignore +from typing import Any, TypeVar from otterdog.models import FailureType, ModelObject, ValidationContext -from otterdog.utils import UNSET, is_unset - -if TYPE_CHECKING: - from otterdog.providers.github import GitHubProvider VT = TypeVar("VT", bound="Variable") @@ -38,26 +32,3 @@ def validate(self, context: ValidationContext, parent_object: Any) -> None: FailureType.ERROR, f"{self.get_model_header()} starts with prefix 'GITHUB_' which is not allowed for variables.", ) - - @classmethod - def from_model_data(cls, data: dict[str, Any]): - 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]): - 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())} - return mapping - - @classmethod - async def get_mapping_to_provider( - cls, org_id: str, data: dict[str, Any], provider: GitHubProvider - ) -> dict[str, Any]: - return { - field.name: S(field.name) for field in cls.provider_fields() if not is_unset(data.get(field.name, UNSET)) - } diff --git a/otterdog/models/webhook.py b/otterdog/models/webhook.py index ffe6617..d0a65c4 100644 --- a/otterdog/models/webhook.py +++ b/otterdog/models/webhook.py @@ -13,7 +13,7 @@ import fnmatch from typing import TYPE_CHECKING, Any, TypeVar, cast -from jsonbender import OptionalS, S, bend # type: ignore +from jsonbender import OptionalS, S # type: ignore from otterdog.models import ( FailureType, @@ -107,19 +107,9 @@ def validate(self, context: ValidationContext, parent_object: Any) -> None: f"'insecure_ssl' has value '{self.insecure_ssl}', " f"only values ('0' | '1') are allowed.", ) - @classmethod - def from_model_data(cls, data: dict[str, Any]): - 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]): - 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( { "url": OptionalS("config", "url", default=UNSET), @@ -134,9 +124,7 @@ def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[st 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) config_mapping = {} for config_prop in ["url", "content_type", "insecure_ssl", "secret"]: diff --git a/otterdog/models/workflow_settings.py b/otterdog/models/workflow_settings.py index db352ee..8fafa60 100644 --- a/otterdog/models/workflow_settings.py +++ b/otterdog/models/workflow_settings.py @@ -12,14 +12,13 @@ import dataclasses from typing import TYPE_CHECKING, Any, ClassVar -from jsonbender import OptionalS, S, bend # type: ignore +from jsonbender import OptionalS, S # type: ignore from otterdog.models import FailureType, ModelObject, PatchContext, ValidationContext from otterdog.utils import ( UNSET, IndentingPrinter, is_set_and_valid, - is_unset, write_patch_object_as_json, ) @@ -108,19 +107,9 @@ def include_field_for_patch_computation(self, field: dataclasses.Field) -> bool: return True - @classmethod - def from_model_data(cls, data: dict[str, Any]): - 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]): - 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( { "allow_github_owned_actions": OptionalS("github_owned_allowed", default=None), @@ -137,9 +126,7 @@ def get_mapping_from_provider(cls, org_id: str, data: dict[str, Any]) -> dict[st 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)) - } + mapping = await super().get_mapping_to_provider(org_id, data, provider) if "allow_github_owned_actions" in data: mapping.pop("allow_github_owned_actions")