Frame13/ayon-core
1
0
Fork 1
mirror of https://github.com/ynput/ayon-core.git synced 2025-12-24 21:04:40 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

added method to get task and folder ids for assignees

Browse source
This commit is contained in:
Jakub Trllo 2025-04-01 12:09:42 +02:00
parent d91e6d07e2
commit 37bf4a28c7
1 changed files with 53 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

53
client/ayon_core/tools/common_models/hierarchy.py
View file

@ -227,6 +227,9 @@ class HierarchyModel(object):
self._tasks_by_id = NestedCacheItem(
levels=2, default_factory=dict, lifetime=self.lifetime)
self._entity_ids_by_assignee = NestedCacheItem(
levels=2, default_factory=dict, lifetime=self.lifetime)
self._folders_refreshing = set()
self._tasks_refreshing = set()
self._controller = controller
@ -238,6 +241,8 @@ class HierarchyModel(object):
self._task_items.reset()
self._tasks_by_id.reset()
self._entity_ids_by_assignee.reset()
def refresh_project(self, project_name):
"""Force to refresh folder items for a project.
@ -461,6 +466,54 @@ class HierarchyModel(object):
output = self.get_task_entities(project_name, {task_id})
return output[task_id]
def get_entity_ids_for_assignees(
self, project_name: str, assignees: list[str]
):
folder_ids = set()
task_ids = set()
output = {
"folder_ids": folder_ids,
"task_ids": task_ids,
}
assignees = set(assignees)
for assignee in tuple(assignees):
cache = self._entity_ids_by_assignee[project_name][assignee]
if cache.is_valid:
assignees.discard(assignee)
assignee_data = cache.get_data()
folder_ids.update(assignee_data["folder_ids"])
task_ids.update(assignee_data["task_ids"])
if not assignees:
return output
tasks = ayon_api.get_tasks(
project_name,
assignees_all=assignees,
fields={"id", "folderId", "assignees"},
)
tasks_assignee = {}
for task in tasks:
folder_ids.add(task["folderId"])
task_ids.add(task["id"])
for assignee in task["assignees"]:
tasks_assignee.setdefault(assignee, []).append(task)
for assignee, tasks in tasks_assignee.items():
cache = self._entity_ids_by_assignee[project_name][assignee]
assignee_folder_ids = set()
assignee_task_ids = set()
assignee_data = {
"folder_ids": assignee_folder_ids,
"task_ids": assignee_task_ids,
}
for task in tasks:
assignee_folder_ids.add(task["folderId"])
assignee_task_ids.add(task["id"])
cache.update_data(assignee_data)
return output
@contextlib.contextmanager
def _folder_refresh_event_manager(self, project_name, sender):
self._folders_refreshing.add(project_name)