Certificate implementation

Signed-off-by: Anton Kremenetsky <anton.kremenetsky@gmail.com>

Author: akremenetsky

genesis_core/secret/constants.py

@@ -18,6 +18,7 @@
 
 DEFAULT_SQL_LIMIT = 100
 PASSWORD_KIND = "password"
+CERTIFICATE_KIND = "certificate"
 
 
 class SecretStatus(str, enum.Enum):

genesis_core/secret/dm/models.py

@@ -27,43 +27,48 @@
 from genesis_core.secret import constants as sc
 
 
-class AbstractPasswordConstructor(
+class AbstractSecretConstructor(
     types_dynamic.AbstractKindModel, ra_models.SimpleViewMixin
 ):
 
-    def build(self, plain_password: str) -> str:
+    def build(self, plain_secret: str) -> str:
         raise NotImplementedError()
 
 
-class PlainPasswordConstructor(AbstractPasswordConstructor):
+class PlainSecretConstructor(AbstractSecretConstructor):
     KIND = "plain"
 
-    def build(self, plain_password: str) -> str:
-        return plain_password
+    def build(self, plain_secret: str) -> str:
+        return plain_secret
 
 
-class Password(
+class Secret(
     cm.ModelWithFullAsset,
-    orm.SQLStorableMixin,
     ua_models.TargetResourceMixin,
-    ua_models.TargetResourceSQLStorableMixin,
 ):
-    __tablename__ = "secret_passwords"
-
-    method = properties.property(
-        types.Enum([s.value for s in sc.SecretMethod]),
-        default=sc.SecretMethod.AUTO_HEX.value,
-    )
     status = properties.property(
         types.Enum([s.value for s in sc.SecretStatus]),
         default=sc.SecretStatus.NEW.value,
     )
     constructor = properties.property(
         types_dynamic.KindModelSelectorType(
-            types_dynamic.KindModelType(PlainPasswordConstructor),
+            types_dynamic.KindModelType(PlainSecretConstructor),
         ),
         required=True,
-        default=PlainPasswordConstructor,
+        default=PlainSecretConstructor,
+    )
+
+
+class Password(
+    Secret,
+    orm.SQLStorableMixin,
+    ua_models.TargetResourceSQLStorableMixin,
+):
+    __tablename__ = "secret_passwords"
+
+    method = properties.property(
+        types.Enum([s.value for s in sc.SecretMethod]),
+        default=sc.SecretMethod.AUTO_HEX.value,
     )
     value = properties.property(
         types.AllowNone(types.String(min_length=1, max_length=512)),
@@ -105,3 +110,87 @@ def get_deleted_passwords(
         return cls.get_deleted_target_resources(
             cls.__tablename__, sc.PASSWORD_KIND, limit
         )
+
+
+class AbstractCertificateMethod(
+    types_dynamic.AbstractKindModel, ra_models.SimpleViewMixin
+):
+    pass
+
+
+class DNSCoreCertificateMethod(AbstractCertificateMethod):
+    KIND = "dns_core"
+
+
+class Certificate(
+    Secret,
+    orm.SQLStorableWithJSONFieldsMixin,
+    ua_models.TargetResourceSQLStorableMixin,
+):
+    __tablename__ = "secret_certificates"
+    __jsonfields__ = ["domains"]
+
+    method = properties.property(
+        types_dynamic.KindModelSelectorType(
+            types_dynamic.KindModelType(DNSCoreCertificateMethod),
+        ),
+        required=True,
+        default=DNSCoreCertificateMethod,
+    )
+    valid_until = properties.property(
+        types.AllowNone(types.UTCDateTimeZ()),
+        default=None,
+    )
+    email = properties.property(types.Email())
+    domains = properties.property(
+        types.TypedList(types.String()),
+        required=True,
+    )
+    key = properties.property(
+        types.AllowNone(types.String(min_length=1, max_length=10240)),
+        default=None,
+    )
+    cert = properties.property(
+        types.AllowNone(types.String(min_length=1, max_length=10240)),
+        default=None,
+    )
+
+    def get_resource_target_fields(self) -> set[str]:
+        """Return the collection of target fields.
+
+        Refer to the Resource model for more details about target fields.
+        """
+        return {
+            "method",
+            "email",
+            "domains",
+            "constructor",
+            "name",
+            "description",
+            "project_id",
+            "uuid",
+        }
+
+    @classmethod
+    def get_new_certificates(
+        cls, limit: int = sc.DEFAULT_SQL_LIMIT
+    ) -> list["Certificate"]:
+        return cls.get_new_entities(
+            cls.__tablename__, sc.CERTIFICATE_KIND, limit
+        )
+
+    @classmethod
+    def get_updated_certificates(
+        cls, limit: int = sc.DEFAULT_SQL_LIMIT
+    ) -> list["Certificate"]:
+        return cls.get_updated_entities(
+            cls.__tablename__, sc.CERTIFICATE_KIND, limit
+        )
+
+    @classmethod
+    def get_deleted_certificates(
+        cls, limit: int = sc.DEFAULT_SQL_LIMIT
+    ) -> list[ua_models.TargetResource]:
+        return cls.get_deleted_target_resources(
+            cls.__tablename__, sc.CERTIFICATE_KIND, limit
+        )