Delete types replaced by pydantic

Author: xDaile

iib/common/pydantic_models.py

@@ -36,7 +36,20 @@
 ]
 
 
-class AddPydanticModel(BaseModel):
+class PydanticModel(BaseModel):
+
+    @classmethod
+    def _get_all_keys_to_check_in_db(cls):
+        raise NotImplementedError("Not implemented")
+
+    def get_keys_to_check_in_db(self):
+        """Filter keys, which need to be checked in db. Return only a keys that are set to values."""
+        return [
+            k for k in self._get_all_keys_to_check_in_db() if getattr(self, k, None)
+        ]
+
+
+class AddPydanticModel(PydanticModel):
     """Datastructure of the request to /builds/add API point."""
 
     add_arches: Optional[List[str]] = None
@@ -109,24 +122,24 @@ def bundles_needed_with_check_related_images(self) -> 'AddPydanticModel':
 
     def get_json_for_request(self):
         """Return json with the parameters we store in the db."""
-        return {
-            # "add_arches": self.add_arches,  # not in db?
-            "binary_image": self.binary_image,
-            # "build_tags": self.build_tags,  # not in db
-            "bundles": self.bundles,
-            "check_related_images": self.check_related_images,
-            "deprecation_list": self.deprecation_list,
-            "distribution_scope": self.distribution_scope,
-            "from_index": self.from_index,
-            "graph_update_mode": self.graph_update_mode,
-            "organization": self.organization,
-         }
+        return self.model_dump(
+            exclude=[
+                "add_arches",
+                "build_tags",
+                "cnr_token",
+                "force_backport",
+                "overwrite_from_index",
+                "overwrite_from_index_token",
+            ],
+            exclude_defaults=True,
+        )
 
-    def get_keys_to_check_in_db(self):
+
+    def _get_all_keys_to_check_in_db(self):
         return ["binary_image", "bundles", "deprecation_list", "from_index"]
 
 
-class RmPydanticModel(BaseModel):
+class RmPydanticModel(PydanticModel):
     """Datastructure of the request to /builds/rm API point."""
 
     add_arches: Optional[List[str]] = None
@@ -152,17 +165,12 @@ def verify_overwrite_from_index_token(self) -> 'RmPydanticModel':
 
     def get_json_for_request(self):
         """Return json with the parameters we store in the db."""
-        return {
-            # "add_arches": self.add_arches,  # not in db?
-            "binary_image": self.binary_image,
-            # "build_tags": self.build_tags,  # not in db
-            "distribution_scope": self.distribution_scope,
-            "from_index": self.from_index,
-            "operators": self.operators,
-            "organization": self.organization,
-        }
+        return self.model_dump(
+            exclude=["add_arches", "build_tags", "overwrite_from_index", "overwrite_from_index_token"],
+            exclude_defaults=True,
+        )
 
-    def get_keys_to_check_in_db(self):
+    def _get_all_keys_to_check_in_db(self):
         return ["binary_image", "from_index", "operators"]
 
 
@@ -179,7 +187,7 @@ class RegistryAuths(BaseModel):  # is {"auths":{}} allowed?
     auths: Annotated[Dict[SecretStr, RegistryAuth], AfterValidator(length_validator)]
 
 
-class RegenerateBundlePydanticModel(BaseModel):
+class RegenerateBundlePydanticModel(PydanticModel):
     """Datastructure of the request to /builds/regenerate-bundle API point."""
 
     # BUNDLE_IMAGE, from_bundle_image_resolved, build_tags?
@@ -190,13 +198,12 @@ class RegenerateBundlePydanticModel(BaseModel):
 
     def get_json_for_request(self):
         """Return json with the parameters we store in the db."""
-        return {
-            "bundle_replacements": self.bundle_replacements,
-            "from_bundle_image": self.from_bundle_image,
-            "organization": self.organization,
-        }
+        return self.model_dump(
+            exclude=["registry_auths"],
+            exclude_defaults=True,
+        )
 
-    def get_keys_to_check_in_db(self):
+    def _get_all_keys_to_check_in_db(self):
         return ["from_bundle_image"]
 
 
@@ -205,7 +212,7 @@ class RegenerateBundleBatchPydanticModel(BaseModel):
     annotations: Dict[str, Any]
 
 
