Added auth & improved database

src/cpl-auth/cpl/auth/schema/_administration/api_key.py

@@ -0,0 +1,59 @@
+import secrets
+from datetime import datetime
+from typing import Optional
+
+from async_property import async_property
+
+from cpl.auth.permission.permissions import Permissions
+from cpl.core.environment import Environment
+from cpl.core.log import Logger
+from cpl.core.typing import SerialId, Id
+from cpl.database.abc import DbModelABC
+from cpl.dependency import ServiceProviderABC
+
+_logger = Logger(__name__)
+
+
+class ApiKey(DbModelABC):
+    def __init__(
+        self,
+        id: SerialId,
+        identifier: str,
+        key: str,
+        deleted: bool = False,
+        editor_id: Optional[Id] = None,
+        created: Optional[datetime] = None,
+        updated: Optional[datetime] = None,
+    ):
+        DbModelABC.__init__(self, id, deleted, editor_id, created, updated)
+        self._identifier = identifier
+        self._key = key
+
+    @property
+    def identifier(self) -> str:
+        return self._identifier
+
+    @property
+    def key(self) -> str:
+        return self._key
+
+    @async_property
+    async def permissions(self):
+        from cpl.auth.schema._permission.api_key_permission_dao import ApiKeyPermissionDao
+
+        api_key_permission_dao: ApiKeyPermissionDao = ServiceProviderABC.get_global_service(ApiKeyPermissionDao)
+        return [await x.permission for x in await api_key_permission_dao.find_by_api_key_id(self.id)]
+
+    async def has_permission(self, permission: Permissions) -> bool:
+        return permission.value in [x.name for x in await self.permissions]
+
+    def set_new_api_key(self):
+        self._key = self.new_key()
+
+    @staticmethod
+    def new_key() -> str:
+        return f"api_{secrets.token_urlsafe(Environment.get("API_KEY_LENGTH", int, 64))}"
+
+    @classmethod
+    def new(cls, identifier: str) -> "ApiKey":
+        return ApiKey(0, identifier, cls.new_key())