feat: Add RBAC Entity Creator

Author: fregataa

src/ai/backend/manager/models/rbac_models/creator.py

@@ -0,0 +1,51 @@
+import logging
+from abc import ABC
+from dataclasses import dataclass
+from typing import Any, override
+
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession as SASession
+
+from ai.backend.logging import BraceStyleAdapter
+from ai.backend.manager.data.permission.association_scopes_entities import (
+    AssociationScopesEntitiesCreateInput,
+)
+from ai.backend.manager.data.permission.id import ObjectId, ScopeId
+from ai.backend.manager.models.rbac_models.association_scopes_entities import (
+    AssociationScopesEntitiesRow,
+)
+from ai.backend.manager.types import Creator
+
+log = BraceStyleAdapter(logging.getLogger(__name__))
+
+
+@dataclass
+class RBACCreator(Creator, ABC):
+    scope_id: ScopeId
+    entity_id: ObjectId
+
+    @override
+    def fields_to_store(self) -> dict[str, Any]:
+        return {
+            "scope_id": self.scope_id,
+            "entity_id": self.entity_id,
+        }
+
+    async def create_rbac_row(self, db_session: SASession) -> None:
+        creator = AssociationScopesEntitiesCreateInput(
+            scope_id=self.scope_id,
+            object_id=self.entity_id,
+        )
+        try:
+            await db_session.execute(
+                pg_insert(AssociationScopesEntitiesRow)
+                .values(creator.fields_to_store())
+                .on_conflict_do_nothing()
+            )
+        except IntegrityError:
+            log.exception(
+                "entity and scope mapping already exists: {}, {}. Skipping.",
+                self.entity_id.to_str(),
+                self.scope_id.to_str(),
+            )