diff --git a/automated_api.py b/automated_api.py index da9fc8af0..16b3ed9cb 100644 --- a/automated_api.py +++ b/automated_api.py @@ -34,6 +34,7 @@ ServerAPI, _PLACEHOLDER, _ActionsAPI, + _ListsAPI, ) from ayon_api.utils import NOT_SET # noqa: E402 @@ -294,6 +295,7 @@ def prepare_api_functions(api_globals): functions = [] _items = list(ServerAPI.__dict__.items()) _items.extend(_ActionsAPI.__dict__.items()) + _items.extend(_ListsAPI.__dict__.items()) for attr_name, attr in _items: if ( attr_name.startswith("_") diff --git a/ayon_api/__init__.py b/ayon_api/__init__.py index cb4f97248..719c54a1c 100644 --- a/ayon_api/__init__.py +++ b/ayon_api/__init__.py @@ -249,6 +249,18 @@ set_action_config, take_action, abort_action, + get_entity_lists, + get_entity_list_rest, + get_entity_list_by_id, + create_entity_list, + update_entity_list, + delete_entity_list, + get_entity_list_attribute_definitions, + set_entity_list_attribute_definitions, + create_entity_list_item, + update_entity_list_items, + update_entity_list_item, + delete_entity_list_item, ) @@ -501,4 +513,16 @@ "set_action_config", "take_action", "abort_action", + "get_entity_lists", + "get_entity_list_rest", + "get_entity_list_by_id", + "create_entity_list", + "update_entity_list", + "delete_entity_list", + "get_entity_list_attribute_definitions", + "set_entity_list_attribute_definitions", + "create_entity_list_item", + "update_entity_list_items", + "update_entity_list_item", + "delete_entity_list_item", ) diff --git a/ayon_api/_api.py b/ayon_api/_api.py index bf16f2848..585c6d44d 100644 --- a/ayon_api/_api.py +++ b/ayon_api/_api.py @@ -41,6 +41,8 @@ from .typing import ( ActivityType, ActivityReferenceType, + EntityListEntityType, + EntityListItemMode, LinkDirection, EventFilter, AttributeScope, @@ -72,6 +74,7 @@ ActionConfigResponse, ActionModeType, StreamType, + EntityListAttributeDefinitionDict, ) @@ -6952,3 +6955,364 @@ def abort_action( action_token=action_token, message=message, ) + + +def get_entity_lists( + project_name: str, + *, + list_ids: Optional[Iterable[str]] = None, + active: Optional[bool] = None, + fields: Optional[Iterable[str]] = None, +) -> Generator[Dict[str, Any], None, None]: + """Fetch entity lists from server. + + Args: + project_name (str): Project name where entity lists are. + list_ids (Optional[Iterable[str]]): List of entity list ids to + fetch. + active (Optional[bool]): Filter by active state of entity lists. + fields (Optional[Iterable[str]]): Fields to fetch from server. + + Returns: + Generator[Dict[str, Any], None, None]: Entity list entities + matching defined filters. + + """ + con = get_server_api_connection() + return con.get_entity_lists( + project_name=project_name, + list_ids=list_ids, + active=active, + fields=fields, + ) + + +def get_entity_list_rest( + project_name: str, + list_id: str, +) -> Optional[Dict[str, Any]]: + """Get entity list by id using REST API. + + Args: + project_name (str): Project name. + list_id (str): Entity list id. + + Returns: + Optional[Dict[str, Any]]: Entity list data or None if not found. + + """ + con = get_server_api_connection() + return con.get_entity_list_rest( + project_name=project_name, + list_id=list_id, + ) + + +def get_entity_list_by_id( + project_name: str, + list_id: str, + fields: Optional[Iterable[str]] = None, +) -> Optional[Dict[str, Any]]: + """Get entity list by id using GraphQl. + + Args: + project_name (str): Project name. + list_id (str): Entity list id. + fields (Optional[Iterable[str]]): Fields to fetch from server. + + Returns: + Optional[Dict[str, Any]]: Entity list data or None if not found. + + """ + con = get_server_api_connection() + return con.get_entity_list_by_id( + project_name=project_name, + list_id=list_id, + fields=fields, + ) + + +def create_entity_list( + project_name: str, + entity_type: "EntityListEntityType", + label: str, + *, + list_type: Optional[str] = None, + access: Optional[Dict[str, Any]] = None, + attrib: Optional[List[Dict[str, Any]]] = None, + data: Optional[List[Dict[str, Any]]] = None, + tags: Optional[List[str]] = None, + template: Optional[Dict[str, Any]] = None, + owner: Optional[str] = None, + active: Optional[bool] = None, + items: Optional[List[Dict[str, Any]]] = None, + list_id: Optional[str] = None, +) -> str: + """Create entity list. + + Args: + project_name (str): Project name where entity list lives. + entity_type (EntityListEntityType): Which entity types can be + used in list. + label (str): Entity list label. + list_type (Optional[str]): Entity list type. + access (Optional[dict[str, Any]]): Access control for entity list. + attrib (Optional[dict[str, Any]]): Attribute values of + entity list. + data (Optional[dict[str, Any]]): Custom data of entity list. + tags (Optional[list[str]]): Entity list tags. + template (Optional[dict[str, Any]]): Dynamic list template. + owner (Optional[str]): New owner of the list. + active (Optional[bool]): Change active state of entity list. + items (Optional[list[dict[str, Any]]]): Initial items in + entity list. + list_id (Optional[str]): Entity list id. + + """ + con = get_server_api_connection() + return con.create_entity_list( + project_name=project_name, + entity_type=entity_type, + label=label, + list_type=list_type, + access=access, + attrib=attrib, + data=data, + tags=tags, + template=template, + owner=owner, + active=active, + items=items, + list_id=list_id, + ) + + +def update_entity_list( + project_name: str, + list_id: str, + *, + label: Optional[str] = None, + access: Optional[Dict[str, Any]] = None, + attrib: Optional[List[Dict[str, Any]]] = None, + data: Optional[List[Dict[str, Any]]] = None, + tags: Optional[List[str]] = None, + owner: Optional[str] = None, + active: Optional[bool] = None, +) -> None: + """Update entity list. + + Args: + project_name (str): Project name where entity list lives. + list_id (str): Entity list id that will be updated. + label (Optional[str]): New label of entity list. + access (Optional[dict[str, Any]]): Access control for entity list. + attrib (Optional[dict[str, Any]]): Attribute values of + entity list. + data (Optional[dict[str, Any]]): Custom data of entity list. + tags (Optional[list[str]]): Entity list tags. + owner (Optional[str]): New owner of the list. + active (Optional[bool]): Change active state of entity list. + + """ + con = get_server_api_connection() + return con.update_entity_list( + project_name=project_name, + list_id=list_id, + label=label, + access=access, + attrib=attrib, + data=data, + tags=tags, + owner=owner, + active=active, + ) + + +def delete_entity_list( + project_name: str, + list_id: str, +) -> None: + """Delete entity list from project. + + Args: + project_name (str): Project name. + list_id (str): Entity list id that will be removed. + + """ + con = get_server_api_connection() + return con.delete_entity_list( + project_name=project_name, + list_id=list_id, + ) + + +def get_entity_list_attribute_definitions( + project_name: str, + list_id: str, +) -> List["EntityListAttributeDefinitionDict"]: + """Get attribute definitioins on entity list. + + Args: + project_name (str): Project name. + list_id (str): Entity list id. + + Returns: + List[EntityListAttributeDefinitionDict]: List of attribute + definitions. + + """ + con = get_server_api_connection() + return con.get_entity_list_attribute_definitions( + project_name=project_name, + list_id=list_id, + ) + + +def set_entity_list_attribute_definitions( + project_name: str, + list_id: str, + attribute_definitions: List["EntityListAttributeDefinitionDict"], +) -> None: + """Set attribute definitioins on entity list. + + Args: + project_name (str): Project name. + list_id (str): Entity list id. + attribute_definitions (List[EntityListAttributeDefinitionDict]): + List of attribute definitions. + + """ + con = get_server_api_connection() + return con.set_entity_list_attribute_definitions( + project_name=project_name, + list_id=list_id, + attribute_definitions=attribute_definitions, + ) + + +def create_entity_list_item( + project_name: str, + list_id: str, + *, + position: Optional[int] = None, + label: Optional[str] = None, + attrib: Optional[Dict[str, Any]] = None, + data: Optional[Dict[str, Any]] = None, + tags: Optional[List[str]] = None, + item_id: Optional[str] = None, +) -> str: + """Create entity list item. + + Args: + project_name (str): Project name where entity list lives. + list_id (str): Entity list id where item will be added. + position (Optional[int]): Position of item in entity list. + label (Optional[str]): Label of item in entity list. + attrib (Optional[dict[str, Any]]): Item attribute values. + data (Optional[dict[str, Any]]): Item data. + tags (Optional[list[str]]): Tags of item in entity list. + item_id (Optional[str]): Id of item that will be created. + + Returns: + str: Item id. + + """ + con = get_server_api_connection() + return con.create_entity_list_item( + project_name=project_name, + list_id=list_id, + position=position, + label=label, + attrib=attrib, + data=data, + tags=tags, + item_id=item_id, + ) + + +def update_entity_list_items( + project_name: str, + list_id: str, + items: List[Dict[str, Any]], + mode: "EntityListItemMode", +) -> None: + """Update items in entity list. + + Args: + project_name (str): Project name where entity list live. + list_id (str): Entity list id. + items (List[Dict[str, Any]]): Entity list items. + mode (EntityListItemMode): Mode of items update. + + """ + con = get_server_api_connection() + return con.update_entity_list_items( + project_name=project_name, + list_id=list_id, + items=items, + mode=mode, + ) + + +def update_entity_list_item( + project_name: str, + list_id: str, + item_id: str, + *, + new_list_id: Optional[str], + position: Optional[int] = None, + label: Optional[str] = None, + attrib: Optional[Dict[str, Any]] = None, + data: Optional[Dict[str, Any]] = None, + tags: Optional[List[str]] = None, +) -> None: + """Update item in entity list. + + Args: + project_name (str): Project name where entity list live. + list_id (str): Entity list id where item lives. + item_id (str): Item id that will be removed from entity list. + new_list_id (Optional[str]): New entity list id where item will be + added. + position (Optional[int]): Position of item in entity list. + label (Optional[str]): Label of item in entity list. + attrib (Optional[dict[str, Any]]): Attributes of item in entity + list. + data (Optional[dict[str, Any]]): Custom data of item in + entity list. + tags (Optional[list[str]]): Tags of item in entity list. + + """ + con = get_server_api_connection() + return con.update_entity_list_item( + project_name=project_name, + list_id=list_id, + item_id=item_id, + new_list_id=new_list_id, + position=position, + label=label, + attrib=attrib, + data=data, + tags=tags, + ) + + +def delete_entity_list_item( + project_name: str, + list_id: str, + item_id: str, +) -> None: + """Delete item from entity list. + + Args: + project_name (str): Project name where entity list live. + list_id (str): Entity list id from which item will be removed. + item_id (str): Item id that will be removed from entity list. + + """ + con = get_server_api_connection() + return con.delete_entity_list_item( + project_name=project_name, + list_id=list_id, + item_id=item_id, + ) diff --git a/ayon_api/_base.py b/ayon_api/_base.py index a87d5800b..b3863a2c4 100644 --- a/ayon_api/_base.py +++ b/ayon_api/_base.py @@ -1,9 +1,46 @@ +import typing +from typing import Set + +if typing.TYPE_CHECKING: + from .typing import AnyEntityDict + + class _BaseServerAPI: + def get(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def post(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def put(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def patch(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def delete(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def raw_get(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def raw_post(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def raw_put(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def raw_patch(self, entrypoint: str, **kwargs): + raise NotImplementedError() + + def raw_delete(self, entrypoint: str, **kwargs): + raise NotImplementedError() + def get_default_settings_variant(self) -> str: raise NotImplementedError() - def get(self, entrypoint: str, **kwargs): - pass + def get_default_fields_for_type(self, entity_type: str) -> Set[str]: + raise NotImplementedError() - def post(self, entrypoint: str, **kwargs): - pass + def _convert_entity_data(self, entity: "AnyEntityDict"): + raise NotImplementedError() diff --git a/ayon_api/_lists.py b/ayon_api/_lists.py new file mode 100644 index 000000000..480c1e613 --- /dev/null +++ b/ayon_api/_lists.py @@ -0,0 +1,414 @@ +import json +import typing +from typing import Optional, Iterable, Any, Dict, List, Generator + +from ._base import _BaseServerAPI +from .utils import create_entity_id +from .graphql_queries import entity_lists_graphql_query + +if typing.TYPE_CHECKING: + from .typing import ( + EntityListEntityType, + EntityListAttributeDefinitionDict, + EntityListItemMode, + ) + + +class _ListsAPI(_BaseServerAPI): + def get_entity_lists( + self, + project_name: str, + *, + list_ids: Optional[Iterable[str]] = None, + active: Optional[bool] = None, + fields: Optional[Iterable[str]] = None, + ) -> Generator[Dict[str, Any], None, None]: + """Fetch entity lists from server. + + Args: + project_name (str): Project name where entity lists are. + list_ids (Optional[Iterable[str]]): List of entity list ids to + fetch. + active (Optional[bool]): Filter by active state of entity lists. + fields (Optional[Iterable[str]]): Fields to fetch from server. + + Returns: + Generator[Dict[str, Any], None, None]: Entity list entities + matching defined filters. + + """ + if fields is None: + fields = self.get_default_fields_for_type("entityList") + fields = set(fields) + + if active is not None: + fields.add("active") + + filters: Dict[str, Any] = {"projectName": project_name} + if list_ids is not None: + if not list_ids: + return + filters["listIds"] = list(set(list_ids)) + + query = entity_lists_graphql_query(fields) + for attr, filter_value in filters.items(): + query.set_variable_value(attr, filter_value) + + for parsed_data in query.continuous_query(self): + for entity_list in parsed_data["project"]["entityLists"]: + if active is not None and entity_list["active"] != active: + continue + + attributes = entity_list.get("attributes") + if isinstance(attributes, str): + entity_list["attributes"] = json.loads(attributes) + + self._convert_entity_data(entity_list) + + yield entity_list + + def get_entity_list_rest( + self, project_name: str, list_id: str + ) -> Optional[Dict[str, Any]]: + """Get entity list by id using REST API. + + Args: + project_name (str): Project name. + list_id (str): Entity list id. + + Returns: + Optional[Dict[str, Any]]: Entity list data or None if not found. + + """ + response = self.get(f"projects/{project_name}/lists/{list_id}") + response.raise_for_status() + return response.data + + def get_entity_list_by_id( + self, + project_name: str, + list_id: str, + fields: Optional[Iterable[str]] = None, + ) -> Optional[Dict[str, Any]]: + """Get entity list by id using GraphQl. + + Args: + project_name (str): Project name. + list_id (str): Entity list id. + fields (Optional[Iterable[str]]): Fields to fetch from server. + + Returns: + Optional[Dict[str, Any]]: Entity list data or None if not found. + + """ + for entity_list in self.get_entity_lists( + project_name, list_ids=[list_id], active=None, fields=fields + ): + return entity_list + return None + + def create_entity_list( + self, + project_name: str, + entity_type: "EntityListEntityType", + label: str, + *, + list_type: Optional[str] = None, + access: Optional[Dict[str, Any]] = None, + attrib: Optional[List[Dict[str, Any]]] = None, + data: Optional[List[Dict[str, Any]]] = None, + tags: Optional[List[str]] = None, + template: Optional[Dict[str, Any]] = None, + owner: Optional[str] = None, + active: Optional[bool] = None, + items: Optional[List[Dict[str, Any]]] = None, + list_id: Optional[str] = None, + ) -> str: + """Create entity list. + + Args: + project_name (str): Project name where entity list lives. + entity_type (EntityListEntityType): Which entity types can be + used in list. + label (str): Entity list label. + list_type (Optional[str]): Entity list type. + access (Optional[dict[str, Any]]): Access control for entity list. + attrib (Optional[dict[str, Any]]): Attribute values of + entity list. + data (Optional[dict[str, Any]]): Custom data of entity list. + tags (Optional[list[str]]): Entity list tags. + template (Optional[dict[str, Any]]): Dynamic list template. + owner (Optional[str]): New owner of the list. + active (Optional[bool]): Change active state of entity list. + items (Optional[list[dict[str, Any]]]): Initial items in + entity list. + list_id (Optional[str]): Entity list id. + + """ + if list_id is None: + list_id = create_entity_id() + kwargs = { + "id": list_id, + "entityType": entity_type, + "label": label, + } + for key, value in ( + ("entityListType", list_type), + ("access", access), + ("attrib", attrib), + ("template", template), + ("tags", tags), + ("owner", owner), + ("data", data), + ("active", active), + ("items", items), + ): + if value is not None: + kwargs[key] = value + + response = self.post( + f"projects/{project_name}/lists/{list_id}/items", + **kwargs + + ) + response.raise_for_status() + return list_id + + def update_entity_list( + self, + project_name: str, + list_id: str, + *, + label: Optional[str] = None, + access: Optional[Dict[str, Any]] = None, + attrib: Optional[List[Dict[str, Any]]] = None, + data: Optional[List[Dict[str, Any]]] = None, + tags: Optional[List[str]] = None, + owner: Optional[str] = None, + active: Optional[bool] = None, + ) -> None: + """Update entity list. + + Args: + project_name (str): Project name where entity list lives. + list_id (str): Entity list id that will be updated. + label (Optional[str]): New label of entity list. + access (Optional[dict[str, Any]]): Access control for entity list. + attrib (Optional[dict[str, Any]]): Attribute values of + entity list. + data (Optional[dict[str, Any]]): Custom data of entity list. + tags (Optional[list[str]]): Entity list tags. + owner (Optional[str]): New owner of the list. + active (Optional[bool]): Change active state of entity list. + + """ + kwargs = { + key: value + for key, value in ( + ("label", label), + ("access", access), + ("attrib", attrib), + ("data", data), + ("tags", tags), + ("owner", owner), + ("active", active), + ) + if value is not None + } + response = self.patch( + f"projects/{project_name}/lists/{list_id}", + **kwargs + ) + response.raise_for_status() + + def delete_entity_list(self, project_name: str, list_id: str) -> None: + """Delete entity list from project. + + Args: + project_name (str): Project name. + list_id (str): Entity list id that will be removed. + + """ + response = self.delete(f"projects/{project_name}/lists/{list_id}") + response.raise_for_status() + + def get_entity_list_attribute_definitions( + self, project_name: str, list_id: str + ) -> List["EntityListAttributeDefinitionDict"]: + """Get attribute definitioins on entity list. + + Args: + project_name (str): Project name. + list_id (str): Entity list id. + + Returns: + List[EntityListAttributeDefinitionDict]: List of attribute + definitions. + + """ + response = self.get( + f"projects/{project_name}/lists/{list_id}/attributes" + ) + response.raise_for_status() + return response.data + + def set_entity_list_attribute_definitions( + self, + project_name: str, + list_id: str, + attribute_definitions: List["EntityListAttributeDefinitionDict"], + ) -> None: + """Set attribute definitioins on entity list. + + Args: + project_name (str): Project name. + list_id (str): Entity list id. + attribute_definitions (List[EntityListAttributeDefinitionDict]): + List of attribute definitions. + + """ + response = self.raw_put( + f"projects/{project_name}/lists/{list_id}/attributes", + json=attribute_definitions, + ) + response.raise_for_status() + + def create_entity_list_item( + self, + project_name: str, + list_id: str, + *, + position: Optional[int] = None, + label: Optional[str] = None, + attrib: Optional[Dict[str, Any]] = None, + data: Optional[Dict[str, Any]] = None, + tags: Optional[List[str]] = None, + item_id: Optional[str] = None, + ) -> str: + """Create entity list item. + + Args: + project_name (str): Project name where entity list lives. + list_id (str): Entity list id where item will be added. + position (Optional[int]): Position of item in entity list. + label (Optional[str]): Label of item in entity list. + attrib (Optional[dict[str, Any]]): Item attribute values. + data (Optional[dict[str, Any]]): Item data. + tags (Optional[list[str]]): Tags of item in entity list. + item_id (Optional[str]): Id of item that will be created. + + Returns: + str: Item id. + + """ + if item_id is None: + item_id = create_entity_id() + kwargs = { + "id": item_id, + "entityId": list_id, + } + for key, value in ( + ("position", position), + ("label", label), + ("attrib", attrib), + ("data", data), + ("tags", tags), + ): + if value is not None: + kwargs[key] = value + + response = self.post( + f"projects/{project_name}/lists/{list_id}/items", + **kwargs + ) + response.raise_for_status() + return item_id + + def update_entity_list_items( + self, + project_name: str, + list_id: str, + items: List[Dict[str, Any]], + mode: "EntityListItemMode", + ) -> None: + """Update items in entity list. + + Args: + project_name (str): Project name where entity list live. + list_id (str): Entity list id. + items (List[Dict[str, Any]]): Entity list items. + mode (EntityListItemMode): Mode of items update. + + """ + response = self.post( + f"projects/{project_name}/lists/{list_id}/items", + items=items, + mode=mode, + ) + response.raise_for_status() + + def update_entity_list_item( + self, + project_name: str, + list_id: str, + item_id: str, + *, + new_list_id: Optional[str], + position: Optional[int] = None, + label: Optional[str] = None, + attrib: Optional[Dict[str, Any]] = None, + data: Optional[Dict[str, Any]] = None, + tags: Optional[List[str]] = None, + ) -> None: + """Update item in entity list. + + Args: + project_name (str): Project name where entity list live. + list_id (str): Entity list id where item lives. + item_id (str): Item id that will be removed from entity list. + new_list_id (Optional[str]): New entity list id where item will be + added. + position (Optional[int]): Position of item in entity list. + label (Optional[str]): Label of item in entity list. + attrib (Optional[dict[str, Any]]): Attributes of item in entity + list. + data (Optional[dict[str, Any]]): Custom data of item in + entity list. + tags (Optional[list[str]]): Tags of item in entity list. + + """ + kwargs = {} + for key, value in ( + ("entityId", new_list_id), + ("position", position), + ("label", label), + ("attrib", attrib), + ("data", data), + ("tags", tags), + ): + if value is not None: + kwargs[key] = value + response = self.patch( + f"projects/{project_name}/lists/{list_id}/items/{item_id}", + **kwargs, + ) + response.raise_for_status() + + def delete_entity_list_item( + self, + project_name: str, + list_id: str, + item_id: str, + ) -> None: + """Delete item from entity list. + + Args: + project_name (str): Project name where entity list live. + list_id (str): Entity list id from which item will be removed. + item_id (str): Item id that will be removed from entity list. + + """ + response = self.delete( + f"projects/{project_name}/lists/{list_id}/items/{item_id}", + ) + response.raise_for_status() diff --git a/ayon_api/constants.py b/ayon_api/constants.py index 191d64219..d47e4fc59 100644 --- a/ayon_api/constants.py +++ b/ayon_api/constants.py @@ -227,3 +227,21 @@ "entityType", "author.name", } + + +DEFAULT_ENTITY_LIST_FIELDS = { + "id", + "count", + "attributes", + "active", + "createdBy", + "createdAt", + "entityListType", + "data", + "entityType", + "label", + "owner", + "tags", + "updatedAt", + "updatedBy", +} diff --git a/ayon_api/graphql_queries.py b/ayon_api/graphql_queries.py index 169f493ec..54648f7d7 100644 --- a/ayon_api/graphql_queries.py +++ b/ayon_api/graphql_queries.py @@ -665,3 +665,33 @@ def activities_graphql_query(fields, order): query_queue.append((k, v, field)) return query + + +def entity_lists_graphql_query(fields): + query = GraphQlQuery("EntityLists") + project_name_var = query.add_variable("projectName", "String!") + entity_list_ids = query.add_variable("listIds", "[String!]") + + project_field = query.add_field("project") + project_field.set_filter("name", project_name_var) + + entity_lists_field = project_field.add_field_with_edges("entityLists") + entity_lists_field.set_filter("ids", entity_list_ids) + + nested_fields = fields_to_dict(set(fields)) + + query_queue = collections.deque() + for key, value in nested_fields.items(): + query_queue.append((key, value, entity_lists_field)) + + while query_queue: + item = query_queue.popleft() + key, value, parent = item + field = parent.add_field(key) + if value is FIELD_VALUE: + continue + + for k, v in value.items(): + query_queue.append((k, v, field)) + + return query diff --git a/ayon_api/server_api.py b/ayon_api/server_api.py index 023b70e95..4fecb75da 100644 --- a/ayon_api/server_api.py +++ b/ayon_api/server_api.py @@ -55,6 +55,7 @@ DEFAULT_EVENT_FIELDS, DEFAULT_ACTIVITY_FIELDS, DEFAULT_USER_FIELDS, + DEFAULT_ENTITY_LIST_FIELDS, ) from .graphql import GraphQlQuery, INTROSPECTION_QUERY from .graphql_queries import ( @@ -100,6 +101,7 @@ get_machine_name, ) from ._actions import _ActionsAPI +from ._lists import _ListsAPI if typing.TYPE_CHECKING: from typing import Union @@ -424,7 +426,7 @@ def as_user(self, username): self._last_user = new_last_user -class ServerAPI(_ActionsAPI): +class ServerAPI(_ListsAPI, _ActionsAPI): """Base handler of connection to server. Requires url to server which is used as base for api and graphql calls. @@ -2784,6 +2786,9 @@ def get_default_fields_for_type(self, entity_type: str) -> Set[str]: elif entity_type == "user": entity_type_defaults = set(DEFAULT_USER_FIELDS) + elif entity_type == "entityList": + entity_type_defaults = set(DEFAULT_ENTITY_LIST_FIELDS) + else: raise ValueError(f"Unknown entity type \"{entity_type}\"") return ( diff --git a/ayon_api/typing.py b/ayon_api/typing.py index 81d854de5..32c582bea 100644 --- a/ayon_api/typing.py +++ b/ayon_api/typing.py @@ -29,6 +29,21 @@ "watching", ] +EntityListEntityType = Literal[ + "folder", + "product", + "version", + "representation", + "task", + "workfile", +] + +EntityListItemMode = Literal[ + "replace", + "merge", + "delete", +] + EventFilterValueType = Union[ None, str, int, float, @@ -472,3 +487,8 @@ class ActionConfigResponse(TypedDict): StreamType = Union[io.BytesIO, BinaryIO] + + +class EntityListAttributeDefinitionDict(TypedDict): + name: str + data: Dict[str, Any]