-class MergeIndexImagePydanticModel(BaseModel):
+class MergeIndexImagePydanticModel(PydanticModel):
     """Datastructure of the request to /builds/regenerate-bundle API point."""
 
     binary_image: Annotated[
@@ -246,23 +253,16 @@ def verify_overwrite_from_index_token(self) -> 'MergeIndexImagePydanticModel':
 
     def get_json_for_request(self):
         """Return json with the parameters we store in the db."""
-        return {
-            "binary_image": self.binary_image,
-            # "build_tags": self.build_tags, # not in db
-            "deprecation_list": self.deprecation_list,
-            "distribution_scope": self.distribution_scope,
-            "graph_update_mode": self.graph_update_mode,
-            "source_from_index": self.source_from_index,
-            "target_index": self.target_index,
-            "batch": self.batch,
-            "user": self.user,
-        }
+        return self.model_dump(
+            exclude=["build_tags", "overwrite_target_index", "overwrite_target_index_token"],
+            exclude_defaults=True,
+        )
 
-    def get_keys_to_check_in_db(self):
+    def _get_all_keys_to_check_in_db(self):
         return ["binary_image", "deprecation_list", "source_from_index", "target_index", "target_index"]
 
 
-class CreateEmptyIndexPydanticModel(BaseModel):
+class CreateEmptyIndexPydanticModel(PydanticModel):
     """Datastructure of the request to /builds/regenerate-bundle API point."""
 
     binary_image: Annotated[
@@ -280,18 +280,15 @@ class CreateEmptyIndexPydanticModel(BaseModel):
 
     def get_json_for_request(self):
         """Return json with the parameters we store in the db."""
-        return {
-            "binary_image": self.binary_image,
-            "from_index": self.from_index,
-            "labels": self.labels,
-            "output_fbc": self.output_fbc,
-        }
+        return self.model_dump(
+            exclude_defaults=True,
+        )
 
-    def get_keys_to_check_in_db(self):
+    def _get_all_keys_to_check_in_db(self):
         return ["binary_image", "from_index"]
 
 
-class RecursiveRelatedBundlesPydanticModel(BaseModel):
+class RecursiveRelatedBundlesPydanticModel(PydanticModel):
     organization: Optional[str] = None
     parent_bundle_image: Annotated[
         str,
@@ -302,17 +299,18 @@ class RecursiveRelatedBundlesPydanticModel(BaseModel):
 
     def get_json_for_request(self):
         """Return json with the parameters we store in the db."""
-        return {
-            "organization": self.organization,
-            "parent_bundle_image": self.parent_bundle_image,
-        }
+        return self.model_dump(
+            exclude=["registry_auths"],
+            exclude_defaults=True,
+        )
 
-    def get_keys_to_check_in_db(self):
+
+    def _get_all_keys_to_check_in_db(self):
         return ["parent_bundle_image"]
 
 
-class FbcOperationsPydanticModel(BaseModel):
-    add_arches: Optional[List[str]]
+class FbcOperationsPydanticModel(PydanticModel):
+    add_arches: Optional[List[str]] = []
     binary_image: Annotated[
         Optional[str],
         AfterValidator(image_format_check),
@@ -349,16 +347,10 @@ def verify_overwrite_from_index_token(self) -> 'FbcOperationsPydanticModel':
 
     def get_json_for_request(self):
         """Return json with the parameters we store in the db."""
-        return {
-            # "add_arches": self.add_arches,  # not in db?
-            "binary_image": self.binary_image,
-            "bundles": self.bundles,
-            # "build_tags": self.build_tags, # not in db
-            "distribution_scope": self.distribution_scope,
-            "fbc_fragment": self.fbc_fragment,
-            "from_index": self.from_index,
-            "organization": self.organization,
-        }
+        return self.model_dump(
+            exclude=["add_arches", "build_tags", "overwrite_from_index", "overwrite_from_index_token"],
+            exclude_defaults=True,
+        )
 
-    def get_keys_to_check_in_db(self):
+    def _get_all_keys_to_check_in_db(self):
         return ["binary_image", "bundles", "fbc_fragment", "from_index"]