Entity lists

ayon_server/entity_lists/entity_list.py

@@ -0,0 +1,94 @@
+from typing import Any, overload
+
+from ayon_server.entities.user import UserEntity
+from ayon_server.exceptions import ForbiddenException, NotFoundException
+from ayon_server.lib.postgres import Connection, Postgres
+
+from .models import EntityListItemModel, EntityListModel
+
+
+class EntityList:
+    project_name: str
+    payload: EntityListModel
+    user: UserEntity | None = None
+    exists: bool = False
+
+    @overload
+    def __init__(self, project_name: str, payload: dict[str, Any]) -> None: ...
+
+    @overload
+    def __init__(self, project_name: str, payload: EntityListModel) -> None: ...
+
+    def __init__(
+        self, project_name: str, payload: dict[str, Any] | EntityListModel
+    ) -> None:
+        if isinstance(payload, dict):
+            self.payload = EntityListModel(**payload)
+        else:
+            self.payload = payload
+        self.project_name = project_name
+
+    @classmethod
+    async def _load(
+        cls,
+        project_name: str,
+        id: str,
+        *,
+        user: UserEntity | None,
+        conn: Connection,
+    ) -> "EntityList":
+        res = await conn.fetch(
+            f"""
+            SELECT * FROM project_{project_name}.entity_lists
+            WHERE id = $1
+            """,
+            id,
+        )
+
+        if not res:
+            raise NotFoundException("Entity list not found")
+
+        entity_list = EntityListModel(**res[0])
+
+        if user is not None:
+            # Check user access to the list
+
+            if entity_list.access.get(user.name, 0) < 10:
+                raise ForbiddenException("User does not have access to this list")
+
+        q = f"""
+            SELECT * FROM project_{project_name}.entity_list_items
+            WHERE list_id = $1
+            ORDER BY position
+            """
+
+        async for row in Postgres.iterate(q, id):
+            list_item = EntityListItemModel(**row)
+            entity_list.items.append(list_item)
+
+        result = cls(project_name, entity_list)
+        result.user = user
+        result.exists = True
+
+    @classmethod
+    async def load(
+        cls,
+        project_name: str,
+        id: str,
+        *,
+        user: UserEntity | None = None,
+        conn: Connection | None = None,
+    ) -> "EntityList":
+        if conn is None:
+            async with Postgres.acquire() as c, c.transaction():
+                return await cls._load(project_name, id, conn=c, user=user)
+        else:
+            return await cls._load(project_name, id, conn=conn, user=user)
+
+    async def save(
+        self,
+        *,
+        user: UserEntity | None = None,
+        conn: Connection | None = None,
+    ) -> None:
+        user = user or self.user

ayon_server/entity_lists/models.py

@@ -0,0 +1,40 @@
+from datetime import datetime
+from enum import IntEnum
+from typing import Annotated, Any
+
+from ayon_server.types import Field, OPModel, ProjectLevelEntityType
+from ayon_server.utils import create_uuid
+
+
+class ListAccessLevel(IntEnum):
+    READ = 10  # Can view the list and its items
+    UPDATE = 20  # Can update attributes of the list items
+    CONSTRUCT = 30  # Can add/remove items from the list or materialize new items
+    ADMIN = 40  # Can update/delete the list itself and add new users to the list
+
+
+class EntityListItemModel(OPModel):
+    id: Annotated[str, Field(default_factory=create_uuid)]
+
+    entity_type: ProjectLevelEntityType
+    entity_id: str = Field(...)
+    position: Annotated[int, Field(0)]
+
+    attrib: dict[str, Any] = Field(default_factory=dict)
+    data: dict[str, Any] = Field(default_factory=dict)
+    tags: list[str] = Field(default_factory=list)
+
+    created_by: str | None = None
+    updated_by: str | None = None
+    created_at: datetime | None = Field(default_factory=datetime.utcnow)
+    updated_at: datetime | None = Field(default_factory=datetime.utcnow)
+
+
+class EntityListModel(OPModel):
+    id: Annotated[str, Field(default_factory=create_uuid)]
+    label: Annotated[str, Field(...)]
+    list_type: Annotated[str, Field(...)]
+    template: Annotated[dict[str, Any] | None, Field()] = None
+    tags: list[str] = Field(default_factory=list)
+    items: list[EntityListItemModel] = Field(default_factory=list)
+    access: dict[str, ListAccessLevel] = Field(default_factory=dict)

schemas/schema.project.sql

@@ -356,6 +356,47 @@ CREATE TABLE workfiles(
     creation_order SERIAL NOT NULL
 );
 
+-----------
+-- Lists --
+-----------
+
+
+CREATE TABLE entity_lists(
+  id UUID NOT NULL PRIMARY KEY,
+  label VARCHAR NOT NULL,
+  list_type VARCHAR NOT NULL, 
+  access JSONB NOT NULL DEFAULT '{}'::JSONB,
+  template JSONB NOT NULL DEFAULT '{}'::JSONB,
+
+  attrib JSONB NOT NULL DEFAULT '{}'::JSONB,
+  data JSONB NOT NULL DEFAULT '{}'::JSONB,
+  tags VARCHAR[] NOT NULL DEFAULT ARRAY[]::VARCHAR[],
+
+  active BOOLEAN NOT NULL DEFAULT TRUE,
+  created_at TIMESTAMPTZ DEFAULT NOW(),
+  updated_at TIMESTAMPTZ DEFAULT NOW()
+);
+
+
+CREATE TABLE entity_list_items(
+  id UUID NOT NULL PRIMARY KEY,
+  entity_list_id UUID NOT NULL REFERENCES lists(id) ON DELETE CASCADE,
+  entity_type VARCHAR NOT NULL,
+  entity_id UUID NOT NULL,
+  position INTEGER NOT NULL,
+
+  attrib JSONB NOT NULL DEFAULT '{}'::JSONB,
+  data JSONB NOT NULL DEFAULT '{}'::JSONB,
+  tags VARCHAR[] NOT NULL DEFAULT ARRAY[]::VARCHAR[],
+
+  created_by VARCHAR REFERENCES public.users(name) ON UPDATE CASCADE ON DELETE SET NULL,
+  updated_by VARCHAR REFERENCES public.users(name) ON UPDATE CASCADE ON DELETE SET NULL,
+  created_at TIMESTAMPTZ DEFAULT NOW(),
+  updated_at TIMESTAMPTZ DEFAULT NOW()
+  creation_order SERIAL NOT NULL
+);
+
+
 -----------
 -- LINKS --
 -----------