From 3a1dd63f9a17cd7e37b0670f43649d60ed9acfd8 Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Mon, 30 Jun 2025 16:41:51 +0200 Subject: [PATCH 01/10] added constnas and typing helpers for entity lists --- ayon_api/constants.py | 18 ++++++++++++++++++ ayon_api/typing.py | 20 ++++++++++++++++++++ 2 files changed, 38 insertions(+) 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/typing.py b/ayon_api/typing.py index 07f041aab..9c84489c2 100644 --- a/ayon_api/typing.py +++ b/ayon_api/typing.py @@ -28,6 +28,21 @@ "watching", ] +EntityListEntityType = Literal[ + "folder", + "product", + "version", + "representation", + "task", + "workfile", +] + +EntityListItemMode = Literal[ + "replace", + "merge", + "delete", +] + EventFilterValueType = Union[ None, str, int, float, @@ -353,3 +368,8 @@ class ProductTypeDict(TypedDict): StreamType = Union[io.BytesIO, BinaryIO] + + +class EntityListAttributeDefinitionDict(TypedDict): + name: str + data: Dict[str, Any] From fed13c387ac9bc6a7111b467f975173d4c5cbd21 Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Mon, 30 Jun 2025 16:42:46 +0200 Subject: [PATCH 02/10] implemented entity list methods --- ayon_api/graphql_queries.py | 30 +++ ayon_api/server_api.py | 406 ++++++++++++++++++++++++++++++++++++ 2 files changed, 436 insertions(+) diff --git a/ayon_api/graphql_queries.py b/ayon_api/graphql_queries.py index aef04730a..8973ee12e 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 a6d5aa249..818409bd2 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 ( @@ -71,6 +72,7 @@ events_graphql_query, users_graphql_query, activities_graphql_query, + entity_lists_graphql_query, ) from .exceptions import ( FailedOperations, @@ -105,6 +107,8 @@ from .typing import ( ActivityType, ActivityReferenceType, + EntityListEntityType, + EntityListItemMode, LinkDirection, EventFilter, AttributeScope, @@ -132,6 +136,7 @@ ProjectHierarchyDict, ProductTypeDict, StreamType, + EntityListAttributeDefinitionDict, ) PatternType = type(re.compile("")) @@ -2764,6 +2769,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 ( @@ -8854,6 +8862,404 @@ def get_representation_links( project_name, [representation_id], link_types, link_direction )[representation_id] + 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 = {"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() + # --- Batch operations processing --- def send_batch_operations( self, From d234dc15bdf475f0d06d94fa553fa2deb2dcfc7c Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Mon, 30 Jun 2025 16:42:59 +0200 Subject: [PATCH 03/10] modified automated api --- automated_api.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/automated_api.py b/automated_api.py index 329c6c858..55b0cb594 100644 --- a/automated_api.py +++ b/automated_api.py @@ -249,6 +249,9 @@ def sig_params_to_str(sig, param_names, api_globals, indent=0): body_params.append(f"*{var_positional}") func_params.append(f"*{var_positional}") + elif kw_only: + func_params.append(f"*") + for param_name, param in kw_only: body_params.append(f"{param_name}={param_name}") func_params.append(_kw_default_to_str(param_name, param, api_globals)) From 46e1d1d6cd4bbff17ecc03323b8df06ac9a64cb4 Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Mon, 30 Jun 2025 16:43:13 +0200 Subject: [PATCH 04/10] created public functions for entity lists --- ayon_api/__init__.py | 24 +++ ayon_api/_api.py | 361 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 385 insertions(+) diff --git a/ayon_api/__init__.py b/ayon_api/__init__.py index 373cbad88..1d6c0eeb5 100644 --- a/ayon_api/__init__.py +++ b/ayon_api/__init__.py @@ -241,6 +241,18 @@ get_version_links, get_representations_links, get_representation_links, + 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, send_batch_operations, send_activities_batch_operations, ) @@ -487,6 +499,18 @@ "get_version_links", "get_representations_links", "get_representation_links", + "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", "send_batch_operations", "send_activities_batch_operations", ) diff --git a/ayon_api/_api.py b/ayon_api/_api.py index 23e9769aa..bfbc77748 100644 --- a/ayon_api/_api.py +++ b/ayon_api/_api.py @@ -6646,6 +6646,367 @@ def get_representation_links( ) +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, + ) + + def send_batch_operations( project_name: str, operations: List[Dict[str, Any]], From df474b3f254b467b3cf9b449d0f6f8fc6f914ff7 Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Fri, 1 Aug 2025 14:32:25 +0200 Subject: [PATCH 05/10] add missing import --- ayon_api/_api.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/ayon_api/_api.py b/ayon_api/_api.py index bfbc77748..3749390b9 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, @@ -66,6 +68,7 @@ ProjectHierarchyDict, ProductTypeDict, StreamType, + EntityListAttributeDefinitionDict, ) From ee25f39c93061b8cf63b4036c8413194372bc2a1 Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Fri, 1 Aug 2025 14:38:17 +0200 Subject: [PATCH 06/10] remove unnecessary f-string --- automated_api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/automated_api.py b/automated_api.py index 55b0cb594..74c936f66 100644 --- a/automated_api.py +++ b/automated_api.py @@ -250,7 +250,7 @@ def sig_params_to_str(sig, param_names, api_globals, indent=0): func_params.append(f"*{var_positional}") elif kw_only: - func_params.append(f"*") + func_params.append("*") for param_name, param in kw_only: body_params.append(f"{param_name}={param_name}") From 2bf8570d685f3c92a37e3b08cf6fce7d3a42caf7 Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Mon, 4 Aug 2025 17:36:46 +0200 Subject: [PATCH 07/10] fix list ids variable type --- ayon_api/graphql_queries.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ayon_api/graphql_queries.py b/ayon_api/graphql_queries.py index 7db022e82..54648f7d7 100644 --- a/ayon_api/graphql_queries.py +++ b/ayon_api/graphql_queries.py @@ -670,7 +670,7 @@ def activities_graphql_query(fields, order): 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!") + entity_list_ids = query.add_variable("listIds", "[String!]") project_field = query.add_field("project") project_field.set_filter("name", project_name_var) From 13309b55da599161dba40bee0d295a9dd824cbc2 Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Tue, 5 Aug 2025 09:41:02 +0200 Subject: [PATCH 08/10] Fix typos Co-authored-by: Petr Kalis --- ayon_api/_api.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ayon_api/_api.py b/ayon_api/_api.py index cd0bb981c..b2e16fafe 100644 --- a/ayon_api/_api.py +++ b/ayon_api/_api.py @@ -6844,7 +6844,7 @@ def get_entity_list_attribute_definitions( project_name: str, list_id: str, ) -> List["EntityListAttributeDefinitionDict"]: - """Get attribute definitioins on entity list. + """Get attribute definitions on entity list. Args: project_name (str): Project name. @@ -6867,7 +6867,7 @@ def set_entity_list_attribute_definitions( list_id: str, attribute_definitions: List["EntityListAttributeDefinitionDict"], ) -> None: - """Set attribute definitioins on entity list. + """Set attribute definitions on entity list. Args: project_name (str): Project name. From 4523c8afe1048ef859abf3cadc5bb7c1d17fe91e Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Wed, 6 Aug 2025 17:00:25 +0200 Subject: [PATCH 09/10] move lists api to separate class --- automated_api.py | 2 + ayon_api/_base.py | 45 ++++- ayon_api/_lists.py | 414 +++++++++++++++++++++++++++++++++++++++++ ayon_api/server_api.py | 405 +--------------------------------------- 4 files changed, 459 insertions(+), 407 deletions(-) create mode 100644 ayon_api/_lists.py 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/_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/server_api.py b/ayon_api/server_api.py index b819a9f9d..4fecb75da 100644 --- a/ayon_api/server_api.py +++ b/ayon_api/server_api.py @@ -72,7 +72,6 @@ events_graphql_query, users_graphql_query, activities_graphql_query, - entity_lists_graphql_query, ) from .exceptions import ( FailedOperations, @@ -102,14 +101,13 @@ get_machine_name, ) from ._actions import _ActionsAPI +from ._lists import _ListsAPI if typing.TYPE_CHECKING: from typing import Union from .typing import ( ActivityType, ActivityReferenceType, - EntityListEntityType, - EntityListItemMode, LinkDirection, EventFilter, AttributeScope, @@ -137,7 +135,6 @@ ProjectHierarchyDict, ProductTypeDict, StreamType, - EntityListAttributeDefinitionDict, ) PatternType = type(re.compile("")) @@ -429,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. @@ -8882,404 +8879,6 @@ def get_representation_links( project_name, [representation_id], link_types, link_direction )[representation_id] - 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 = {"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() - # --- Batch operations processing --- def send_batch_operations( self, From cc326e82ee8932414ef84151c13618e67de7a02e Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Wed, 6 Aug 2025 17:08:31 +0200 Subject: [PATCH 10/10] change order of functions --- ayon_api/__init__.py | 32 +- ayon_api/_api.py | 908 +++++++++++++++++++++---------------------- 2 files changed, 470 insertions(+), 470 deletions(-) diff --git a/ayon_api/__init__.py b/ayon_api/__init__.py index 7515df97f..719c54a1c 100644 --- a/ayon_api/__init__.py +++ b/ayon_api/__init__.py @@ -241,6 +241,14 @@ get_version_links, get_representations_links, get_representation_links, + send_batch_operations, + send_activities_batch_operations, + get_actions, + trigger_action, + get_action_config, + set_action_config, + take_action, + abort_action, get_entity_lists, get_entity_list_rest, get_entity_list_by_id, @@ -253,14 +261,6 @@ update_entity_list_items, update_entity_list_item, delete_entity_list_item, - send_batch_operations, - send_activities_batch_operations, - get_actions, - trigger_action, - get_action_config, - set_action_config, - take_action, - abort_action, ) @@ -505,6 +505,14 @@ "get_version_links", "get_representations_links", "get_representation_links", + "send_batch_operations", + "send_activities_batch_operations", + "get_actions", + "trigger_action", + "get_action_config", + "set_action_config", + "take_action", + "abort_action", "get_entity_lists", "get_entity_list_rest", "get_entity_list_by_id", @@ -517,12 +525,4 @@ "update_entity_list_items", "update_entity_list_item", "delete_entity_list_item", - "send_batch_operations", - "send_activities_batch_operations", - "get_actions", - "trigger_action", - "get_action_config", - "set_action_config", - "take_action", - "abort_action", ) diff --git a/ayon_api/_api.py b/ayon_api/_api.py index 0a7e0d63a..585c6d44d 100644 --- a/ayon_api/_api.py +++ b/ayon_api/_api.py @@ -6657,662 +6657,662 @@ def get_representation_links( ) -def get_entity_lists( +def send_batch_operations( 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. + operations: List[Dict[str, Any]], + can_fail: bool = False, + raise_on_fail: bool = True, +) -> List[Dict[str, Any]]: + """Post multiple CRUD operations to server. + + When multiple changes should be made on server side this is the best + way to go. It is possible to pass multiple operations to process on a + server side and do the changes in a transaction. 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. + project_name (str): On which project should be operations + processed. + operations (list[dict[str, Any]]): Operations to be processed. + can_fail (Optional[bool]): Server will try to process all + operations even if one of them fails. + raise_on_fail (Optional[bool]): Raise exception if an operation + fails. You can handle failed operations on your own + when set to 'False'. + + Raises: + ValueError: Operations can't be converted to json string. + FailedOperations: When output does not contain server operations + or 'raise_on_fail' is enabled and any operation fails. Returns: - Generator[Dict[str, Any], None, None]: Entity list entities - matching defined filters. + list[dict[str, Any]]: Operations result with process details. """ con = get_server_api_connection() - return con.get_entity_lists( + return con.send_batch_operations( project_name=project_name, - list_ids=list_ids, - active=active, - fields=fields, + operations=operations, + can_fail=can_fail, + raise_on_fail=raise_on_fail, ) -def get_entity_list_rest( +def send_activities_batch_operations( project_name: str, - list_id: str, -) -> Optional[Dict[str, Any]]: - """Get entity list by id using REST API. + operations: List[Dict[str, Any]], + can_fail: bool = False, + raise_on_fail: bool = True, +) -> List[Dict[str, Any]]: + """Post multiple CRUD activities operations to server. + + When multiple changes should be made on server side this is the best + way to go. It is possible to pass multiple operations to process on a + server side and do the changes in a transaction. Args: - project_name (str): Project name. - list_id (str): Entity list id. + project_name (str): On which project should be operations + processed. + operations (list[dict[str, Any]]): Operations to be processed. + can_fail (Optional[bool]): Server will try to process all + operations even if one of them fails. + raise_on_fail (Optional[bool]): Raise exception if an operation + fails. You can handle failed operations on your own + when set to 'False'. + + Raises: + ValueError: Operations can't be converted to json string. + FailedOperations: When output does not contain server operations + or 'raise_on_fail' is enabled and any operation fails. Returns: - Optional[Dict[str, Any]]: Entity list data or None if not found. + list[dict[str, Any]]: Operations result with process details. """ con = get_server_api_connection() - return con.get_entity_list_rest( + return con.send_activities_batch_operations( project_name=project_name, - list_id=list_id, + operations=operations, + can_fail=can_fail, + raise_on_fail=raise_on_fail, ) -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. +def get_actions( + project_name: Optional[str] = None, + entity_type: Optional["ActionEntityTypes"] = None, + entity_ids: Optional[List[str]] = None, + entity_subtypes: Optional[List[str]] = None, + form_data: Optional[Dict[str, Any]] = None, + *, + variant: Optional[str] = None, + mode: Optional["ActionModeType"] = None, +) -> List["ActionManifestDict"]: + """Get actions for a context. Args: - project_name (str): Project name. - list_id (str): Entity list id. - fields (Optional[Iterable[str]]): Fields to fetch from server. + project_name (Optional[str]): Name of the project. None for global + actions. + entity_type (Optional[ActionEntityTypes]): Entity type where the + action is triggered. None for global actions. + entity_ids (Optional[List[str]]): List of entity ids where the + action is triggered. None for global actions. + entity_subtypes (Optional[List[str]]): List of entity subtypes + folder types for folder ids, task types for tasks ids. + form_data (Optional[Dict[str, Any]]): Form data of the action. + variant (Optional[str]): Settings variant. + mode (Optional[ActionModeType]): Action modes. Returns: - Optional[Dict[str, Any]]: Entity list data or None if not found. + List[ActionManifestDict]: List of action manifests. """ con = get_server_api_connection() - return con.get_entity_list_by_id( + return con.get_actions( project_name=project_name, - list_id=list_id, - fields=fields, + entity_type=entity_type, + entity_ids=entity_ids, + entity_subtypes=entity_subtypes, + form_data=form_data, + variant=variant, + mode=mode, ) -def create_entity_list( - project_name: str, - entity_type: "EntityListEntityType", - label: str, +def trigger_action( + identifier: str, + addon_name: str, + addon_version: str, + project_name: Optional[str] = None, + entity_type: Optional["ActionEntityTypes"] = None, + entity_ids: Optional[List[str]] = None, + entity_subtypes: Optional[List[str]] = None, + form_data: Optional[Dict[str, Any]] = None, *, - 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. + variant: Optional[str] = None, +) -> "ActionTriggerResponse": + """Trigger action. 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. + identifier (str): Identifier of the action. + addon_name (str): Name of the addon. + addon_version (str): Version of the addon. + project_name (Optional[str]): Name of the project. None for global + actions. + entity_type (Optional[ActionEntityTypes]): Entity type where the + action is triggered. None for global actions. + entity_ids (Optional[List[str]]): List of entity ids where the + action is triggered. None for global actions. + entity_subtypes (Optional[List[str]]): List of entity subtypes + folder types for folder ids, task types for tasks ids. + form_data (Optional[Dict[str, Any]]): Form data of the action. + variant (Optional[str]): Settings variant. """ con = get_server_api_connection() - return con.create_entity_list( + return con.trigger_action( + identifier=identifier, + addon_name=addon_name, + addon_version=addon_version, 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, + entity_ids=entity_ids, + entity_subtypes=entity_subtypes, + form_data=form_data, + variant=variant, ) -def update_entity_list( - project_name: str, - list_id: str, +def get_action_config( + identifier: str, + addon_name: str, + addon_version: str, + project_name: Optional[str] = None, + entity_type: Optional["ActionEntityTypes"] = None, + entity_ids: Optional[List[str]] = None, + entity_subtypes: Optional[List[str]] = None, + form_data: Optional[Dict[str, Any]] = None, *, - 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. + variant: Optional[str] = None, +) -> "ActionConfigResponse": + """Get action configuration. 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. + identifier (str): Identifier of the action. + addon_name (str): Name of the addon. + addon_version (str): Version of the addon. + project_name (Optional[str]): Name of the project. None for global + actions. + entity_type (Optional[ActionEntityTypes]): Entity type where the + action is triggered. None for global actions. + entity_ids (Optional[List[str]]): List of entity ids where the + action is triggered. None for global actions. + entity_subtypes (Optional[List[str]]): List of entity subtypes + folder types for folder ids, task types for tasks ids. + form_data (Optional[Dict[str, Any]]): Form data of the action. + variant (Optional[str]): Settings variant. + + Returns: + ActionConfigResponse: Action configuration data. """ con = get_server_api_connection() - return con.update_entity_list( + return con.get_action_config( + identifier=identifier, + addon_name=addon_name, + addon_version=addon_version, project_name=project_name, - list_id=list_id, - label=label, - access=access, - attrib=attrib, - data=data, - tags=tags, - owner=owner, - active=active, + entity_type=entity_type, + entity_ids=entity_ids, + entity_subtypes=entity_subtypes, + form_data=form_data, + variant=variant, ) -def delete_entity_list( - project_name: str, - list_id: str, -) -> None: - """Delete entity list from project. +def set_action_config( + identifier: str, + addon_name: str, + addon_version: str, + value: Dict[str, Any], + project_name: Optional[str] = None, + entity_type: Optional["ActionEntityTypes"] = None, + entity_ids: Optional[List[str]] = None, + entity_subtypes: Optional[List[str]] = None, + form_data: Optional[Dict[str, Any]] = None, + *, + variant: Optional[str] = None, +) -> "ActionConfigResponse": + """Set action configuration. Args: - project_name (str): Project name. - list_id (str): Entity list id that will be removed. + identifier (str): Identifier of the action. + addon_name (str): Name of the addon. + addon_version (str): Version of the addon. + value (Optional[Dict[str, Any]]): Value of the action + configuration. + project_name (Optional[str]): Name of the project. None for global + actions. + entity_type (Optional[ActionEntityTypes]): Entity type where the + action is triggered. None for global actions. + entity_ids (Optional[List[str]]): List of entity ids where the + action is triggered. None for global actions. + entity_subtypes (Optional[List[str]]): List of entity subtypes + folder types for folder ids, task types for tasks ids. + form_data (Optional[Dict[str, Any]]): Form data of the action. + variant (Optional[str]): Settings variant. + + Returns: + ActionConfigResponse: New action configuration data. """ con = get_server_api_connection() - return con.delete_entity_list( + return con.set_action_config( + identifier=identifier, + addon_name=addon_name, + addon_version=addon_version, + value=value, project_name=project_name, - list_id=list_id, + entity_type=entity_type, + entity_ids=entity_ids, + entity_subtypes=entity_subtypes, + form_data=form_data, + variant=variant, ) -def get_entity_list_attribute_definitions( - project_name: str, - list_id: str, -) -> List["EntityListAttributeDefinitionDict"]: - """Get attribute definitions on entity list. +def take_action( + action_token: str, +) -> "ActionTakeResponse": + """Take action metadata using an action token. Args: - project_name (str): Project name. - list_id (str): Entity list id. + action_token (str): AYON launcher action token. Returns: - List[EntityListAttributeDefinitionDict]: List of attribute - definitions. + ActionTakeResponse: Action metadata describing how to launch + action. """ con = get_server_api_connection() - return con.get_entity_list_attribute_definitions( - project_name=project_name, - list_id=list_id, + return con.take_action( + action_token=action_token, ) -def set_entity_list_attribute_definitions( - project_name: str, - list_id: str, - attribute_definitions: List["EntityListAttributeDefinitionDict"], +def abort_action( + action_token: str, + message: Optional[str] = None, ) -> None: - """Set attribute definitions on entity list. + """Abort action using an action token. Args: - project_name (str): Project name. - list_id (str): Entity list id. - attribute_definitions (List[EntityListAttributeDefinitionDict]): - List of attribute definitions. + action_token (str): AYON launcher action token. + message (Optional[str]): Message to display in the UI. """ con = get_server_api_connection() - return con.set_entity_list_attribute_definitions( - project_name=project_name, - list_id=list_id, - attribute_definitions=attribute_definitions, + return con.abort_action( + action_token=action_token, + message=message, ) -def create_entity_list_item( +def get_entity_lists( 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. + 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 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. + 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: - str: Item id. + Generator[Dict[str, Any], None, None]: Entity list entities + matching defined filters. """ con = get_server_api_connection() - return con.create_entity_list_item( + return con.get_entity_lists( project_name=project_name, - list_id=list_id, - position=position, - label=label, - attrib=attrib, - data=data, - tags=tags, - item_id=item_id, + list_ids=list_ids, + active=active, + fields=fields, ) -def update_entity_list_items( +def get_entity_list_rest( project_name: str, list_id: str, - items: List[Dict[str, Any]], - mode: "EntityListItemMode", -) -> None: - """Update items in entity list. +) -> Optional[Dict[str, Any]]: + """Get entity list by id using REST API. Args: - project_name (str): Project name where entity list live. + project_name (str): Project name. list_id (str): Entity list id. - items (List[Dict[str, Any]]): Entity list items. - mode (EntityListItemMode): Mode of items update. + + Returns: + Optional[Dict[str, Any]]: Entity list data or None if not found. """ con = get_server_api_connection() - return con.update_entity_list_items( + return con.get_entity_list_rest( project_name=project_name, list_id=list_id, - items=items, - mode=mode, ) -def update_entity_list_item( +def get_entity_list_by_id( 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. + fields: Optional[Iterable[str]] = None, +) -> Optional[Dict[str, Any]]: + """Get entity list by id using GraphQl. 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. + 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.update_entity_list_item( + return con.get_entity_list_by_id( 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, + fields=fields, ) -def delete_entity_list_item( +def create_entity_list( project_name: str, - list_id: str, - item_id: str, -) -> None: - """Delete item from entity list. + 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 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. + 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.delete_entity_list_item( + 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, - item_id=item_id, ) -def send_batch_operations( +def update_entity_list( project_name: str, - operations: List[Dict[str, Any]], - can_fail: bool = False, - raise_on_fail: bool = True, -) -> List[Dict[str, Any]]: - """Post multiple CRUD operations to server. - - When multiple changes should be made on server side this is the best - way to go. It is possible to pass multiple operations to process on a - server side and do the changes in a transaction. + 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): On which project should be operations - processed. - operations (list[dict[str, Any]]): Operations to be processed. - can_fail (Optional[bool]): Server will try to process all - operations even if one of them fails. - raise_on_fail (Optional[bool]): Raise exception if an operation - fails. You can handle failed operations on your own - when set to 'False'. - - Raises: - ValueError: Operations can't be converted to json string. - FailedOperations: When output does not contain server operations - or 'raise_on_fail' is enabled and any operation fails. - - Returns: - list[dict[str, Any]]: Operations result with process details. + 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.send_batch_operations( + return con.update_entity_list( project_name=project_name, - operations=operations, - can_fail=can_fail, - raise_on_fail=raise_on_fail, + list_id=list_id, + label=label, + access=access, + attrib=attrib, + data=data, + tags=tags, + owner=owner, + active=active, ) -def send_activities_batch_operations( +def delete_entity_list( project_name: str, - operations: List[Dict[str, Any]], - can_fail: bool = False, - raise_on_fail: bool = True, -) -> List[Dict[str, Any]]: - """Post multiple CRUD activities operations to server. - - When multiple changes should be made on server side this is the best - way to go. It is possible to pass multiple operations to process on a - server side and do the changes in a transaction. + list_id: str, +) -> None: + """Delete entity list from project. Args: - project_name (str): On which project should be operations - processed. - operations (list[dict[str, Any]]): Operations to be processed. - can_fail (Optional[bool]): Server will try to process all - operations even if one of them fails. - raise_on_fail (Optional[bool]): Raise exception if an operation - fails. You can handle failed operations on your own - when set to 'False'. - - Raises: - ValueError: Operations can't be converted to json string. - FailedOperations: When output does not contain server operations - or 'raise_on_fail' is enabled and any operation fails. - - Returns: - list[dict[str, Any]]: Operations result with process details. + project_name (str): Project name. + list_id (str): Entity list id that will be removed. """ con = get_server_api_connection() - return con.send_activities_batch_operations( + return con.delete_entity_list( project_name=project_name, - operations=operations, - can_fail=can_fail, - raise_on_fail=raise_on_fail, + list_id=list_id, ) - -def get_actions( - project_name: Optional[str] = None, - entity_type: Optional["ActionEntityTypes"] = None, - entity_ids: Optional[List[str]] = None, - entity_subtypes: Optional[List[str]] = None, - form_data: Optional[Dict[str, Any]] = None, - *, - variant: Optional[str] = None, - mode: Optional["ActionModeType"] = None, -) -> List["ActionManifestDict"]: - """Get actions for a context. - - Args: - project_name (Optional[str]): Name of the project. None for global - actions. - entity_type (Optional[ActionEntityTypes]): Entity type where the - action is triggered. None for global actions. - entity_ids (Optional[List[str]]): List of entity ids where the - action is triggered. None for global actions. - entity_subtypes (Optional[List[str]]): List of entity subtypes - folder types for folder ids, task types for tasks ids. - form_data (Optional[Dict[str, Any]]): Form data of the action. - variant (Optional[str]): Settings variant. - mode (Optional[ActionModeType]): Action modes. - + +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[ActionManifestDict]: List of action manifests. + List[EntityListAttributeDefinitionDict]: List of attribute + definitions. """ con = get_server_api_connection() - return con.get_actions( + return con.get_entity_list_attribute_definitions( project_name=project_name, - entity_type=entity_type, - entity_ids=entity_ids, - entity_subtypes=entity_subtypes, - form_data=form_data, - variant=variant, - mode=mode, + list_id=list_id, ) -def trigger_action( - identifier: str, - addon_name: str, - addon_version: str, - project_name: Optional[str] = None, - entity_type: Optional["ActionEntityTypes"] = None, - entity_ids: Optional[List[str]] = None, - entity_subtypes: Optional[List[str]] = None, - form_data: Optional[Dict[str, Any]] = None, - *, - variant: Optional[str] = None, -) -> "ActionTriggerResponse": - """Trigger action. +def set_entity_list_attribute_definitions( + project_name: str, + list_id: str, + attribute_definitions: List["EntityListAttributeDefinitionDict"], +) -> None: + """Set attribute definitioins on entity list. Args: - identifier (str): Identifier of the action. - addon_name (str): Name of the addon. - addon_version (str): Version of the addon. - project_name (Optional[str]): Name of the project. None for global - actions. - entity_type (Optional[ActionEntityTypes]): Entity type where the - action is triggered. None for global actions. - entity_ids (Optional[List[str]]): List of entity ids where the - action is triggered. None for global actions. - entity_subtypes (Optional[List[str]]): List of entity subtypes - folder types for folder ids, task types for tasks ids. - form_data (Optional[Dict[str, Any]]): Form data of the action. - variant (Optional[str]): Settings variant. + 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.trigger_action( - identifier=identifier, - addon_name=addon_name, - addon_version=addon_version, + return con.set_entity_list_attribute_definitions( project_name=project_name, - entity_type=entity_type, - entity_ids=entity_ids, - entity_subtypes=entity_subtypes, - form_data=form_data, - variant=variant, + list_id=list_id, + attribute_definitions=attribute_definitions, ) -def get_action_config( - identifier: str, - addon_name: str, - addon_version: str, - project_name: Optional[str] = None, - entity_type: Optional["ActionEntityTypes"] = None, - entity_ids: Optional[List[str]] = None, - entity_subtypes: Optional[List[str]] = None, - form_data: Optional[Dict[str, Any]] = None, +def create_entity_list_item( + project_name: str, + list_id: str, *, - variant: Optional[str] = None, -) -> "ActionConfigResponse": - """Get action configuration. + 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: - identifier (str): Identifier of the action. - addon_name (str): Name of the addon. - addon_version (str): Version of the addon. - project_name (Optional[str]): Name of the project. None for global - actions. - entity_type (Optional[ActionEntityTypes]): Entity type where the - action is triggered. None for global actions. - entity_ids (Optional[List[str]]): List of entity ids where the - action is triggered. None for global actions. - entity_subtypes (Optional[List[str]]): List of entity subtypes - folder types for folder ids, task types for tasks ids. - form_data (Optional[Dict[str, Any]]): Form data of the action. - variant (Optional[str]): Settings variant. + 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: - ActionConfigResponse: Action configuration data. + str: Item id. """ con = get_server_api_connection() - return con.get_action_config( - identifier=identifier, - addon_name=addon_name, - addon_version=addon_version, + return con.create_entity_list_item( project_name=project_name, - entity_type=entity_type, - entity_ids=entity_ids, - entity_subtypes=entity_subtypes, - form_data=form_data, - variant=variant, + list_id=list_id, + position=position, + label=label, + attrib=attrib, + data=data, + tags=tags, + item_id=item_id, ) -def set_action_config( - identifier: str, - addon_name: str, - addon_version: str, - value: Dict[str, Any], - project_name: Optional[str] = None, - entity_type: Optional["ActionEntityTypes"] = None, - entity_ids: Optional[List[str]] = None, - entity_subtypes: Optional[List[str]] = None, - form_data: Optional[Dict[str, Any]] = None, - *, - variant: Optional[str] = None, -) -> "ActionConfigResponse": - """Set action configuration. +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: - identifier (str): Identifier of the action. - addon_name (str): Name of the addon. - addon_version (str): Version of the addon. - value (Optional[Dict[str, Any]]): Value of the action - configuration. - project_name (Optional[str]): Name of the project. None for global - actions. - entity_type (Optional[ActionEntityTypes]): Entity type where the - action is triggered. None for global actions. - entity_ids (Optional[List[str]]): List of entity ids where the - action is triggered. None for global actions. - entity_subtypes (Optional[List[str]]): List of entity subtypes - folder types for folder ids, task types for tasks ids. - form_data (Optional[Dict[str, Any]]): Form data of the action. - variant (Optional[str]): Settings variant. - - Returns: - ActionConfigResponse: New action configuration data. + 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.set_action_config( - identifier=identifier, - addon_name=addon_name, - addon_version=addon_version, - value=value, + return con.update_entity_list_items( project_name=project_name, - entity_type=entity_type, - entity_ids=entity_ids, - entity_subtypes=entity_subtypes, - form_data=form_data, - variant=variant, + list_id=list_id, + items=items, + mode=mode, ) -def take_action( - action_token: str, -) -> "ActionTakeResponse": - """Take action metadata using an action token. +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: - action_token (str): AYON launcher action token. - - Returns: - ActionTakeResponse: Action metadata describing how to launch - action. + 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.take_action( - action_token=action_token, + 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 abort_action( - action_token: str, - message: Optional[str] = None, +def delete_entity_list_item( + project_name: str, + list_id: str, + item_id: str, ) -> None: - """Abort action using an action token. + """Delete item from entity list. Args: - action_token (str): AYON launcher action token. - message (Optional[str]): Message to display in the UI. + 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.abort_action( - action_token=action_token, - message=message, + return con.delete_entity_list_item( + project_name=project_name, + list_id=list_id, + item_id=item_id, )